Add missing type annotations to updater

Add missing annotations and resolve partially mypy
errors in updater.py and trusted_metadata_set.py

Signed-off-by: Teodora Sechkova <tsechkova@vmware.com>

Author: sechkova

tuf/ngclient/_internal/trusted_metadata_set.py

@@ -68,10 +68,10 @@
 import logging
 from collections import abc
 from datetime import datetime
-from typing import Dict, Iterator, Optional
+from typing import Dict, Iterator, Mapping, Optional
 
 from tuf import exceptions
-from tuf.api.metadata import Metadata, Root, Targets
+from tuf.api.metadata import Key, Metadata, Root, Targets
 from tuf.api.serialization import DeserializationError
 
 logger = logging.getLogger(__name__)
@@ -83,7 +83,7 @@ def verify_with_threshold(
 ) -> bool:
     """Verify 'unverified' with keys and threshold defined in delegator"""
     role = None
-    keys = {}
+    keys: Mapping[str, Key] = {}
     if isinstance(delegator.signed, Root):
         keys = delegator.signed.keys
         role = delegator.signed.roles.get(role_name)
@@ -133,7 +133,7 @@ def __init__(self, root_data: bytes):
             RepositoryError: Metadata failed to load or verify. The actual
                 error type and content will contain more details.
         """
-        self._trusted_set = {}  # type: Dict[str: Metadata]
+        self._trusted_set = {}  # type: Dict[str, Metadata]
         self.reference_time = datetime.utcnow()
         self._root_update_finished = False
 
@@ -152,7 +152,7 @@ def __len__(self) -> int:
 
     def __iter__(self) -> Iterator[Metadata]:
         """Returns iterator over all Metadata objects in TrustedMetadataSet"""
-        return iter(self._trusted_set)
+        return iter(self._trusted_set.values())
 
     # Helper properties for top level metadata
     @property
@@ -176,7 +176,7 @@ def targets(self) -> Optional[Metadata]:
         return self._trusted_set.get("targets")
 
     # Methods for updating metadata
-    def update_root(self, data: bytes):
+    def update_root(self, data: bytes) -> None:
         """Verifies and loads 'data' as new root metadata.
 
         Note that an expired intermediate root is considered valid: expiry is
@@ -225,7 +225,7 @@ def update_root(self, data: bytes):
         self._trusted_set["root"] = new_root
         logger.debug("Updated root")
 
-    def root_update_finished(self):
+    def root_update_finished(self) -> None:
         """Marks root metadata as final and verifies it is not expired
 
         Raises:
@@ -245,7 +245,7 @@ def root_update_finished(self):
         self._root_update_finished = True
         logger.debug("Verified final root.json")
 
-    def update_timestamp(self, data: bytes):
+    def update_timestamp(self, data: bytes) -> None:
         """Verifies and loads 'data' as new timestamp metadata.
 
         Args:
@@ -302,7 +302,7 @@ def update_timestamp(self, data: bytes):
         self._trusted_set["timestamp"] = new_timestamp
         logger.debug("Updated timestamp")
 
-    def update_snapshot(self, data: bytes):
+    def update_snapshot(self, data: bytes) -> None:
         """Verifies and loads 'data' as new snapshot metadata.
 
         Args:
@@ -379,7 +379,7 @@ def update_snapshot(self, data: bytes):
         self._trusted_set["snapshot"] = new_snapshot
         logger.debug("Updated snapshot")
 
-    def update_targets(self, data: bytes):
+    def update_targets(self, data: bytes) -> None:
         """Verifies and loads 'data' as new top-level targets metadata.
 
         Args:
@@ -393,7 +393,7 @@ def update_targets(self, data: bytes):
 
     def update_delegated_targets(
         self, data: bytes, role_name: str, delegator_name: str
-    ):
+    ) -> None:
         """Verifies and loads 'data' as new metadata for target 'role_name'.
 
         Args:

tuf/ngclient/updater.py

@@ -14,6 +14,7 @@
 from securesystemslib import util as sslib_util
 
 from tuf import exceptions
+from tuf.api.metadata import DelegatedRole, TargetFile, Targets
 from tuf.ngclient._internal import requests_fetcher, trusted_metadata_set
 from tuf.ngclient.config import UpdaterConfig
 from tuf.ngclient.fetcher import FetcherInterface
@@ -59,12 +60,7 @@ def __init__(
         # Read trusted local root metadata
         data = self._load_local_metadata("root")
         self._trusted_set = trusted_metadata_set.TrustedMetadataSet(data)
-
-        if fetcher is None:
-            self._fetcher = requests_fetcher.RequestsFetcher()
-        else:
-            self._fetcher = fetcher
-
+        self._fetcher = fetcher or requests_fetcher.RequestsFetcher()
         self.config = config or UpdaterConfig()
 
     def refresh(self) -> None:
@@ -91,7 +87,7 @@ def refresh(self) -> None:
         self._load_snapshot()
         self._load_targets("targets", "root")
 
-    def get_one_valid_targetinfo(self, target_path: str) -> Dict:
+    def get_one_valid_targetinfo(self, target_path: str) -> Dict[str, Any]:
         """
         Returns the target information for a target identified by target_path.
 
