Address pylint warnings on tests for the new code

MVrachev · MVrachev · commit ce10b69a8b79 · 2021-11-08T18:11:12.000+02:00
Address or disable pylint warnings raised on all test files inside
the "tests/" directory testing the code of the new implementation.

Signed-off-by: Martin Vrachev &lt;mvrachev@vmware.com&gt;
diff --git a/tests/repository_simulator.py b/tests/repository_simulator.py
@@ -73,7 +73,7 @@
     Timestamp,
 )
 from tuf.api.serialization.json import JSONSerializer
-from tuf.exceptions import FetcherHTTPError, RepositoryError
+from tuf.exceptions import FetcherHTTPError
 from tuf.ngclient.fetcher import FetcherInterface
 
 logger = logging.getLogger(__name__)
@@ -83,13 +83,16 @@
 
 @dataclass
 class RepositoryTarget:
-    """Contains actual target data and the related target metadata"""
+    """Contains actual target data and the related target metadata."""
 
     data: bytes
     target_file: TargetFile
 
 
 class RepositorySimulator(FetcherInterface):
+    """Simulates a repository that can be used for testing."""
+
+    # pylint: disable=too-many-instance-attributes
     def __init__(self):
         self.md_root: Metadata[Root] = None
         self.md_timestamp: Metadata[Timestamp] = None
@@ -136,6 +139,7 @@ def targets(self) -> Targets:
         return self.md_targets.signed
 
     def all_targets(self) -> Iterator[Tuple[str, Targets]]:
+        """Yield role name and signed portion of targets one by one."""
         yield "targets", self.md_targets.signed
         for role, md in self.md_delegates.items():
             yield role, md.signed
@@ -151,7 +155,7 @@ def add_signer(self, role: str, signer: SSlibSigner):
         self.signers[role][signer.key_dict["keyid"]] = signer
 
     def _initialize(self):
-        """Setup a minimal valid repository"""
+        """Setup a minimal valid repository."""
 
         targets = Targets(1, SPEC_VER, self.safe_expiry, {}, None)
         self.md_targets = Metadata(targets, OrderedDict())
@@ -176,7 +180,7 @@ def _initialize(self):
         self.publish_root()
 
     def publish_root(self):
-        """Sign and store a new serialized version of root"""
+        """Sign and store a new serialized version of root."""
         self.md_root.signatures.clear()
         for signer in self.signers["root"].values():
             self.md_root.sign(signer, append=True)
@@ -185,6 +189,9 @@ def publish_root(self):
         logger.debug("Published root v%d", self.root.version)
 
     def fetch(self, url: str) -> Iterator[bytes]:
+        """Fetches data from the given url and returns an Iterator (or yields
+        bytes).
+        """
         if not self.root.consistent_snapshot:
             raise NotImplementedError("non-consistent snapshot not supported")
         path = parse.urlparse(url).path
@@ -209,15 +216,20 @@ def fetch(self, url: str) -> Iterator[bytes]:
         else:
             raise FetcherHTTPError(f"Unknown path '{path}'", 404)
 
-    def _fetch_target(self, target_path: str, hash: Optional[str]) -> bytes:
-        """Return data for 'target_path', checking 'hash' if it is given.
+    def _fetch_target(
+        self, target_path: str, target_hash: Optional[str]
+    ) -> bytes:
+        """Return data for 'target_path', checking 'target_hash' if it is given.
 
-        If hash is None, then consistent_snapshot is not used
+        If hash is None, then consistent_snapshot is not used.
         """
         repo_target = self.target_files.get(target_path)
         if repo_target is None:
             raise FetcherHTTPError(f"No target {target_path}", 404)
-        if hash and hash not in repo_target.target_file.hashes.values():
+        if (
+            target_hash
+            and target_hash not in repo_target.target_file.hashes.values()
+        ):
             raise FetcherHTTPError(f"hash mismatch for {target_path}", 404)
 
         logger.debug("fetched target %s", target_path)
@@ -228,41 +240,41 @@ def _fetch_metadata(
     ) -> bytes:
         """Return signed metadata for 'role', using 'version' if it is given.
 
