diff --git a/tests/repository_simulator.py b/tests/repository_simulator.py index a763fc053d..339be54283 100644 --- a/tests/repository_simulator.py +++ b/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,19 @@ 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) + + def clear(self): + """Clear all counters.""" + self.metadata = list + self.targets = list + + @dataclass class RepositoryTarget: """Contains actual target data and the related target metadata.""" @@ -116,6 +129,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) @@ -239,6 +254,9 @@ def _fetch_target( raise FetcherHTTPError(f"hash mismatch for {target_path}", 404) logger.debug("fetched target %s", target_path) + + self.fetch_tracker.targets.append(target_path) + return repo_target.data def _fetch_metadata( @@ -248,11 +266,16 @@ 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): raise FetcherHTTPError(f"Unknown root version {version}", 404) logger.debug("fetched root version %d", version) + + # self.fetch_tracker.metadata.append(role) + return self.signed_roots[version - 1] # sign and serialize the requested metadata @@ -279,6 +302,9 @@ def _fetch_metadata( md.signed.version, len(self.signers[role]), ) + + # self.fetch_tracker.metadata.append(role) + return md.to_bytes(JSONSerializer()) def _compute_hashes_and_length( diff --git a/tests/test_updater_consistent_snapshot.py b/tests/test_updater_consistent_snapshot.py index 4289d7b860..a5036ebd01 100644 --- a/tests/test_updater_consistent_snapshot.py +++ b/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), ], }, } @@ -116,17 +116,15 @@ def test_top_level_roles_update(self, test_case_data: Dict[str, Any]): sim = self._init_repo(consistent_snapshot) updater = self._init_updater(sim) + sim.fetch_tracker.metadata.clear() + updater.refresh() - with patch.object( - sim, "_fetch_metadata", wraps=sim._fetch_metadata - ) as wrapped_fetch: - 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) + sim.fetch_tracker.clear() self._cleanup_dir(self.metadata_dir) delegated_roles_data: utils.DataSet = { @@ -147,7 +145,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,18 +155,19 @@ 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 the fetched metadata from repository simulator + 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) + sim.fetch_tracker.clear() self._cleanup_dir(self.metadata_dir) targets_download_data: utils.DataSet = { @@ -204,30 +203,20 @@ def test_download_targets(self, test_case_data: Dict[str, Any]): 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() + for targetpath in targetpaths: + info = updater.get_targetinfo(targetpath) + updater.download_target(info) + + self.assertListEqual(sim.fetch_tracker.targets, targetpaths) + sim.fetch_tracker.clear() self._cleanup_dir(self.targets_dir)