Issue#1682/repository simulator fetch tracker

docs/CONTRIBUTORS.rst

@@ -113,8 +113,9 @@ CI/CD will check that new TUF code is formatted with `black
 Auto-formatting can be done on the command line:
 ::
 
-    $ black <filename>
-    $ isort <filename>
+    $ # TODO: configure black and isort args in pyproject.toml (see #1161)
+    $ black --line-length 80 tuf/api
+    $ isort --line-length 80 --profile black -p tuf tuf/api
 
 or via source code editor plugin
 [`black <https://black.readthedocs.io/en/stable/editor_integration.html>`__,

pyproject.toml

@@ -1,90 +1,3 @@
-# Build-system section
 [build-system]
 requires = ["setuptools>=40.8.0", "wheel"]
 build-backend = "setuptools.build_meta"
-
-# Black section
-# Read more here: https://black.readthedocs.io/en/stable/usage_and_configuration/the_basics.html#configuration-via-a-file
-[tool.black]
-line-length=80
-
-# Isort section
-# Read more here: https://pycqa.github.io/isort/docs/configuration/config_files.html
-[tool.isort]
-profile="black"
-line_length=80
-known_first_party = ["tuf"]
-
-# Pylint section
-
-# Minimal pylint configuration file for Secure Systems Lab Python Style Guide:
-#     https://github.com/secure-systems-lab/code-style-guidelines
-#
-# Based on Google Python Style Guide pylintrc and pylint defaults:
-#     https://google.github.io/styleguide/pylintrc
-#     http://pylint.pycqa.org/en/latest/technical_reference/features.html
-
-[tool.pylint.message_control]
-# Disable the message, report, category or checker with the given id(s).
-# NOTE: To keep this config as short as possible we only disable checks that
-# are currently in conflict with our code. If new code displeases the linter
-# (for good reasons) consider updating this config file, or disable checks with.
-disable=[
-  "fixme",
-  "too-few-public-methods",
-  "too-many-arguments",
-  "format",
-  "duplicate-code"
-]
-
-[tool.pylint.basic]
-good-names = ["i","j","k","v","e","f","fn","fp","_type","_"]
-# Regexes for allowed names are copied from the Google pylintrc
-# NOTE: Pylint captures regex name groups such as 'snake_case' or 'camel_case'.
-# If there are multiple groups it enfoces the prevalent naming style inside
-# each modules. Names in the exempt capturing group are ignored.
-function-rgx="^(?:(?P<exempt>setUp|tearDown|setUpModule|tearDownModule)|(?P<camel_case>_?[A-Z][a-zA-Z0-9]*)|(?P<snake_case>_?[a-z][a-z0-9_]*))$"
-method-rgx="(?x)^(?:(?P<exempt>_[a-z0-9_]+__|runTest|setUp|tearDown|setUpTestCase|tearDownTestCase|setupSelf|tearDownClass|setUpClass|(test|assert)_*[A-Z0-9][a-zA-Z0-9_]*|next)|(?P<camel_case>_{0,2}[A-Z][a-zA-Z0-9_]*)|(?P<snake_case>_{0,2}[a-z][a-z0-9_]*))$"
-argument-rgx="^[a-z][a-z0-9_]*$"
-attr-rgx="^_{0,2}[a-z][a-z0-9_]*$"
-class-attribute-rgx="^(_?[A-Z][A-Z0-9_]*|__[a-z0-9_]+__|_?[a-z][a-z0-9_]*)$"
-class-rgx="^_?[A-Z][a-zA-Z0-9]*$"
-const-rgx="^(_?[A-Z][A-Z0-9_]*|__[a-z0-9_]+__|_?[a-z][a-z0-9_]*)$"
-inlinevar-rgx="^[a-z][a-z0-9_]*$"
-module-rgx="^(_?[a-z][a-z0-9_]*|__init__)$"
-no-docstring-rgx="(__.*__|main|test.*|.*test|.*Test)$"
-variable-rgx="^[a-z][a-z0-9_]*$"
-docstring-min-length=10
-
-[tool.pylint.logging]
-logging-format-style="old"
-
-[tool.pylint.miscellaneous]
-notes="TODO"
-
-[tool.pylint.STRING]
-check-quote-consistency="yes"
-
-# mypy section
-# Read more here: https://mypy.readthedocs.io/en/stable/config_file.html#using-a-pyproject-toml-file
-[tool.mypy]
-warn_unused_configs = "True"
-warn_redundant_casts = "True"
-warn_unused_ignores = "True"
-warn_unreachable = "True"
-strict_equality = "True"
-disallow_untyped_defs = "True"
-disallow_untyped_calls = "True"
-show_error_codes = "True"
-files = [
-  "tuf/api/",
-  "tuf/ngclient",
-  "tuf/exceptions.py"
-]
-
-[[tool.mypy.overrides]]
-module = [
-  "securesystemslib.*",
-  "urllib3.*"
-]
-ignore_missing_imports = "True"