-        If version is None, non-versioned metadata is being requested
+        If version is None, non-versioned metadata is being requested.
         """
         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", role, version)
+            logger.debug("fetched root version %d", version)
             return self.signed_roots[version - 1]
+
+        # sign and serialize the requested metadata
+        if role == "timestamp":
+            md: Metadata = self.md_timestamp
+        elif role == "snapshot":
+            md = self.md_snapshot
+        elif role == "targets":
+            md = self.md_targets
         else:
-            # sign and serialize the requested metadata
-            if role == "timestamp":
-                md: Metadata = self.md_timestamp
-            elif role == "snapshot":
-                md = self.md_snapshot
-            elif role == "targets":
-                md = self.md_targets
-            else:
-                md = self.md_delegates[role]
-
-            if md is None:
-                raise FetcherHTTPError(f"Unknown role {role}", 404)
-            if version is not None and version != md.signed.version:
-                raise FetcherHTTPError(f"Unknown {role} version {version}", 404)
-
-            md.signatures.clear()
-            for signer in self.signers[role].values():
-                md.sign(signer, append=True)
-
-            logger.debug(
-                "fetched %s v%d with %d sigs",
-                role,
-                md.signed.version,
-                len(self.signers[role]),
-            )
-            return md.to_bytes(JSONSerializer())
+            md = self.md_delegates[role]
+
+        if md is None:
+            raise FetcherHTTPError(f"Unknown role {role}", 404)
+        if version is not None and version != md.signed.version:
+            raise FetcherHTTPError(f"Unknown {role} version {version}", 404)
+
+        md.signatures.clear()
+        for signer in self.signers[role].values():
+            md.sign(signer, append=True)
+
+        logger.debug(
+            "fetched %s v%d with %d sigs",
+            role,
+            md.signed.version,
+            len(self.signers[role]),
+        )
+        return md.to_bytes(JSONSerializer())
 
     def _compute_hashes_and_length(
         self, role: str
@@ -274,6 +286,9 @@ def _compute_hashes_and_length(
         return hashes, len(data)
 
     def update_timestamp(self):
+        """Update timestamp and assign snapshot version to snapshot_meta
+        version.
+        """
         self.timestamp.snapshot_meta.version = self.snapshot.version
 
         if self.compute_metafile_hashes_length:
@@ -284,6 +299,7 @@ def update_timestamp(self):
         self.timestamp.version += 1
 
     def update_snapshot(self):
+        """Update snapshot, assign targets versions and update timestamp."""
         for role, delegate in self.all_targets():
             hashes = None
             length = None
@@ -298,6 +314,7 @@ def update_snapshot(self):
         self.update_timestamp()
 
     def add_target(self, role: str, data: bytes, path: str):
+        """Create a target from data and add it to the target_files."""
         if role == "targets":
             targets = self.targets
         else:
@@ -316,6 +333,7 @@ def add_delegation(
         paths: Optional[List[str]],
         hash_prefixes: Optional[List[str]],
     ):
+        """Add delegated target role to the repository."""
         if delegator_name == "targets":
             delegator = self.targets
         else:
@@ -347,17 +365,17 @@ def write(self):
             print(f"Repository Simulator dumps in {self.dump_dir}")
 
         self.dump_version += 1
-        dir = os.path.join(self.dump_dir, str(self.dump_version))
-        os.makedirs(dir)
+        dest_dir = os.path.join(self.dump_dir, str(self.dump_version))
+        os.makedirs(dest_dir)
 
         for ver in range(1, len(self.signed_roots) + 1):
-            with open(os.path.join(dir, f"{ver}.root.json"), "wb") as f:
+            with open(os.path.join(dest_dir, f"{ver}.root.json"), "wb") as f:
                 f.write(self._fetch_metadata("root", ver))
 
         for role in ["timestamp", "snapshot", "targets"]:
-            with open(os.path.join(dir, f"{role}.json"), "wb") as f:
+            with open(os.path.join(dest_dir, f"{role}.json"), "wb") as f:
                 f.write(self._fetch_metadata(role))
 
-        for role in self.md_delegates.keys():
-            with open(os.path.join(dir, f"{role}.json"), "wb") as f:
+        for role in self.md_delegates:
+            with open(os.path.join(dest_dir, f"{role}.json"), "wb") as f:
                 f.write(self._fetch_metadata(role))
diff --git a/tests/test_api.py b/tests/test_api.py
@@ -44,6 +44,8 @@
 
 
 class TestMetadata(unittest.TestCase):
+    """Tests for public API of all classes in 'tuf/api/metadata.py'."""
+
     @classmethod
     def setUpClass(cls):
         # Create a temporary directory to store the repository, metadata, and
@@ -155,8 +157,8 @@ def test_to_from_bytes(self):
             self.assertEqual(metadata_obj_2.to_bytes(), obj_bytes)
 
     def test_sign_verify(self):
-        root_path = os.path.join(self.repo_dir, "metadata", "root.json")
-        root = Metadata[Root].from_file(root_path).signed
+        path = os.path.join(self.repo_dir, "metadata", "root.json")
+        root = Metadata[Root].from_file(path).signed
 
         # Locate the public keys we need from root
         targets_keyid = next(iter(root.roles["targets"].keyids))
@@ -203,7 +205,8 @@ def test_sign_verify(self):
         with self.assertRaises(exceptions.UnsignedMetadataError):
             targets_key.verify_signature(md_obj)
 
-        # Test failure on unknown scheme (securesystemslib UnsupportedAlgorithmError)
+        # Test failure on unknown scheme (securesystemslib
+        # UnsupportedAlgorithmError)
         scheme = timestamp_key.scheme
         timestamp_key.scheme = "foo"
         with self.assertRaises(exceptions.UnsignedMetadataError):
@@ -231,8 +234,8 @@ def test_sign_verify(self):
         sig.signature = correct_sig
 
     def test_metadata_base(self):
-        # Use of Snapshot is arbitrary, we're just testing the base class features
-        # with real data
+        # Use of Snapshot is arbitrary, we're just testing the base class
+        # features with real data
         snapshot_path = os.path.join(self.repo_dir, "metadata", "snapshot.json")
         md = Metadata.from_file(snapshot_path)
 
@@ -458,6 +461,7 @@ def test_root_add_key_and_remove_key(self):
             root.signed.remove_key("nosuchrole", keyid)
 
     def test_is_target_in_pathpattern(self):
+        # pylint: disable=protected-access
         supported_use_cases = [
             ("foo.tgz", "foo.tgz"),
             ("foo.tgz", "*"),
@@ -492,7 +496,7 @@ def test_metadata_targets(self):
         targets_path = os.path.join(self.repo_dir, "metadata", "targets.json")
         targets = Metadata[Targets].from_file(targets_path)
 
-        # Create a fileinfo dict representing what we expect the updated data to be
+        # Create a fileinfo dict representing the expected updated data.
         filename = "file2.txt"
         hashes = {
             "sha256": "141f740f53781d1ca54b8a50af22cbf74e44c21a998fa2a8a05aaac2c002886b",
diff --git a/tests/test_fetcher.py b/tests/test_fetcher.py
@@ -17,13 +17,15 @@
 import tuf
 import tuf.exceptions
 import tuf.requests_fetcher
-import tuf.unittest_toolbox as unittest_toolbox
 from tests import utils
+from tuf import unittest_toolbox
 
 logger = logging.getLogger(__name__)
 
 
 class TestFetcher(unittest_toolbox.Modified_TestCase):
+    """Unit tests for RequestFetcher."""
+
     def setUp(self):
         """
         Create a temporary file and launch a simple server in the
