experimental client: Add MetadataBundle

tests/test_api.py

@@ -450,21 +450,23 @@ def test_delegated_role_class(self):
             with self.assertRaises(ValueError):
                 DelegatedRole.from_dict(role.copy())
 
-            # Test creating DelegatedRole only with "path_hash_prefixes"
+            # Test creating DelegatedRole only with "path_hash_prefixes" (an empty one)
             del role["paths"]
-            DelegatedRole.from_dict(role.copy())
-            role["paths"] = "foo"
+            role["path_hash_prefixes"] = []
+            role_obj = DelegatedRole.from_dict(role.copy())
+            self.assertEqual(role_obj.to_dict(), role)
 
-            # Test creating DelegatedRole only with "paths"
+            # Test creating DelegatedRole only with "paths" (now an empty one)
             del role["path_hash_prefixes"]
-            DelegatedRole.from_dict(role.copy())
-            role["path_hash_prefixes"] = "foo"
+            role["paths"] = []
+            role_obj = DelegatedRole.from_dict(role.copy())
+            self.assertEqual(role_obj.to_dict(), role)
 
             # Test creating DelegatedRole without "paths" and
             # "path_hash_prefixes" set
             del role["paths"]
-            del role["path_hash_prefixes"]
-            DelegatedRole.from_dict(role)
+            role_obj = DelegatedRole.from_dict(role.copy())
+            self.assertEqual(role_obj.to_dict(), role)
 
 
     def test_delegation_class(self):
@@ -494,6 +496,21 @@ def test_delegation_class(self):
         delegations = Delegations.from_dict(copy.deepcopy(delegations_dict))
         self.assertEqual(delegations_dict, delegations.to_dict())
 
