Add --dump option to tests

Extend updater tests with the option to dump repository
metadata locally.

Signed-off-by: Teodora Sechkova <[email protected]>

Author: sechkova

tests/test_updater_consistent_snapshot.py

@@ -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")
@@ -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:
@@ -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:
@@ -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": {
@@ -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": {
@@ -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()
 
@@ -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()

tests/test_updater_delegation_graphs.py

@@ -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")
@@ -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(
@@ -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()
@@ -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()

tests/test_updater_top_level_update.py

@@ -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:
@@ -53,7 +56,16 @@ 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:
+        if self.dump_dir is not None:
+            self.sim.write()
+
         self.temp_dir.cleanup()
 
     def _run_refresh(self) -> Updater:
@@ -455,6 +467,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()