@@ -35,9 +37,9 @@ def setUp(self):
         # Making a temporary file.
         current_dir = os.getcwd()
         target_filepath = self.make_temp_data_file(directory=current_dir)
-        self.target_fileobj = open(target_filepath, "r")
-        self.file_contents = self.target_fileobj.read()
-        self.file_length = len(self.file_contents)
+        with open(target_filepath, "r", encoding="utf8") as target_fileobj:
+            self.file_contents = target_fileobj.read()
+            self.file_length = len(self.file_contents)
 
         # Launch a SimpleHTTPServer (serves files in the current dir).
         self.server_process_handler = utils.TestServerProcess(log=logger)
@@ -54,6 +56,7 @@ def setUp(self):
 
         # Create a temporary file where the target file chunks are written
         # during fetching
+        # pylint: disable-next=consider-using-with
         self.temp_file = tempfile.TemporaryFile()
         self.fetcher = tuf.requests_fetcher.RequestsFetcher()
 
@@ -62,7 +65,6 @@ def tearDown(self):
         # Cleans the resources and flush the logged lines (if any).
         self.server_process_handler.clean()
 
-        self.target_fileobj.close()
         self.temp_file.close()
 
         # Remove temporary directory
diff --git a/tests/test_fetcher_ng.py b/tests/test_fetcher_ng.py
@@ -26,6 +26,8 @@
 
 
 class TestFetcher(unittest_toolbox.Modified_TestCase):
