Issue#1682/repository simulator fetch tracker

tests/repository_simulator.py

@@ -48,7 +48,7 @@
 import os
 import tempfile
 from collections import OrderedDict
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 from typing import Dict, Iterator, List, Optional, Tuple
 from urllib import parse
@@ -81,6 +81,14 @@
 SPEC_VER = ".".join(SPECIFICATION_VERSION)
 
 
+@dataclass
+class FetchCounter:
+    """Fetcher counter for metadata and targets."""
+
+    metadata: list = field(default_factory=list)
+    targets: list = field(default_factory=list)
+
+
 @dataclass
 class RepositoryTarget:
     """Contains actual target data and the related target metadata."""
@@ -116,6 +124,8 @@ def __init__(self) -> None:
         self.dump_dir: Optional[str] = None
         self.dump_version = 0
 
+        self.fetch_tracker: FetchCounter = FetchCounter()
+
         now = datetime.utcnow()
         self.safe_expiry = now.replace(microsecond=0) + timedelta(days=30)
 
@@ -229,6 +239,8 @@ def _fetch_target(
 
         If hash is None, then consistent_snapshot is not used.
         """
+        self.fetch_tracker.targets.append((target_path, target_hash))
+
         repo_target = self.target_files.get(target_path)
         if repo_target is None:
             raise FetcherHTTPError(f"No target {target_path}", 404)
@@ -248,6 +260,8 @@ def _fetch_metadata(
 
         If version is None, non-versioned metadata is being requested.
         """
+        self.fetch_tracker.metadata.append((role, version))
+
         if role == "root":
             # return a version previously serialized in publish_root()
             if version is None or version > len(self.signed_roots):

tests/test_updater_consistent_snapshot.py

@@ -90,19 +90,19 @@ def _assert_targets_files_exist(self, filenames: Iterable[str]) -> None:
         "consistent_snaphot disabled": {
             "consistent_snapshot": False,
             "calls": [
-                call("root", 3),
-                call("timestamp", None),
-                call("snapshot", None),
-                call("targets", None),
+                ("root", 3),
+                ("timestamp", None),
+                ("snapshot", None),
+                ("targets", None),
             ],
         },
         "consistent_snaphot enabled": {
             "consistent_snapshot": True,
             "calls": [
-                call("root", 3),
-                call("timestamp", None),
-                call("snapshot", 1),
-                call("targets", 1),
+                ("root", 3),
+                ("timestamp", None),
+                ("snapshot", 1),
+                ("targets", 1),
             ],
         },
     }
@@ -117,15 +117,14 @@ def test_top_level_roles_update(self, test_case_data: Dict[str, Any]):
         sim = self._init_repo(consistent_snapshot)
         updater = self._init_updater(sim)
 
-        with patch.object(
-            sim, "_fetch_metadata", wraps=sim._fetch_metadata
-        ) as wrapped_fetch:
-            updater.refresh()
+        # cleanup fetch tracker metadata
+        sim.fetch_tracker.metadata.clear()
+        updater.refresh()
 
-            # metadata files are fetched with the expected version (or None)
-            self.assertListEqual(wrapped_fetch.call_args_list, expected_calls)
-            # metadata files are always persisted without a version prefix
-            self._assert_metadata_files_exist(TOP_LEVEL_ROLE_NAMES)
+        # metadata files are fetched with the expected version (or None)
+        self.assertListEqual(sim.fetch_tracker.metadata, expected_calls)
+        # metadata files are always persisted without a version prefix
+        self._assert_metadata_files_exist(TOP_LEVEL_ROLE_NAMES)
 
         self._cleanup_dir(self.metadata_dir)
 
@@ -147,7 +146,7 @@ def test_delegated_roles_update(self, test_case_data: Dict[str, Any]):
         consistent_snapshot: bool = test_case_data["consistent_snapshot"]
         expected_version: Optional[int] = test_case_data["expected_version"]
         rolenames = ["role1", "..", "."]
