MetadataBundle: Don't do any file IO

Remove file IO from MetadataBundle:
* This make the bundle API very clear and easy to understand
* This means caller must now read from and persist data to disk
  but initial prototypes suggest this won't make Updater too
  complex

This change is something we can still back out from if it turns out to
be the wrong decision: the file-persisting MetadataBundle has been tested
and works fine.

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

Author: Jussi Kukkonen

tests/test_metadata_bundle.py

@@ -15,166 +15,108 @@
 logger = logging.getLogger(__name__)
 
 class TestMetadataBundle(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        cls.temp_dir = tempfile.mkdtemp(dir=os.getcwd())
-
-    @classmethod
-    def tearDownClass(cls):
-        shutil.rmtree(cls.temp_dir)
-
-    def setUp(self):
-        # copy metadata to "local repo"
-        shutil.copytree(
-            os.path.join(os.getcwd(), 'repository_data', 'repository', 'metadata'),
-            self.temp_dir,
-            dirs_exist_ok=True
-        )
-
-    def test_local_load(self):
-
-        # test loading all local metadata succesfully
-        bundle = MetadataBundle(self.temp_dir)
-        bundle.root_update_finished()
-        bundle.load_local_timestamp()
-        bundle.load_local_snapshot()
-        bundle.load_local_targets()
-        bundle.load_local_delegated_targets('role1','targets')
-        bundle.load_local_delegated_targets('role2','role1')
-
-        # Make sure loading metadata without its "dependencies" fails
-        bundle = MetadataBundle(self.temp_dir)
-
-        with self.assertRaises(RuntimeError):
-            bundle.load_local_timestamp()
-        bundle.root_update_finished()
-        with self.assertRaises(RuntimeError):
-            bundle.load_local_snapshot()
-        bundle.load_local_timestamp()
-        with self.assertRaises(RuntimeError):
-            bundle.load_local_targets()
-        bundle.load_local_snapshot()
-        with self.assertRaises(RuntimeError):
-            bundle.load_local_delegated_targets('role1','targets')
-        bundle.load_local_targets()
-        with self.assertRaises(RuntimeError):
-            bundle.load_local_delegated_targets('role2','role1')
-        bundle.load_local_delegated_targets('role1','targets')
-        bundle.load_local_delegated_targets('role2','role1')
 
     def test_update(self):
-        remote_dir = os.path.join(os.getcwd(), 'repository_data', 'repository', 'metadata')
-
-        # remove all but root.json from local repo
-        os.remove(os.path.join(self.temp_dir, "timestamp.json"))
-        os.remove(os.path.join(self.temp_dir, "snapshot.json"))
-        os.remove(os.path.join(self.temp_dir, "targets.json"))
-        os.remove(os.path.join(self.temp_dir, "role1.json"))
-        os.remove(os.path.join(self.temp_dir, "role2.json"))
+        repo_dir = os.path.join(os.getcwd(), 'repository_data', 'repository', 'metadata')
 
-        bundle = MetadataBundle(self.temp_dir)
+        with open(os.path.join(repo_dir, "root.json"), "rb") as f:
+            bundle = MetadataBundle(f.read())
         bundle.root_update_finished()
 
-        # test local load failure, then updating metadata succesfully
-        with self.assertRaises(exceptions.RepositoryError):
-            bundle.load_local_timestamp()
-        with open(os.path.join(remote_dir, "timestamp.json"), "rb") as f:
+        with open(os.path.join(repo_dir, "timestamp.json"), "rb") as f:
             bundle.update_timestamp(f.read())
-        with open(os.path.join(remote_dir, "snapshot.json"), "rb") as f:
+        with open(os.path.join(repo_dir, "snapshot.json"), "rb") as f:
             bundle.update_snapshot(f.read())
-        with open(os.path.join(remote_dir, "targets.json"), "rb") as f:
+        with open(os.path.join(repo_dir, "targets.json"), "rb") as f:
             bundle.update_targets(f.read())
-        with open(os.path.join(remote_dir, "role1.json"), "rb") as f:
+        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(remote_dir, "role2.json"), "rb") as f:
+        with open(os.path.join(repo_dir, "role2.json"), "rb") as f:
             bundle.update_delegated_targets(f.read(), "role2", "role1")
 
-        # test loading the metadata (that should now be locally available)
-        bundle = MetadataBundle(self.temp_dir)
-        bundle.root_update_finished()
-        bundle.load_local_timestamp()
-        bundle.load_local_snapshot()
-        bundle.load_local_targets()
-        bundle.load_local_delegated_targets('role1','targets')
-        bundle.load_local_delegated_targets('role2','role1')
-
-        # TODO test loading one version, then updating to new versions of each metadata
+    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()
 