+        # empty keys and roles
+        delegations_dict = {"keys":{}, "roles":[]}
+        delegations = Delegations.from_dict(delegations_dict.copy())
+        self.assertEqual(delegations_dict, delegations.to_dict())
+
+        # Test some basic missing or broken input
+        invalid_delegations_dicts = [
+            {},
+            {"keys":None, "roles":None},
+            {"keys":{"foo":0}, "roles":[]},
+            {"keys":{}, "roles":["foo"]},
+        ]
+        for d in invalid_delegations_dicts:
+            with self.assertRaises((KeyError, AttributeError)):
+                Delegations.from_dict(d)
 
     def test_metadata_targets(self):
         targets_path = os.path.join(

tests/test_metadata_bundle.py

@@ -0,0 +1,124 @@
+import json
+import logging
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+
+from tuf import exceptions
+from tuf.api.metadata import Metadata
+from tuf.client_rework.metadata_bundle import MetadataBundle
+
+from tests import utils
+
+logger = logging.getLogger(__name__)
+
+class TestMetadataBundle(unittest.TestCase):
+
+    def test_update(self):
+        repo_dir = os.path.join(os.getcwd(), 'repository_data', 'repository', 'metadata')
+
+        with open(os.path.join(repo_dir, "root.json"), "rb") as f:
+            bundle = MetadataBundle(f.read())
+        bundle.root_update_finished()
+
+        with open(os.path.join(repo_dir, "timestamp.json"), "rb") as f:
+            bundle.update_timestamp(f.read())
+        with open(os.path.join(repo_dir, "snapshot.json"), "rb") as f:
+            bundle.update_snapshot(f.read())
+        with open(os.path.join(repo_dir, "targets.json"), "rb") as f:
+            bundle.update_targets(f.read())
+        with open(os.path.join(repo_dir, "role1.json"), "rb") as f:
+            bundle.update_delegated_targets(f.read(), "role1", "targets")
+        with open(os.path.join(repo_dir, "role2.json"), "rb") as f:
+            bundle.update_delegated_targets(f.read(), "role2", "role1")
+
+    def test_out_of_order_ops(self):
+        repo_dir = os.path.join(os.getcwd(), 'repository_data', 'repository', 'metadata')
+        data={}
+        for md in ["root", "timestamp", "snapshot", "targets", "role1"]:
+            with open(os.path.join(repo_dir, f"{md}.json"), "rb") as f:
+                data[md] = f.read()
+
+        bundle = MetadataBundle(data["root"])
+
+        # Update timestamp before root is finished
+        with self.assertRaises(RuntimeError):
+            bundle.update_timestamp(data["timestamp"])
+
+        bundle.root_update_finished()
+        with self.assertRaises(RuntimeError):
+            bundle.root_update_finished()
+
+        # Update snapshot before timestamp
+        with self.assertRaises(RuntimeError):
+            bundle.update_snapshot(data["snapshot"])
+
+        bundle.update_timestamp(data["timestamp"])
+
+        # Update targets before snapshot
+        with self.assertRaises(RuntimeError):
+            bundle.update_targets(data["targets"])
+
+        bundle.update_snapshot(data["snapshot"])
+
+        #update timestamp after snapshot
+        with self.assertRaises(RuntimeError):
+            bundle.update_timestamp(data["timestamp"])
+
+        # Update delegated targets before targets
+        with self.assertRaises(RuntimeError):
+            bundle.update_delegated_targets(data["role1"], "role1", "targets")
+
+        bundle.update_targets(data["targets"])
+        bundle.update_delegated_targets(data["role1"], "role1", "targets")
+
+    def test_update_with_invalid_json(self):
+        repo_dir = os.path.join(os.getcwd(), 'repository_data', 'repository', 'metadata')
+        data={}
+        for md in ["root", "timestamp", "snapshot", "targets", "role1"]:
+            with open(os.path.join(repo_dir, f"{md}.json"), "rb") as f:
+                data[md] = f.read()
+
+        # root.json not a json file at all
+        with self.assertRaises(exceptions.RepositoryError):
+            MetadataBundle(b"")
+        # root.json is invalid
+        root = Metadata.from_bytes(data["root"])
+        root.signed.version += 1
+        with self.assertRaises(exceptions.RepositoryError):
+            MetadataBundle(json.dumps(root.to_dict()).encode())
+
+        bundle = MetadataBundle(data["root"])
+        bundle.root_update_finished()
+
+        top_level_md = [
+            (data["timestamp"], bundle.update_timestamp),
+            (data["snapshot"], bundle.update_snapshot),
+            (data["targets"], bundle.update_targets),
+        ]
+        for metadata, update_func in top_level_md:
+            # metadata is not json
+            with self.assertRaises(exceptions.RepositoryError):
+                update_func(b"")
+            # metadata is invalid
+            md = Metadata.from_bytes(metadata)
+            md.signed.version += 1
+            with self.assertRaises(exceptions.RepositoryError):
+                update_func(json.dumps(md.to_dict()).encode())
+
+            # metadata is of wrong type
+            with self.assertRaises(exceptions.RepositoryError):
+                update_func(data["root"])
+
+            update_func(metadata)
+
+
+    # TODO test updating over initial metadata (new keys, newer timestamp, etc)
+    # TODO test the actual specification checks
+
+
+if __name__ == '__main__':
+  utils.configure_test_logging(sys.argv)
+  unittest.main()

tuf/api/metadata.py

@@ -770,7 +770,7 @@ def __init__(
         super().__init__(keyids, threshold, unrecognized_fields)
         self.name = name
         self.terminating = terminating
-        if paths and path_hash_prefixes:
+        if paths is not None and path_hash_prefixes is not None:
             raise ValueError(
                 "Only one of the attributes 'paths' and"
                 "'path_hash_prefixes' can be set!"
@@ -806,9 +806,9 @@ def to_dict(self) -> Dict[str, Any]:
             "terminating": self.terminating,
             **base_role_dict,
         }
-        if self.paths:
+        if self.paths is not None:
             res_dict["paths"] = self.paths
-        elif self.path_hash_prefixes:
+        elif self.path_hash_prefixes is not None:
             res_dict["path_hash_prefixes"] = self.path_hash_prefixes
         return res_dict
 
@@ -911,17 +911,20 @@ def from_dict(cls, targets_dict: Dict[str, Any]) -> "Targets":
         """Creates Targets object from its dict representation."""
         common_args = cls._common_fields_from_dict(targets_dict)
         targets = targets_dict.pop("targets")
-        delegations = targets_dict.pop("delegations", None)
-        if delegations:
-            delegations = Delegations.from_dict(delegations)
+        try:
+            delegations_dict = targets_dict.pop("delegations")
+        except KeyError:
+            delegations = None
+        else:
+            delegations = Delegations.from_dict(delegations_dict)
         # All fields left in the targets_dict are unrecognized.
         return cls(*common_args, targets, delegations, targets_dict)
 
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self."""
         targets_dict = self._common_fields_to_dict()
         targets_dict["targets"] = self.targets
-        if self.delegations:
+        if self.delegations is not None:
             targets_dict["delegations"] = self.delegations.to_dict()
         return targets_dict