Address pylint warnings on tests for the new code

Address or disable pylint warnings raised on all test files inside
tests/ testing the code of the new implementation.

Signed-off-by: Martin Vrachev <mvrachev@vmware.com>

Author: MVrachev

tests/repository_simulator.py

@@ -72,7 +72,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__)
@@ -89,6 +89,9 @@ class RepositoryTarget:
 
 
 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
@@ -134,11 +137,14 @@ 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
 
+    # pylint: disable-next=no-self-use
     def create_key(self) -> Tuple[Key, SSlibSigner]:
+        """Create ed25519 key"""
         sslib_key = generate_ed25519_key()
         return Key.from_securesystemslib_key(sslib_key), SSlibSigner(sslib_key)
 
@@ -178,6 +184,7 @@ 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 yeilds an Iterator"""
         if not self.root.consistent_snapshot:
             raise NotImplementedError("non-consistent snapshot not supported")
         path = parse.urlparse(url).path
@@ -188,20 +195,24 @@ def fetch(self, url: str) -> Iterator[bytes]:
                 version = None
                 role = "timestamp"
             else:
+                # pylint: disable-next=invalid-name
                 version, _, role = ver_and_name.partition(".")
                 version = int(version)
             yield self._fetch_metadata(role, version)
         elif path.startswith("/targets/"):
             # figure out target path and hash prefix
             target_path = path[len("/targets/") :]
             dir_parts, sep, prefixed_filename = target_path.rpartition("/")
+            # pylint: disable-next=invalid-name
             prefix, _, filename = prefixed_filename.partition(".")
             target_path = f"{dir_parts}{sep}{filename}"
 
             yield self._fetch_target(target_path, prefix)
         else:
             raise FetcherHTTPError(f"Unknown path '{path}'", 404)
 
+    # Disable the redefined-builtin warning raised because of the hash func.
+    # pylint: disable-next=redefined-builtin
     def _fetch_target(self, target_path: str, hash: Optional[str]) -> bytes:
         """Return data for 'target_path', checking 'hash' if it is given.
 
@@ -227,35 +238,32 @@ def _fetch_metadata(
             # 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]:
-                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]:
+            md.sign(signer, append=True)
+
+        num_signers = len(self.signers[role])
+        msg = f"fetched {role} {md.signed.version} with {num_signers} sigs"
+        logger.debug(msg)
+        return md.to_bytes(JSONSerializer())
 
     def _compute_hashes_and_length(
         self, role: str
@@ -267,6 +275,7 @@ def _compute_hashes_and_length(
         return hashes, len(data)
 
     def update_timestamp(self):
+        """Update timestamp and assign snapshot ver to snapshot_meta ver"""
         self.timestamp.snapshot_meta.version = self.snapshot.version
 
         if self.compute_metafile_hashes_length:
@@ -277,6 +286,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
@@ -291,6 +301,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:
@@ -309,6 +320,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:
@@ -342,17 +354,18 @@ 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))
 
+        # pylint: disable-next=consider-iterating-dictionary
         for role in self.md_delegates.keys():
-            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))

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)
 
@@ -279,7 +282,7 @@ def test_metadata_snapshot(self):
         # Create a MetaFile instance representing what we expect
         # the updated data to be.
         hashes = {
-            "sha256": "c2986576f5fdfd43944e2b19e775453b96748ec4fe2638a6d2f32f1310967095"
+            "sha256": "c2986576f5fdfd43944e2b19e775453b96748ec4fe2638a6d2f32f1310967095"  # pylint: disable=line-too-long
         }
         fileinfo = MetaFile(2, 123, hashes)
 
@@ -319,7 +322,7 @@ def test_metadata_timestamp(self):
         # Create a MetaFile instance representing what we expect
         # the updated data to be.
         hashes = {
-            "sha256": "0ae9664468150a9aa1e7f11feecb32341658eb84292851367fea2da88e8a58dc"
+            "sha256": "0ae9664468150a9aa1e7f11feecb32341658eb84292851367fea2da88e8a58dc"  # pylint: disable=line-too-long
         }
         fileinfo = MetaFile(2, 520, hashes)
 