-    def test_local_load_with_invalid_data(self):
-        # Test root and one of the top-level metadata files
+        bundle = MetadataBundle(data["root"])
 
-        with tempfile.TemporaryDirectory() as tempdir:
-            # Missing root.json
-            with self.assertRaises(exceptions.RepositoryError):
-                MetadataBundle(tempdir)
+        # Update timestamp before root is finished
+        with self.assertRaises(RuntimeError):
+            bundle.update_timestamp(data["timestamp"])
 
-            # root.json not a json file at all
-            with open(os.path.join(tempdir, "root.json"), "w") as f:
-                f.write("")
-            with self.assertRaises(exceptions.RepositoryError):
-                MetadataBundle(tempdir)
+        bundle.root_update_finished()
+        with self.assertRaises(RuntimeError):
+            bundle.root_update_finished()
 
-            # root.json does not validate
-            md = Metadata.from_file(os.path.join(self.temp_dir, "root.json"))
-            md.signed.version += 1
-            md.to_file(os.path.join(tempdir, "root.json"))
-            with self.assertRaises(exceptions.RepositoryError):
-                MetadataBundle(tempdir)
+        # Update snapshot before timestamp
+        with self.assertRaises(RuntimeError):
+            bundle.update_snapshot(data["snapshot"])
 
-            md.signed.version -= 1
-            md.to_file(os.path.join(tempdir, "root.json"))
-            bundle = MetadataBundle(tempdir)
-            bundle.root_update_finished()
+        bundle.update_timestamp(data["timestamp"])
 
-            # Missing timestamp.json
-            with self.assertRaises(exceptions.RepositoryError):
-                bundle.load_local_timestamp()
+        # Update targets before snapshot
+        with self.assertRaises(RuntimeError):
+            bundle.update_targets(data["targets"])
 
-            # timestamp not a json file at all
-            with open(os.path.join(tempdir, "timestamp.json"), "w") as f:
-                f.write("")
-            with self.assertRaises(exceptions.RepositoryError):
-                bundle.load_local_timestamp()
+        bundle.update_snapshot(data["snapshot"])
 
-            # timestamp does not validate
-            md = Metadata.from_file(os.path.join(self.temp_dir, "timestamp.json"))
-            md.signed.version += 1
-            md.to_file(os.path.join(tempdir, "timestamp.json"))
-            with self.assertRaises(exceptions.RepositoryError):
-                bundle.load_local_timestamp()
+        #update timestamp after snapshot
+        with self.assertRaises(RuntimeError):
+            bundle.update_timestamp(data["timestamp"])
 
-            md.signed.version -= 1
-            md.to_file(os.path.join(tempdir, "timestamp.json"))
-            bundle.load_local_timestamp()
+        # Update delegated targets before targets
+        with self.assertRaises(RuntimeError):
+            bundle.update_delegated_targets(data["role1"], "role1", "targets")
 
-    def test_update_with_invalid_data(self):
-        # Test on of the top level metadata files
+        bundle.update_targets(data["targets"])
+        bundle.update_delegated_targets(data["role1"], "role1", "targets")
 
-        timestamp_md = Metadata.from_file(os.path.join(self.temp_dir, "timestamp.json"))
+    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()
 
-        # remove all but root.json from local repo
-        os.remove(os.path.join(self.temp_dir, "timestamp.json"))
-        os.remove(os.path.join(self.temp_dir, "snapshot.json"))
-        os.remove(os.path.join(self.temp_dir, "targets.json"))
-        os.remove(os.path.join(self.temp_dir, "role1.json"))
-        os.remove(os.path.join(self.temp_dir, "role2.json"))
+        # 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(self.temp_dir)
+        bundle = MetadataBundle(data["root"])
         bundle.root_update_finished()
 
-        # timestamp not a json file at all
-        with self.assertRaises(exceptions.RepositoryError):
-            bundle.update_timestamp(b"")
+        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)
 
-        # timestamp does not validate
-        timestamp_md.signed.version += 1
-        data = timestamp_md.to_dict()
-        with self.assertRaises(exceptions.RepositoryError):
-            bundle.update_timestamp(json.dumps(data).encode())
 
-        timestamp_md.signed.version -= 1
-        data = timestamp_md.to_dict()
-        bundle.update_timestamp(json.dumps(data).encode())
+    # TODO test updating over initial metadata (new keys, newer timestamp, etc)
+    # TODO test the actual specification checks
 
 
 if __name__ == '__main__':