tests: lint fixes

Signed-off-by: Jussi Kukkonen <[email protected]>

Author: jku

tests/test_updater_delegation_graphs.py

@@ -135,7 +135,9 @@ def _init_updater(self) -> Updater:
     def _assert_files_exist(self, roles: Iterable[str]) -> None:
         """Assert that local metadata files match 'roles'"""
         expected_files = [f"{role}.json" for role in roles]
-        found_files = [e.name for e in os.scandir(self.metadata_dir) if e.is_file()]
+        found_files = [
+            e.name for e in os.scandir(self.metadata_dir) if e.is_file()
+        ]
 
         self.assertListEqual(sorted(found_files), sorted(expected_files))
 

tests/test_updater_ng.py

@@ -5,14 +5,13 @@
 
 from __future__ import annotations
 
-from collections.abc import Iterable
 import logging
 import os
 import shutil
 import sys
 import tempfile
 import unittest
-from typing import Callable, ClassVar
+from typing import TYPE_CHECKING, Callable, ClassVar
 from unittest.mock import MagicMock, patch
 
 from securesystemslib.signer import Signer
@@ -29,6 +28,9 @@
 )
 from tuf.ngclient import Updater, UpdaterConfig
 
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
 logger = logging.getLogger(__name__)
 
 
@@ -153,7 +155,9 @@ def _modify_repository_root(
     def _assert_files_exist(self, roles: Iterable[str]) -> None:
         """Assert that local metadata files match 'roles'"""
         expected_files = [f"{role}.json" for role in roles]
-        found_files = [e.name for e in os.scandir(self.client_directory) if e.is_file()]
+        found_files = [
+            e.name for e in os.scandir(self.client_directory) if e.is_file()
+        ]
 
         self.assertListEqual(sorted(found_files), sorted(expected_files))
 

tests/test_updater_top_level_update.py

@@ -5,7 +5,6 @@
 
 import builtins
 import datetime
-import logging
 import os
 import sys
 import tempfile
@@ -68,13 +67,13 @@ def setUp(self) -> None:
     def tearDown(self) -> None:
         self.temp_dir.cleanup()
 
-    def _run_refresh(self, skip_bootstrap:bool=False) -> Updater:
+    def _run_refresh(self, skip_bootstrap: bool = False) -> Updater:
         """Create a new Updater instance and refresh"""
         updater = self._init_updater(skip_bootstrap)
         updater.refresh()
         return updater
 
-    def _init_updater(self, skip_bootstrap:bool=False) -> Updater:
+    def _init_updater(self, skip_bootstrap: bool = False) -> Updater:
         """Create a new Updater instance"""
         if self.dump_dir is not None:
             self.sim.write()
@@ -85,13 +84,15 @@ def _init_updater(self, skip_bootstrap:bool=False) -> Updater:
             self.targets_dir,
             "https://example.com/targets/",
             self.sim,
-            bootstrap=None if skip_bootstrap else self.sim.signed_roots[0]
+            bootstrap=None if skip_bootstrap else self.sim.signed_roots[0],
         )
 
     def _assert_files_exist(self, roles: Iterable[str]) -> None:
         """Assert that local metadata files match 'roles'"""
         expected_files = [f"{role}.json" for role in roles]
-        found_files = [e.name for e in os.scandir(self.metadata_dir) if e.is_file()]
+        found_files = [
+            e.name for e in os.scandir(self.metadata_dir) if e.is_file()
+        ]
 
         self.assertListEqual(sorted(found_files), sorted(expected_files))
 
@@ -700,9 +701,10 @@ def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None:
         updater.get_targetinfo("non_existent_target")
 
         # Test that metadata is loaded from cache and not downloaded
+        root_dir = os.path.join(self.metadata_dir, "root_history")
         wrapped_open.assert_has_calls(
             [
-                call(os.path.join(self.metadata_dir, "root_history/2.root.json"), "rb"),
+                call(os.path.join(root_dir, "2.root.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "timestamp.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "snapshot.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "targets.json"), "rb"),
@@ -713,7 +715,6 @@ def test_load_metadata_from_cache(self, wrapped_open: MagicMock) -> None:
         expected_calls = [("root", 2), ("timestamp", None)]
         self.assertListEqual(self.sim.fetch_tracker.metadata, expected_calls)
 
-
     @patch.object(builtins, "open", wraps=builtins.open)
     def test_intermediate_root_cache(self, wrapped_open: MagicMock) -> None:
         """Test that refresh uses the intermediate roots from cache"""
@@ -727,17 +728,25 @@ def test_intermediate_root_cache(self, wrapped_open: MagicMock) -> None:
         self._run_refresh()
 
         # assert that cache lookups happened but data was downloaded from remote
+        root_dir = os.path.join(self.metadata_dir, "root_history")
         wrapped_open.assert_has_calls(
             [
-                call(os.path.join(self.metadata_dir, "root_history/2.root.json"), "rb"),
-                call(os.path.join(self.metadata_dir, "root_history/3.root.json"), "rb"),
-                call(os.path.join(self.metadata_dir, "root_history/4.root.json"), "rb"),
+                call(os.path.join(root_dir, "2.root.json"), "rb"),
+                call(os.path.join(root_dir, "3.root.json"), "rb"),
+                call(os.path.join(root_dir, "4.root.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "timestamp.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "snapshot.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "targets.json"), "rb"),
             ]
         )
-        expected_calls = [("root", 2), ("root", 3), ("root", 4), ("timestamp", None), ("snapshot", 1), ("targets", 1)]
+        expected_calls = [
+            ("root", 2),
+            ("root", 3),
+            ("root", 4),
+            ("timestamp", None),
+            ("snapshot", 1),
+            ("targets", 1),
+        ]
         self.assertListEqual(self.sim.fetch_tracker.metadata, expected_calls)
 
         # Clear statistics for open() calls and metadata requests
@@ -748,9 +757,9 @@ def test_intermediate_root_cache(self, wrapped_open: MagicMock) -> None:
         self._run_refresh()
         wrapped_open.assert_has_calls(
             [
-                call(os.path.join(self.metadata_dir, "root_history/2.root.json"), "rb"),
-                call(os.path.join(self.metadata_dir, "root_history/3.root.json"), "rb"),
-                call(os.path.join(self.metadata_dir, "root_history/4.root.json"), "rb"),
+                call(os.path.join(root_dir, "2.root.json"), "rb"),
+                call(os.path.join(root_dir, "3.root.json"), "rb"),
+                call(os.path.join(root_dir, "4.root.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "timestamp.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "snapshot.json"), "rb"),
                 call(os.path.join(self.metadata_dir, "targets.json"), "rb"),
@@ -771,7 +780,9 @@ def test_intermediate_root_cache_poisoning(self) -> None:
         self._run_refresh()
 
         # Modify cached intermediate root v2 so that it's no longer signed correctly
-        root_path = os.path.join(self.metadata_dir, "root_history", "2.root.json")
+        root_path = os.path.join(
+            self.metadata_dir, "root_history", "2.root.json"
+        )
         md = Metadata.from_file(root_path)
         md.signatures.clear()
         md.to_file(root_path)