@@ -135,7 +131,7 @@ def updated_targets(
             # 'destination_directory' if 'filepath' contains a leading path
             # separator (i.e., is treated as an absolute path).
             filepath = target["filepath"]
-            target_fileinfo: "TargetFile" = target["fileinfo"]
+            target_fileinfo: TargetFile = target["fileinfo"]
 
             target_filepath = os.path.join(destination_directory, filepath)
 
@@ -155,10 +151,10 @@ def updated_targets(
 
     def download_target(
         self,
-        targetinfo: Dict,
+        targetinfo: Dict[str, Any],
         destination_directory: str,
         target_base_url: Optional[str] = None,
-    ):
+    ) -> None:
         """
         Download target specified by 'targetinfo' into 'destination_directory'.
 
@@ -175,18 +171,20 @@ def download_target(
             TODO: download-related errors
             TODO: file write errors
         """
-        if target_base_url is None and self._target_base_url is None:
-            raise ValueError(
-                "target_base_url must be set in either download_target() or "
-                "constructor"
-            )
+
         if target_base_url is None:
+            if self._target_base_url is None:
+                raise ValueError(
+                    "target_base_url must be set in either "
+                    "download_target() or constructor"
+                )
+
             target_base_url = self._target_base_url
         else:
             target_base_url = _ensure_trailing_slash(target_base_url)
 
-        target_filepath = targetinfo["filepath"]
-        target_fileinfo: "TargetFile" = targetinfo["fileinfo"]
+        target_filepath: str = targetinfo["filepath"]
+        target_fileinfo: TargetFile = targetinfo["fileinfo"]
         full_url = parse.urljoin(target_base_url, target_filepath)
 
         with self._fetcher.download_file(
@@ -217,7 +215,7 @@ def _load_local_metadata(self, rolename: str) -> bytes:
         with open(os.path.join(self._dir, f"{rolename}.json"), "rb") as f:
             return f.read()
 
-    def _persist_metadata(self, rolename: str, data: bytes):
+    def _persist_metadata(self, rolename: str, data: bytes) -> None:
         with open(os.path.join(self._dir, f"{rolename}.json"), "wb") as f:
             f.write(data)
 
@@ -275,6 +273,7 @@ def _load_snapshot(self) -> None:
             # Local snapshot does not exist or is invalid: update from remote
             logger.debug("Failed to load local snapshot %s", e)
 
+            assert self._trusted_set.timestamp is not None
             metainfo = self._trusted_set.timestamp.signed.meta["snapshot.json"]
             length = metainfo.length or self.config.snapshot_max_length
             version = None
@@ -295,6 +294,7 @@ def _load_targets(self, role: str, parent_role: str) -> None:
             # Local 'role' does not exist or is invalid: update from remote
             logger.debug("Failed to load local %s: %s", role, e)
 
+            assert self._trusted_set.snapshot is not None
             metainfo = self._trusted_set.snapshot.signed.meta[f"{role}.json"]
             length = metainfo.length or self.config.targets_max_length
             version = None
@@ -305,7 +305,9 @@ def _load_targets(self, role: str, parent_role: str) -> None:
             self._trusted_set.update_delegated_targets(data, role, parent_role)
             self._persist_metadata(role, data)
 
-    def _preorder_depth_first_walk(self, target_filepath) -> Dict:
+    def _preorder_depth_first_walk(
+        self, target_filepath: str
+    ) -> Dict[str, Any]:
         """
         Interrogates the tree of target delegations in order of appearance
         (which implicitly order trustworthiness), and returns the matching
@@ -333,7 +335,7 @@ def _preorder_depth_first_walk(self, target_filepath) -> Dict:
             # The metadata for 'role_name' must be downloaded/updated before
             # its targets, delegations, and child roles can be inspected.
 
-            role_metadata = self._trusted_set[role_name].signed
+            role_metadata: Targets = self._trusted_set[role_name].signed
             target = role_metadata.targets.get(target_filepath)
 
             # After preorder check, add current role to set of visited roles.
@@ -400,7 +402,9 @@ def _preorder_depth_first_walk(self, target_filepath) -> Dict:
         return {"filepath": target_filepath, "fileinfo": target}
 
 
-def _visit_child_role(child_role: Dict, target_filepath: str) -> str:
+def _visit_child_role(
+    child_role: DelegatedRole, target_filepath: str
+) -> Optional[str]:
     """
     <Purpose>
       Non-public method that determines whether the given 'target_filepath'
@@ -496,7 +500,9 @@ def _visit_child_role(child_role: Dict, target_filepath: str) -> str:
     return None
 
 
-def _get_filepath_hash(target_filepath, hash_function="sha256"):
+def _get_filepath_hash(
+    target_filepath: str, hash_function: str = "sha256"
+) -> str:
     """
     Calculate the hash of the filepath to determine which bin to find the
     target.
@@ -511,6 +517,6 @@ def _get_filepath_hash(target_filepath, hash_function="sha256"):
     return target_filepath_hash
 
 
-def _ensure_trailing_slash(url: str):
+def _ensure_trailing_slash(url: str) -> str:
     """Return url guaranteed to end in a slash"""
     return url if url.endswith("/") else f"{url}/"