setup.cfg

@@ -52,3 +52,23 @@ tuf = py.typed
 ignore =
   .fossa.yml
   .readthedocs.yaml
+
+[mypy]
+warn_unused_configs = True
+warn_redundant_casts = True
+warn_unused_ignores = True
+warn_unreachable = True
+strict_equality = True
+disallow_untyped_defs = True
+disallow_untyped_calls = True
+show_error_codes = True
+files =
+  tuf/api/,
+  tuf/ngclient,
+  tuf/exceptions.py
+
+[mypy-securesystemslib.*]
+ignore_missing_imports = True
+
+[mypy-urllib3.*]
+ignore_missing_imports = True

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 FetchTracker:
+    """Fetcher counter for metadata and targets."""
+
+    metadata: List[Tuple[str, Optional[int]]] = field(default_factory=list)
+    targets: List[Tuple[str, Optional[str]]] = 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 = FetchTracker()
+
         now = datetime.utcnow()
         self.safe_expiry = now.replace(microsecond=0) + timedelta(days=30)
 
@@ -139,7 +149,7 @@ def targets(self) -> Targets:
 
     def all_targets(self) -> Iterator[Tuple[str, Targets]]:
         """Yield role name and signed portion of targets one by one."""
-        yield Targets.type, self.md_targets.signed
+        yield "targets", self.md_targets.signed
         for role, md in self.md_delegates.items():
             yield role, md.signed
 
@@ -181,7 +191,7 @@ def _initialize(self) -> None:
     def publish_root(self) -> None:
         """Sign and store a new serialized version of root."""
         self.md_root.signatures.clear()
-        for signer in self.signers[Root.type].values():
+        for signer in self.signers["root"].values():
             self.md_root.sign(signer, append=True)
 
         self.signed_roots.append(self.md_root.to_bytes(JSONSerializer()))
@@ -197,8 +207,8 @@ def fetch(self, url: str) -> Iterator[bytes]:
             ver_and_name = path[len("/metadata/") :][: -len(".json")]
             version_str, _, role = ver_and_name.partition(".")
             # root is always version-prefixed while timestamp is always NOT
-            if role == Root.type or (
-                self.root.consistent_snapshot and ver_and_name != Timestamp.type
+            if role == "root" or (
+                self.root.consistent_snapshot and ver_and_name != "timestamp"
             ):
                 version: Optional[int] = int(version_str)
             else:
@@ -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,7 +260,9 @@ def _fetch_metadata(
 
         If version is None, non-versioned metadata is being requested.
         """
-        if role == Root.type:
+        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)
@@ -257,11 +271,11 @@ def _fetch_metadata(
 
         # sign and serialize the requested metadata
         md: Optional[Metadata]
-        if role == Timestamp.type:
+        if role == "timestamp":
             md = self.md_timestamp
-        elif role == Snapshot.type:
+        elif role == "snapshot":
             md = self.md_snapshot
-        elif role == Targets.type:
+        elif role == "targets":
             md = self.md_targets
         else:
             md = self.md_delegates.get(role)
@@ -297,7 +311,7 @@ def update_timestamp(self) -> None:
         self.timestamp.snapshot_meta.version = self.snapshot.version
 
         if self.compute_metafile_hashes_length:
-            hashes, length = self._compute_hashes_and_length(Snapshot.type)
+            hashes, length = self._compute_hashes_and_length("snapshot")
             self.timestamp.snapshot_meta.hashes = hashes
             self.timestamp.snapshot_meta.length = length
 
@@ -320,7 +334,7 @@ def update_snapshot(self) -> None:
 
     def add_target(self, role: str, data: bytes, path: str) -> None:
         """Create a target from data and add it to the target_files."""
-        if role == Targets.type:
+        if role == "targets":
             targets = self.targets
         else:
             targets = self.md_delegates[role].signed
@@ -339,7 +353,7 @@ def add_delegation(
         hash_prefixes: Optional[List[str]],
     ) -> None:
         """Add delegated target role to the repository."""
-        if delegator_name == Targets.type:
+        if delegator_name == "targets":
             delegator = self.targets
         else:
             delegator = self.md_delegates[delegator_name].signed
@@ -375,9 +389,9 @@ def write(self) -> None:
 
         for ver in range(1, len(self.signed_roots) + 1):
             with open(os.path.join(dest_dir, f"{ver}.root.json"), "wb") as f:
-                f.write(self._fetch_metadata(Root.type, ver))
+                f.write(self._fetch_metadata("root", ver))
 
-        for role in [Timestamp.type, Snapshot.type, Targets.type]:
+        for role in ["timestamp", "snapshot", "targets"]:
             with open(os.path.join(dest_dir, f"{role}.json"), "wb") as f:
                 f.write(self._fetch_metadata(role))