@@ -472,6 +475,7 @@ def test_is_target_in_pathpattern(self):
         ]
         for targetpath, pathpattern in supported_use_cases:
             self.assertTrue(
+                # pylint: disable-next=protected-access
                 DelegatedRole._is_target_in_pathpattern(targetpath, pathpattern)
             )
 
@@ -485,18 +489,19 @@ def test_is_target_in_pathpattern(self):
         ]
         for targetpath, pathpattern in invalid_use_cases:
             self.assertFalse(
+                # pylint: disable-next=protected-access
                 DelegatedRole._is_target_in_pathpattern(targetpath, pathpattern)
             )
 
     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",
-            "sha512": "ef5beafa16041bcdd2937140afebd485296cd54f7348ecd5a4d035c09759608de467a7ac0eb58753d0242df873c305e8bffad2454aa48f44480f15efae1cacd0",
+            "sha256": "141f740f53781d1ca54b8a50af22cbf74e44c21a998fa2a8a05aaac2c002886b",  # pylint: disable=line-too-long
+            "sha512": "ef5beafa16041bcdd2937140afebd485296cd54f7348ecd5a4d035c09759608de467a7ac0eb58753d0242df873c305e8bffad2454aa48f44480f15efae1cacd0",  # pylint: disable=line-too-long
         }
 
         fileinfo = TargetFile(length=28, hashes=hashes, path=filename)
@@ -531,7 +536,7 @@ def test_targets_key_api(self):
         key_dict = {
             "keytype": "ed25519",
             "keyval": {
-                "public": "edcd0a32a07dce33f7c7873aaffbff36d20ea30787574ead335eefd337e4dacd"
+                "public": "edcd0a32a07dce33f7c7873aaffbff36d20ea30787574ead335eefd337e4dacd"  # pylint: disable=line-too-long
             },
             "scheme": "ed25519",
         }
@@ -628,7 +633,7 @@ def test_length_and_hash_validation(self):
             )
 
             snapshot_metafile.hashes = {
-                "unsupported-alg": "8f88e2ba48b412c3843e9bb26e1b6f8fc9e98aceb0fbaa97ba37b4c98717d7ab"
+                "unsupported-alg": "8f88e2ba48b412c3843e9bb26e1b6f8fc9e98aceb0fbaa97ba37b4c98717d7ab"  # pylint: disable=line-too-long
             }
             self.assertRaises(
                 exceptions.LengthOrHashMismatchError,
@@ -638,7 +643,7 @@ def test_length_and_hash_validation(self):
 
             # Test wrong algorithm format (sslib.FormatError)
             snapshot_metafile.hashes = {
-                256: "8f88e2ba48b412c3843e9bb26e1b6f8fc9e98aceb0fbaa97ba37b4c98717d7ab"
+                256: "8f88e2ba48b412c3843e9bb26e1b6f8fc9e98aceb0fbaa97ba37b4c98717d7ab"  # pylint: disable=line-too-long
             }
             self.assertRaises(
                 exceptions.LengthOrHashMismatchError,

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,7 +37,8 @@ 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")
+        # pylint: disable-next=consider-using-with
+        self.target_fileobj = open(target_filepath, "r", encoding="utf8")
         self.file_contents = self.target_fileobj.read()
         self.file_length = len(self.file_contents)
 
@@ -54,6 +57,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()
 

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,11 +50,16 @@ def setUp(self):
         current_dir = os.getcwd()
         target_filepath = self.make_temp_data_file(directory=current_dir)
 
-        self.target_fileobj = open(target_filepath, "r")
+        # pylint: disable-next=consider-using-with
+        self.target_fileobj = open(target_filepath, "r", encoding="utf8")
         self.file_contents = self.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()
@@ -106,7 +113,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)
 

tests/test_metadata_serialization.py

@@ -35,6 +35,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.