-        expected_calls = [call(role, expected_version) for role in rolenames]
+        expected_calls = [(role, expected_version) for role in rolenames]
 
         sim = self._init_repo(consistent_snapshot)
         # Add new delegated targets
@@ -157,17 +156,17 @@ def test_delegated_roles_update(self, test_case_data: Dict[str, Any]):
             sim.add_delegation("targets", role, targets, False, ["*"], None)
         sim.update_snapshot()
         updater = self._init_updater(sim)
+
         updater.refresh()
 
-        with patch.object(
-            sim, "_fetch_metadata", wraps=sim._fetch_metadata
-        ) as wrapped_fetch:
-            # trigger updater to fetch the delegated metadata
-            updater.get_targetinfo("anything")
-            # metadata files are fetched with the expected version (or None)
-            self.assertListEqual(wrapped_fetch.call_args_list, expected_calls)
-            # metadata files are always persisted without a version prefix
-            self._assert_metadata_files_exist(rolenames)
+        # cleanup fetch tracker metadata
+        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, expected_calls)
+        # metadata files are always persisted without a version prefix
+        self._assert_metadata_files_exist(rolenames)
 
         self._cleanup_dir(self.metadata_dir)
 
@@ -176,16 +175,19 @@ def test_delegated_roles_update(self, test_case_data: Dict[str, Any]):
             "consistent_snapshot": False,
             "prefix_targets": True,
             "hash_algo": None,
+            "targetpaths": ["file", "file.txt", "..file.ext", "f.le"],
         },
         "consistent_snaphot enabled without prefixed targets": {
             "consistent_snapshot": True,
             "prefix_targets": False,
             "hash_algo": None,
+            "targetpaths": ["file", "file.txt", "..file.ext", "f.le"],
         },
         "consistent_snaphot enabled with prefixed targets": {
             "consistent_snapshot": True,
             "prefix_targets": True,
             "hash_algo": "sha256",
+            "targetpaths": ["file", "file.txt", "..file.ext", "f.le"],
         },
     }
 
@@ -197,36 +199,29 @@ def test_download_targets(self, test_case_data: Dict[str, Any]):
         consistent_snapshot: bool = test_case_data["consistent_snapshot"]
         prefix_targets_with_hash: bool = test_case_data["prefix_targets"]
         hash_algo: Optional[str] = test_case_data["hash_algo"]
-        targetpaths = ["file", "file.txt", "..file.ext", "f.le"]
+        targetpaths: list = test_case_data["targetpaths"]
 
         sim = self._init_repo(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()
 
         updater = self._init_updater(sim)
         updater.config.prefix_targets_with_hash = prefix_targets_with_hash
         updater.refresh()
 
-        with patch.object(
-            sim, "_fetch_target", wraps=sim._fetch_target
-            ) as wrapped_fetch_target:
-
-            for targetpath in targetpaths:
-                info = updater.get_targetinfo(targetpath)
-                updater.download_target(info)
-                expected_prefix = (
-                    None if not hash_algo else info.hashes[hash_algo]
-                )
-                # files are fetched with the expected hash prefix (or None)
-                wrapped_fetch_target.assert_called_once_with(
-                    info.path, expected_prefix
-                )
-                # target files are always persisted without hash prefix
-                self._assert_targets_files_exist([info.path])
-                wrapped_fetch_target.reset_mock()
+        expected_result = []
+        for targetpath in targetpaths:
+            info = updater.get_targetinfo(targetpath)
+            updater.download_target(info)
+            expected_prefix = None if not hash_algo else info.hashes[hash_algo]
+            expected_result.append((targetpath, expected_prefix))
+
+        # files are fetched with the expected hash prefix (or None)
+        self.assertListEqual(sim.fetch_tracker.targets, expected_result)
 
         self._cleanup_dir(self.targets_dir)