Skip to content

Commit

Permalink
Add --dump option to tests
Browse files Browse the repository at this point in the history 
Extend updater tests with the option to dump repository
metadata locally.

Signed-off-by: Teodora Sechkova <[email protected]>
  • Loading branch information
@sechkova
sechkova committed Dec 15, 2021
1 parent db0e0d6 commit 1ba9301
Show file tree
Hide file tree
Showing 3 changed files with 104 additions and 28 deletions.
80 changes: 55 additions & 25 deletions tests/test_updater_consistent_snapshot.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,12 @@ class TestConsistentSnapshot(unittest.TestCase):
'prefix_targets_with_hash' and verify that the correct URLs
are formed for each combination"""

# set dump_dir to trigger repository state dumps
dump_dir: Optional[str] = None

def setUp(self) -> None:
# pylint: disable=consider-using-with
self.subtest_count = 0
self.temp_dir = tempfile.TemporaryDirectory()
self.metadata_dir = os.path.join(self.temp_dir.name, "metadata")
self.targets_dir = os.path.join(self.temp_dir.name, "targets")
Expand All @@ -39,6 +43,24 @@ def setUp(self) -> None:
def tearDown(self) -> None:
self.temp_dir.cleanup()

def setup_subtest(
self, consistent_snapshot: bool, prefix_targets: bool = True
) -> None:
self.sim = self._init_repo(consistent_snapshot, prefix_targets)

self.subtest_count += 1
if self.dump_dir is not None:
# create subtest dumpdir
name = f"{self.id().split('.')[-1]}-{self.subtest_count}"
self.sim.dump_dir = os.path.join(self.dump_dir, name)
os.mkdir(self.sim.dump_dir)

def teardown_subtest(self):
if self.dump_dir is not None:
self.sim.write()

utils.cleanup_dir(self.metadata_dir)

def _init_repo(
self, consistent_snapshot: bool, prefix_targets: bool = True
) -> RepositorySimulator:
Expand All @@ -55,14 +77,14 @@ def _init_repo(

return sim

def _init_updater(self, sim: RepositorySimulator) -> Updater:
def _init_updater(self) -> Updater:
"""Create a new Updater instance"""
return Updater(
self.metadata_dir,
"https://example.com/metadata/",
self.targets_dir,
"https://example.com/targets/",
sim,
self.sim,
)

def _assert_metadata_files_exist(self, roles: Iterable[str]) -> None:
Expand Down Expand Up @@ -106,21 +128,21 @@ def test_top_level_roles_update(
# correct version prefix, depending on 'consistent_snapshot' config
try:
consistent_snapshot: bool = test_case_data["consistent_snapshot"]
expected_calls: List[Any] = test_case_data["calls"]
exp_calls: List[Any] = test_case_data["calls"]

sim = self._init_repo(consistent_snapshot)
updater = self._init_updater(sim)
self.setup_subtest(consistent_snapshot)
updater = self._init_updater()

# cleanup fetch tracker metadata
sim.fetch_tracker.metadata.clear()
self.sim.fetch_tracker.metadata.clear()
updater.refresh()

# metadata files are fetched with the expected version (or None)
self.assertListEqual(sim.fetch_tracker.metadata, expected_calls)
self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls)
# metadata files are always persisted without a version prefix
self._assert_metadata_files_exist(TOP_LEVEL_ROLE_NAMES)
finally:
utils.cleanup_dir(self.metadata_dir)
self.teardown_subtest()

delegated_roles_data: utils.DataSet = {
"consistent_snaphot disabled": {
Expand All @@ -145,27 +167,29 @@ def test_delegated_roles_update(
rolenames = ["role1", "..", "."]
exp_calls = [(role, exp_version) for role in rolenames]

sim = self._init_repo(consistent_snapshot)
self.setup_subtest(consistent_snapshot)
# Add new delegated targets
spec_version = ".".join(SPECIFICATION_VERSION)
for role in rolenames:
delegated_role = DelegatedRole(role, [], 1, False, ["*"], None)
targets = Targets(1, spec_version, sim.safe_expiry, {}, None)
sim.add_delegation("targets", delegated_role, targets)
sim.update_snapshot()
updater = self._init_updater(sim)
targets = Targets(
1, spec_version, self.sim.safe_expiry, {}, None
)
self.sim.add_delegation("targets", delegated_role, targets)
self.sim.update_snapshot()
updater = self._init_updater()
updater.refresh()

# cleanup fetch tracker metadata
sim.fetch_tracker.metadata.clear()
self.sim.fetch_tracker.metadata.clear()
# trigger updater to fetch the delegated metadata
updater.get_targetinfo("anything")
# metadata files are fetched with the expected version (or None)
self.assertListEqual(sim.fetch_tracker.metadata, exp_calls)
self.assertListEqual(self.sim.fetch_tracker.metadata, exp_calls)
# metadata files are always persisted without a version prefix
self._assert_metadata_files_exist(rolenames)
finally:
utils.cleanup_dir(self.metadata_dir)
self.teardown_subtest()

targets_download_data: utils.DataSet = {
"consistent_snaphot disabled": {
Expand Down Expand Up @@ -199,14 +223,14 @@ def test_download_targets(self, test_case_data: Dict[str, Any]) -> None:
hash_algo: Optional[str] = test_case_data["hash_algo"]
targetpaths: List[str] = test_case_data["targetpaths"]

sim = self._init_repo(consistent_snapshot, prefix_targets_with_hash)
self.setup_subtest(consistent_snapshot, prefix_targets_with_hash)
# Add targets to repository
for targetpath in targetpaths:
sim.targets.version += 1
sim.add_target("targets", b"content", targetpath)
sim.update_snapshot()
self.sim.targets.version += 1
self.sim.add_target("targets", b"content", targetpath)
self.sim.update_snapshot()

updater = self._init_updater(sim)
updater = self._init_updater()
updater.config.prefix_targets_with_hash = prefix_targets_with_hash
updater.refresh()

Expand All @@ -219,17 +243,23 @@ def test_download_targets(self, test_case_data: Dict[str, Any]) -> None:
self._assert_targets_files_exist([info.path])

# files are fetched with the expected hash prefix (or None)
exp_fetches = [
exp_calls = [
(path, None if not hash_algo else info.hashes[hash_algo])
]

self.assertListEqual(sim.fetch_tracker.targets, exp_fetches)
sim.fetch_tracker.targets.clear()
self.assertListEqual(self.sim.fetch_tracker.targets, exp_calls)
self.sim.fetch_tracker.targets.clear()
finally:
utils.cleanup_dir(self.targets_dir)
self.teardown_subtest()


if __name__ == "__main__":
if "--dump" in sys.argv:
TestConsistentSnapshot.dump_dir = tempfile.mkdtemp()
print(
f"Repository Simulator dumps in {TestConsistentSnapshot.dump_dir}"
)
sys.argv.remove("--dump")

utils.configure_test_logging(sys.argv)
unittest.main()
33 changes: 30 additions & 3 deletions tests/test_updater_delegation_graphs.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,11 @@ class TestDelegationsGraphs(unittest.TestCase):
"""Test creating delegations graphs with different complexity
and successfully updating the delegated roles metadata"""

# set dump_dir to trigger repository state dumps
dump_dir: Optional[str] = None

def setUp(self) -> None:
self.subtest_count = 0
self.temp_dir = tempfile.TemporaryDirectory()
self.metadata_dir = os.path.join(self.temp_dir.name, "metadata")
self.targets_dir = os.path.join(self.temp_dir.name, "targets")
Expand All @@ -59,6 +63,26 @@ def setUp(self) -> None:
def tearDown(self) -> None:
self.temp_dir.cleanup()

def setup_subtest(
self, delegations: List[TestDelegation]
) -> RepositorySimulator:
sim = self._init_repo(delegations)

self.subtest_count += 1
if self.dump_dir is not None:
# create subtest dumpdir
name = f"{self.id().split('.')[-1]}-{self.subtest_count}"
sim.dump_dir = os.path.join(self.dump_dir, name)
os.mkdir(sim.dump_dir)
# dump the repo simulator metadata
sim.write()

return sim

def teardown_subtest(self) -> None:
# clean up after each subtest
utils.cleanup_dir(self.metadata_dir)

def _init_updater(self, sim: RepositorySimulator) -> Updater:
"""Create a new Updater instance"""
return Updater(
Expand Down Expand Up @@ -213,7 +237,7 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None:
exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order]
exp_calls = [call(role, 1) for role in test_data.visited_order]

sim = self._init_repo(test_data.delegations)
sim = self.setup_subtest(test_data.delegations)
updater = self._init_updater(sim)
# Call explicitly refresh to simplify the expected_calls list
updater.refresh()
Expand All @@ -233,11 +257,14 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None:
self.assertListEqual(wrapped_fetch.call_args_list, exp_calls)
self._assert_files_exist(exp_files)
finally:
# clean up after each subtest
utils.cleanup_dir(self.metadata_dir)
self.teardown_subtest()


if __name__ == "__main__":
if "--dump" in sys.argv:
TestDelegationsGraphs.dump_dir = tempfile.mkdtemp()
print(f"Repository Simulator dumps in {TestDelegationsGraphs.dump_dir}")
sys.argv.remove("--dump")

utils.configure_test_logging(sys.argv)
unittest.main()
19 changes: 19 additions & 0 deletions tests/test_updater_top_level_update.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,9 @@ class TestRefresh(unittest.TestCase):
"""Test update of top-level metadata following
'Detailed client workflow' in the specification."""

# set dump_dir to trigger repository state dumps
dump_dir: Optional[str] = None

past_datetime = datetime.utcnow().replace(microsecond=0) - timedelta(days=5)

def setUp(self) -> None:
Expand All @@ -53,11 +56,20 @@ def setUp(self) -> None:
with open(os.path.join(self.metadata_dir, "root.json"), "bw") as f:
f.write(self.sim.signed_roots[0])

if self.dump_dir is not None:
# create test specific dump directory
name = self.id().split(".")[-1]
self.sim.dump_dir = os.path.join(self.dump_dir, name)
os.mkdir(self.sim.dump_dir)

def tearDown(self) -> None:
self.temp_dir.cleanup()

def _run_refresh(self) -> Updater:
"""Create a new Updater instance and refresh"""
if self.dump_dir is not None:
self.sim.write()

updater = Updater(
self.metadata_dir,
"https://example.com/metadata/",
Expand All @@ -70,6 +82,9 @@ def _run_refresh(self) -> Updater:

def _init_updater(self) -> Updater:
"""Create a new Updater instance"""
if self.dump_dir is not None:
self.sim.write()

return Updater(
self.metadata_dir,
"https://example.com/metadata/",
Expand Down Expand Up @@ -455,6 +470,10 @@ def test_compute_metafile_hashes_length(self) -> None:


if __name__ == "__main__":
if "--dump" in sys.argv:
TestRefresh.dump_dir = tempfile.mkdtemp()
print(f"Repository Simulator dumps in {TestRefresh.dump_dir}")
sys.argv.remove("--dump")

utils.configure_test_logging(sys.argv)
unittest.main()

0 comments on commit 1ba9301

Please sign in to comment.