Add missing type annotations to updater

Add missing annotations and partially resolve 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

@@ -92,7 +92,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: Dict[str, Metadata] = {}
         self.reference_time = datetime.utcnow()
 
         # Load and validate the local root metadata. Valid initial trusted root
@@ -134,7 +134,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
@@ -175,7 +175,7 @@ def update_root(self, data: bytes):
         self._trusted_set["root"] = new_root
         logger.debug("Updated root")
 
-    def update_timestamp(self, data: bytes):
+    def update_timestamp(self, data: bytes) -> None:
         """Verifies and loads 'data' as new timestamp metadata.
 
         Note that an expired intermediate timestamp is considered valid so it
@@ -237,7 +237,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.
 
         Note that intermediate snapshot is considered valid even if it is
@@ -314,7 +314,9 @@ def update_snapshot(self, data: bytes):
         self._trusted_set["snapshot"] = new_snapshot
         logger.debug("Updated snapshot")
 
-    def _check_final_snapshot(self):
+    def _check_final_snapshot(self) -> None:
+        """Check snapshot expiry and version before targets is updated"""
+
         if self.snapshot.signed.is_expired(self.reference_time):
             raise exceptions.ExpiredMetadataError("snapshot.json is expired")
 
@@ -328,7 +330,7 @@ def _check_final_snapshot(self):
                 f"got {self.snapshot.signed.version}"
             )
 
-    def update_targets(self, data: bytes):
+    def update_targets(self, data: bytes) -> None:
         """Verifies and loads 'data' as new top-level targets metadata.
 
         Args:
@@ -342,7 +344,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

@@ -68,7 +68,7 @@
 from securesystemslib import util as sslib_util
 
 from tuf import exceptions
-from tuf.api.metadata import Targets
+from tuf.api.metadata import TargetFile, Targets
 from tuf.ngclient._internal import requests_fetcher, trusted_metadata_set
 from tuf.ngclient.config import UpdaterConfig
 from tuf.ngclient.fetcher import FetcherInterface
@@ -112,12 +112,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:
@@ -205,7 +200,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)
 
@@ -225,10 +220,10 @@ def updated_targets(
 
     def download_target(
         self,
-        targetinfo: Dict,
+        targetinfo: Dict[str, Any],
         destination_directory: str,
         target_base_url: Optional[str] = None,
-    ):
+    ) -> None:
         """Downloads the target file specified by 'targetinfo'.
 
         Args:
@@ -244,18 +239,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_fileinfo: "TargetFile" = targetinfo["fileinfo"]
-        target_filepath = targetinfo["filepath"]
+        target_fileinfo: TargetFile = targetinfo["fileinfo"]
+        target_filepath: str = targetinfo["filepath"]
         consistent_snapshot = self._trusted_set.root.signed.consistent_snapshot
         if consistent_snapshot and self.config.prefix_targets_with_hash:
             hashes = list(target_fileinfo.hashes.values())
@@ -293,7 +290,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)
 
@@ -454,6 +451,6 @@ def _preorder_depth_first_walk(
         return None
 
 
-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}/"