Annotations and use extensively modify_metadata

Signed-off-by: Martin Vrachev <[email protected]>

Author: MVrachev

tests/test_trusted_metadata_set.py

@@ -6,7 +6,15 @@
 from datetime import datetime
 
 from tuf import exceptions
-from tuf.api.metadata import Metadata, Signed, Timestamp, Snapshot, MetaFile
+from tuf.api.metadata import (
+    Metadata,
+    Signed,
+    Root,
+    Timestamp,
+    Snapshot,
+    MetaFile,
+    Targets
+)
 from tuf.ngclient._internal.trusted_metadata_set import TrustedMetadataSet
 
 from securesystemslib.signer import SSlibSigner
@@ -21,6 +29,22 @@
 
 class TestTrustedMetadataSet(unittest.TestCase):
 
+    def modify_metadata(
+        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.
+            modification_func: Function that will be called to modify the signed
+                portion of metadata bytes.
+        """
+        metadata = Metadata.from_bytes(self.metadata[rolename])
+        modification_func(metadata.signed)
+        metadata.sign(self.keystore[rolename])
+        return metadata.to_bytes()
+
     @classmethod
     def setUpClass(cls):
         cls.repo_dir = os.path.join(
@@ -45,12 +69,20 @@ def setUpClass(cls):
             )
             cls.keystore[role] = SSlibSigner(key_dict)
 
+        def hashes_length_modifier(timestamp: Timestamp) -> None:
+            timestamp.meta["snapshot.json"].hashes = None
+            timestamp.meta["snapshot.json"].length = None
+
+        cls.metadata["timestamp"] = cls.modify_metadata(
+            cls, "timestamp", hashes_length_modifier
+        )
+
     def setUp(self) -> None:
         self.trusted_set = TrustedMetadataSet(self.metadata["root"])
 
     def _root_updated_and_update_timestamp(
         self, timestamp_bytes: Optional[bytes] = None
-    ):
+    ) -> None:
         """Finsh root update and update timestamp with passed timestamp_bytes.
 
         Args:
@@ -84,15 +116,6 @@ def _update_all_besides_targets(
         snapshot_bytes = snapshot_bytes or self.metadata["snapshot"]
         self.trusted_set.update_snapshot(snapshot_bytes)
 
-    def modify_metadata(
-        self, rolename: str, modification_func: Callable[["Signed"], None]
-    ):
-        """Instantiate metadata from rolename type, call modification_func and
-        sign it again with self.keystore[rolename] signer."""
-        metadata = Metadata.from_bytes(self.metadata[rolename])
-        modification_func(metadata.signed)
-        metadata.sign(self.keystore[rolename])
-        return metadata.to_bytes()
 
     def test_update(self):
         self.trusted_set.root_update_finished()
@@ -177,10 +200,6 @@ def test_update_with_invalid_json(self):
         ]
         for metadata, update_func in top_level_md:
             md = Metadata.from_bytes(metadata)
-            if md.signed.type == "snapshot":
-                # timestamp hashes and length intervene when testing snapshot
-                self.trusted_set.timestamp.signed.meta["snapshot.json"].hashes = None
-                self.trusted_set.timestamp.signed.meta["snapshot.json"].length = None
             # metadata is not json
             with self.assertRaises(exceptions.RepositoryError):
                 update_func(b"")
@@ -210,26 +229,28 @@ def test_update_root_new_root_ver_same_as_trusted_root_ver(self):
 
 
     def test_root_update_finished_expired(self):
-        root = Metadata.from_bytes(self.metadata["root"])
-        root.signed.expires = datetime(1970, 1, 1)
-        root.sign(self.keystore["root"])
-        tmp_trusted_set = TrustedMetadataSet(root.to_bytes())
+        def root_expired_modifier(root: Root) -> None:
+            root.expires = datetime(1970, 1, 1)
+ 
+        root = self.modify_metadata("root", root_expired_modifier)
+        tmp_trusted_set = TrustedMetadataSet(root)
         # call root_update_finished when trusted root has expired
         with self.assertRaises(exceptions.ExpiredMetadataError):
             tmp_trusted_set.root_update_finished()
 
 
     def test_update_timestamp_new_timestamp_ver_below_trusted_ver(self):
         # new_timestamp.version < trusted_timestamp.version
-        timestamp = Metadata.from_bytes(self.metadata["timestamp"])
-        timestamp.signed.version = 3
-        timestamp.sign(self.keystore["timestamp"])
-        self._root_updated_and_update_timestamp(timestamp.to_bytes())
+        def version_modifier(timestamp: Timestamp) -> None:
+            timestamp.version = 3
+    
+        timestamp = self.modify_metadata("timestamp", version_modifier)
+        self._root_updated_and_update_timestamp(timestamp)
         with self.assertRaises(exceptions.ReplayedMetadataError):
             self.trusted_set.update_timestamp(self.metadata["timestamp"])
 
     def test_update_timestamp_snapshot_ver_below_trusted_snapshot_ver(self):
-        def version_modifier(timestamp: Timestamp):
+        def version_modifier(timestamp: Timestamp) -> None:
             timestamp.version = 3
 
         modified_timestamp = self.modify_metadata("timestamp", version_modifier)
@@ -241,142 +262,109 @@ def version_modifier(timestamp: Timestamp):
     def test_update_timestamp_expired(self):
         self.trusted_set.root_update_finished()
         # new_timestamp has expired
-        timestamp = Metadata.from_bytes(self.metadata["timestamp"])
-        timestamp.signed.expires = datetime(1970, 1, 1)
-        timestamp.sign(self.keystore["timestamp"])
+        def timestamp_expired_modifier(timestamp: Timestamp) -> None:
+            timestamp.expires = datetime(1970, 1, 1)
+
+        timestamp = self.modify_metadata("timestamp", timestamp_expired_modifier)
         with self.assertRaises(exceptions.ExpiredMetadataError):
-            self.trusted_set.update_timestamp(timestamp.to_bytes())
+            self.trusted_set.update_timestamp(timestamp)
 
 
     def test_update_snapshot_cannot_verify_snapshot_with_threshold(self):
-        def hashes_length_modifier(timestamp: Timestamp):
-            timestamp.meta["snapshot.json"].hashes = None
-            timestamp.meta["snapshot.json"].length = None
-
-        timestamp = self.modify_metadata("timestamp", hashes_length_modifier)
-        self._root_updated_and_update_timestamp(timestamp)
+        self._root_updated_and_update_timestamp(self.metadata["timestamp"])
         snapshot = Metadata.from_bytes(self.metadata["snapshot"])
         snapshot.signatures.clear()
         with self.assertRaises(exceptions.UnsignedMetadataError):
             self.trusted_set.update_snapshot(snapshot.to_bytes())
 
     def test_update_snapshot_version_different_timestamp_snapshot_version(self):
-        def hashes_length_version_modifier(timestamp: Timestamp):
-            timestamp.meta["snapshot.json"].hashes = None
-            timestamp.meta["snapshot.json"].length = None
+        def timestamp_version_modifier(timestamp: Timestamp) -> None:
             timestamp.meta["snapshot.json"].version = 2
 
-        timestamp = self.modify_metadata(
-            "timestamp", hashes_length_version_modifier
-        )
+        timestamp = self.modify_metadata("timestamp", timestamp_version_modifier)
         self._root_updated_and_update_timestamp(timestamp)
         # new_snapshot.version != trusted timestamp.meta["snapshot"].version
-        snapshot = Metadata.from_bytes(self.metadata["snapshot"])
-        snapshot.signed.version = 3
-        snapshot.sign(self.keystore["snapshot"])
+        def snapshot_version_modifier(snapshot: Snapshot) -> None:
+            snapshot.version = 3
+
+        snapshot = self.modify_metadata("snapshot", snapshot_version_modifier)
         with self.assertRaises(exceptions.BadVersionNumberError):
-            self.trusted_set.update_snapshot(snapshot.to_bytes())
+            self.trusted_set.update_snapshot(snapshot)
 
     def test_update_snapshot_after_successful_update_new_snapshot_no_meta(self):
-        def hashes_length_modifier(timestamp: Timestamp):
-            timestamp.meta["snapshot.json"].hashes = None
-            timestamp.meta["snapshot.json"].length = None
-
-        timestamp = self.modify_metadata("timestamp", hashes_length_modifier)
-        self._update_all_besides_targets(timestamp)
+        self._update_all_besides_targets(self.metadata["timestamp"])
         # Test removing a meta_file in new_snapshot compared to the old snapshot
-        snapshot = Metadata.from_bytes(self.metadata["snapshot"])
-        snapshot.signed.meta = {}
-        snapshot.sign(self.keystore["snapshot"])
+        def no_meta_modifier(snapshot: Snapshot) -> None:
+            snapshot.meta = {}
+
+        snapshot = self.modify_metadata("snapshot", no_meta_modifier)
         with self.assertRaises(exceptions.RepositoryError):
-            self.trusted_set.update_snapshot(snapshot.to_bytes())
+            self.trusted_set.update_snapshot(snapshot)
 
     def test_update_snapshot_after_succesfull_update_new_snapshot_meta_version_different(self):
-        def hashes_length_modifier(timestamp: Timestamp):
-            timestamp.meta["snapshot.json"].hashes = None
-            timestamp.meta["snapshot.json"].length = None
-
-        timestamp = self.modify_metadata("timestamp", hashes_length_modifier)
-        self._root_updated_and_update_timestamp(timestamp)
+        self._root_updated_and_update_timestamp(self.metadata["timestamp"])
         # snapshot.meta["project1"].version != new_snapshot.meta["project1"].version
-        snapshot = Metadata.from_bytes(self.metadata["snapshot"])
-        for metafile_path in snapshot.signed.meta.keys():
-            snapshot.signed.meta[metafile_path].version += 1
-        snapshot.sign(self.keystore["snapshot"])
-        self.trusted_set.update_snapshot(snapshot.to_bytes())
+        def version_meta_modifier(snapshot: Snapshot) -> None:
+            for metafile_path in snapshot.meta.keys():
+                snapshot.meta[metafile_path].version += 1
+
+        snapshot = self.modify_metadata("snapshot", version_meta_modifier)
+        self.trusted_set.update_snapshot(snapshot)
         with self.assertRaises(exceptions.BadVersionNumberError):
             self.trusted_set.update_snapshot(self.metadata["snapshot"])
 
     def test_update_snapshot_expired_new_snapshot(self):
-        def hashes_length_modifier(timestamp: Timestamp):
-            timestamp.meta["snapshot.json"].hashes = None
-            timestamp.meta["snapshot.json"].length = None
-
-        timestamp = self.modify_metadata("timestamp", hashes_length_modifier)
-        self._root_updated_and_update_timestamp(timestamp)
+        self._root_updated_and_update_timestamp(self.metadata["timestamp"])
         # new_snapshot has expired
-        snapshot = Metadata.from_bytes(self.metadata["snapshot"])
-        snapshot.signed.expires = datetime(1970, 1, 1)
-        snapshot.sign(self.keystore["snapshot"])
+        def snapshot_expired_modifier(snapshot: Snapshot) -> None:
+            snapshot.expires = datetime(1970, 1, 1)
+
+        snapshot = self.modify_metadata("snapshot", snapshot_expired_modifier)
         with self.assertRaises(exceptions.ExpiredMetadataError):
-            self.trusted_set.update_snapshot(snapshot.to_bytes())
+            self.trusted_set.update_snapshot(snapshot)
 
 
     def test_update_targets_no_meta_in_snapshot(self):
-        def hashes_length_modifier(timestamp: Timestamp):
-            timestamp.meta["snapshot.json"].hashes = None
-            timestamp.meta["snapshot.json"].length = None
-
-        timestamp = self.modify_metadata("timestamp", hashes_length_modifier)
-        def no_meta_modifier(snapshot: Snapshot):
+        def no_meta_modifier(snapshot: Snapshot) -> None:
             snapshot.meta = {}
 
         snapshot = self.modify_metadata("snapshot", no_meta_modifier)
-        self._update_all_besides_targets(timestamp, snapshot)
+        self._update_all_besides_targets(self.metadata["timestamp"], snapshot)
         # remove meta information with information about targets from snapshot
         with self.assertRaises(exceptions.RepositoryError):
             self.trusted_set.update_targets(self.metadata["targets"])
 
     def test_update_targets_hash_different_than_snapshot_meta_hash(self):
-        def hashes_length_modifier(timestamp: Timestamp):
-            timestamp.meta["snapshot.json"].hashes = None
-            timestamp.meta["snapshot.json"].length = None
-
-        timestamp = self.modify_metadata("timestamp", hashes_length_modifier)
-        def meta_length_modifier(snapshot: Snapshot):
+        def meta_length_modifier(snapshot: Snapshot) -> None:
             for metafile_path in snapshot.meta:
                 snapshot.meta[metafile_path] = MetaFile(version=1, length=1)
 
         snapshot = self.modify_metadata("snapshot", meta_length_modifier)
-        self._update_all_besides_targets(timestamp, snapshot)
+        self._update_all_besides_targets(self.metadata["timestamp"], snapshot)
         # observed_hash != stored hash in snapshot meta for targets
         with self.assertRaises(exceptions.RepositoryError):
             self.trusted_set.update_targets(self.metadata["targets"])
 
     def test_update_targets_version_different_snapshot_meta_version(self):
-        def hashes_length_modifier(timestamp: Timestamp):
-            timestamp.meta["snapshot.json"].hashes = None
-            timestamp.meta["snapshot.json"].length = None
-
-        timestamp = self.modify_metadata("timestamp", hashes_length_modifier)
-        def meta_modifier(snapshot: Snapshot):
+        def meta_modifier(snapshot: Snapshot) -> None:
             for metafile_path in snapshot.meta:
                 snapshot.meta[metafile_path] = MetaFile(version=2)
 
         snapshot = self.modify_metadata("snapshot", meta_modifier)
-        self._update_all_besides_targets(timestamp, snapshot)
+        self._update_all_besides_targets(self.metadata["timestamp"], snapshot)
         # new_delegate.signed.version != meta.version stored in snapshot
         with self.assertRaises(exceptions.BadVersionNumberError):
             self.trusted_set.update_targets(self.metadata["targets"])
 
     def test_update_targets_expired_new_target(self):
         self._update_all_besides_targets()
         # new_delegated_target has expired
-        targets = Metadata.from_bytes(self.metadata["targets"])
-        targets.signed.expires = datetime(1970, 1, 1)
-        targets.sign(self.keystore["targets"])
+        def target_expired_modifier(target: Targets) -> None:
+            target.expires = datetime(1970, 1, 1)
+
+        targets = self.modify_metadata("targets", target_expired_modifier)
         with self.assertRaises(exceptions.ExpiredMetadataError):
-            self.trusted_set.update_targets(targets.to_bytes())
+            self.trusted_set.update_targets(targets)
 
     # TODO test updating over initial metadata (new keys, newer timestamp, etc)