PoC: Tap 14

tests/repository_simulator.py

@@ -205,6 +205,11 @@ def _fetch(self, url: str) -> Iterator[bytes]:
         if path.startswith("/metadata/") and path.endswith(".json"):
             # figure out rolename and version
             ver_and_name = path[len("/metadata/") :][: -len(".json")]
+            # inside a version folder
+            if "/" in ver_and_name:
+                ver_and_name_split = ver_and_name.partition("/")
+                assert len(ver_and_name_split) == 3
+                ver_and_name = ver_and_name_split[2]
             version_str, _, role = ver_and_name.partition(".")
             # root is always version-prefixed while timestamp is always NOT
             if role == Root.type or (

tests/test_updater_delegation_graphs.py

@@ -134,7 +134,7 @@ def _assert_files_exist(self, roles: Iterable[str]) -> None:
         """Assert that local metadata files exist for 'roles'"""
         expected_files = sorted([f"{role}.json" for role in roles])
         local_metadata_files = sorted(os.listdir(self.metadata_dir))
-        self.assertListEqual(local_metadata_files, expected_files)
+        self.assertEqual(expected_files, local_metadata_files)
 
 
 class TestDelegationsGraphs(TestDelegations):
@@ -263,7 +263,11 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None:
         in the delegator's metadata, using pre-order depth-first search"""
 
         try:
-            exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order]
+            exp_files = [
+                *TOP_LEVEL_ROLE_NAMES,
+                *test_data.visited_order,
+                "spec_version",
+            ]
             exp_calls = [(role, 1) for role in test_data.visited_order]
 
             self._init_repo(test_data)
@@ -276,7 +280,7 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None:
             updater.refresh()
             self.sim.fetch_tracker.metadata.clear()
             # Check that metadata dir contains only top-level roles
-            self._assert_files_exist(TOP_LEVEL_ROLE_NAMES)
+            self._assert_files_exist([*TOP_LEVEL_ROLE_NAMES, "spec_version"])
 
             # Looking for a non-existing targetpath forces updater
             # to visit all possible delegated roles
@@ -311,7 +315,11 @@ def test_invalid_metadata(self, test_data: DelegationsTestCase) -> None:
 
             self.setup_subtest()
             # The invalid role metadata must not be persisted
-            exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order[:-1]]
+            exp_files = [
+                *TOP_LEVEL_ROLE_NAMES,
+                *test_data.visited_order[:-1],
+                "spec_version",
+            ]
             exp_calls = [(role, 1) for role in test_data.visited_order]
 
             updater = self._init_updater()
@@ -397,7 +405,11 @@ def test_hash_bins_graph_traversal(
         they correctly reffer to the corresponding hash bin prefixes"""
 
         try:
-            exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order]
+            exp_files = [
+                *TOP_LEVEL_ROLE_NAMES,
+                *test_data.visited_order,
+                "spec_version",
+            ]
             exp_calls = [(role, 1) for role in test_data.visited_order]
 
             self._init_repo(test_data)
@@ -408,7 +420,7 @@ def test_hash_bins_graph_traversal(
             updater.refresh()
             self.sim.fetch_tracker.metadata.clear()
             # Check that metadata dir contains only top-level roles
-            self._assert_files_exist(TOP_LEVEL_ROLE_NAMES)
+            self._assert_files_exist([*TOP_LEVEL_ROLE_NAMES, "spec_version"])
 
             # Looking for a non-existing targetpath forces updater
             # to visit a correspondning delegated role
@@ -481,7 +493,11 @@ def test_succinct_roles_graph_traversal(
         # bin should exist locally and only one bin must be downloaded.
 
         try:
-            exp_files = [*TOP_LEVEL_ROLE_NAMES, test_data.expected_target_bin]
+            exp_files = [
+                *TOP_LEVEL_ROLE_NAMES,
+                test_data.expected_target_bin,
+                "spec_version",
+            ]
             exp_calls = [(test_data.expected_target_bin, 1)]
 
             self.sim = RepositorySimulator()
@@ -495,7 +511,7 @@ def test_succinct_roles_graph_traversal(
             updater.refresh()
             self.sim.fetch_tracker.metadata.clear()
             # Check that metadata dir contains only top-level roles
-            self._assert_files_exist(TOP_LEVEL_ROLE_NAMES)
+            self._assert_files_exist([*TOP_LEVEL_ROLE_NAMES, "spec_version"])
 
             # Looking for a non-existing targetpath forces updater
             # to visit a corresponding delegated role.
@@ -564,7 +580,11 @@ def setUp(self) -> None:
     def test_targetfile_search(self, test_data: TargetTestCase) -> None:
         try:
             self.setup_subtest()
-            exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order]
+            exp_files = [
+                *TOP_LEVEL_ROLE_NAMES,
+                *test_data.visited_order,
+                "spec_version",
+            ]
             exp_calls = [(role, 1) for role in test_data.visited_order]
             exp_target = self.sim.target_files[test_data.targetpath].target_file