+    """Test RequestsFetcher class."""
+
     @classmethod
     def setUpClass(cls):
         # Launch a SimpleHTTPServer (serves files in the current dir).
@@ -48,17 +50,21 @@ def setUp(self):
         current_dir = os.getcwd()
         target_filepath = self.make_temp_data_file(directory=current_dir)
 
-        self.target_fileobj = open(target_filepath, "r")
-        self.file_contents = self.target_fileobj.read()
-        self.file_length = len(self.file_contents)
+        with open(target_filepath, "r", encoding="utf8") as target_fileobj:
+            self.file_contents = target_fileobj.read()
+            self.file_length = len(self.file_contents)
+
         self.rel_target_filepath = os.path.basename(target_filepath)
-        self.url = f"http://{utils.TEST_HOST_ADDRESS}:{str(self.server_process_handler.port)}/{self.rel_target_filepath}"
+        self.url_prefix = (
+            f"http://{utils.TEST_HOST_ADDRESS}:"
+            f"{str(self.server_process_handler.port)}"
+        )
+        self.url = f"{self.url_prefix}/{self.rel_target_filepath}"
 
         # Instantiate a concrete instance of FetcherInterface
         self.fetcher = RequestsFetcher()
 
     def tearDown(self):
-        self.target_fileobj.close()
         # Remove temporary directory
         unittest_toolbox.Modified_TestCase.tearDown(self)
 
@@ -106,7 +112,7 @@ def test_url_parsing(self):
     # File not found error
     def test_http_error(self):
         with self.assertRaises(exceptions.FetcherHTTPError) as cm:
-            self.url = f"http://{utils.TEST_HOST_ADDRESS}:{str(self.server_process_handler.port)}/non-existing-path"
+            self.url = f"{self.url_prefix}/non-existing-path"
             self.fetcher.fetch(self.url)
         self.assertEqual(cm.exception.status_code, 404)
 
diff --git a/tests/test_metadata_serialization.py b/tests/test_metadata_serialization.py
@@ -31,6 +31,7 @@
 
 
 class TestSerialization(unittest.TestCase):
+    """Test serialization for all classes in 'tuf/api/metadata.py'."""
 
     # Snapshot instances with meta = {} are valid, but for a full valid
     # repository it's required that meta has at least one element inside it.
diff --git a/tests/test_trusted_metadata_set.py b/tests/test_trusted_metadata_set.py
@@ -1,3 +1,4 @@
+"""Unit tests for 'tuf/ngclient/_internal/trusted_metadata_set.py'."""
 import logging
 import os
 import sys
@@ -26,16 +27,18 @@
 
 logger = logging.getLogger(__name__)
 
-
+# pylint: disable=too-many-public-methods
 class TestTrustedMetadataSet(unittest.TestCase):
+    """Tests for all public API of the TrustedMetadataSet class."""
+
     def modify_metadata(
-        self, rolename: str, modification_func: Callable[["Signed"], None]
+        self, rolename: str, modification_func: Callable[[Signed], None]
     ) -> bytes:
         """Instantiate metadata from rolename type, call modification_func and
         sign it again with self.keystore[rolename] signer.
 
         Attributes:
-            rolename: A denoting the name of the metadata which will be modified.
+            rolename: Denoting the name of the metadata which will be modified.
             modification_func: Function that will be called to modify the signed
                 portion of metadata bytes.
         """
diff --git a/tests/test_updater_ng.py b/tests/test_updater_ng.py
diff --git a/tests/test_updater_with_simulator.py b/tests/test_updater_with_simulator.py
