Implemented call_stats to RepositorySimulator

This change is to implement a feature to track the fetch calls to
the metadata and targets. This feature was mentioned in PR theupdateframework#1666 that
generated issue theupdateframework#1682.
It is the ``RepositorySimulator.fetch_tracker``. It also changes the
``test_updater_consistent_snapshot`` to use the fetch_tracker
instead of mocking it.

It implements a ``dataclass`` that stores the calls to fetch metadata
(``_fetch_metadata``) and targets (``_fetch_targets``).
This dataclass has a method to clean (``clear``) the current metadata
and targets tracker.

The fetch calls for metadata are stored as a list of tuples that
contains metadata role name and version, and fetch calls for targets
are stored as a list of target paths.

Signed-off-by: Kairo de Araujo <kdearaujo@vmware.com>

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(

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)