MetadataBundle: Save original files on disk

Don't use the serialized format as that won't match any hashes in
"meta".

Add basic tests for updating metadata.

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

Author: Jussi Kukkonen

tests/test_metadata_bundle.py

@@ -1,6 +1,8 @@
 import logging
 import os
+import shutil
 import sys
+import tempfile
 import unittest
 
 from tuf.api import metadata
@@ -11,11 +13,26 @@
 logger = logging.getLogger(__name__)
 
 class TestMetadataBundle(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.temporary_directory = tempfile.mkdtemp(dir=os.getcwd())
+
+    @classmethod
+    def tearDownClass(cls):
+        shutil.rmtree(cls.temporary_directory)
+
+    def setUp(self):
+        # copy metadata to "local repo"
+        shutil.copytree(
+            os.path.join(os.getcwd(), 'repository_data', 'repository', 'metadata'),
+            self.temporary_directory,
+            dirs_exist_ok=True
+        )
+
     def test_local_load(self):
-        repo_dir = os.path.join(os.getcwd(), 'repository_data', 'repository', 'metadata')
 
         # test loading all local metadata succesfully
-        bundle = MetadataBundle(repo_dir)
+        bundle = MetadataBundle(self.temporary_directory)
         bundle.root_update_finished()
         self.assertTrue(bundle.load_local_timestamp())
         self.assertTrue(bundle.load_local_snapshot())
@@ -24,7 +41,7 @@ def test_local_load(self):
         self.assertTrue(bundle.load_local_delegated_targets('role2','role1'))
 
         # Make sure loading metadata without its "dependencies" fails
-        bundle = MetadataBundle(repo_dir)
+        bundle = MetadataBundle(self.temporary_directory)
 
         with self.assertRaises(RuntimeError):
             bundle.load_local_timestamp()
@@ -43,6 +60,42 @@ def test_local_load(self):
         self.assertTrue(bundle.load_local_delegated_targets('role1','targets'))
         self.assertTrue(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.temporary_directory, "timestamp.json"))
+        os.remove(os.path.join(self.temporary_directory, "snapshot.json"))
+        os.remove(os.path.join(self.temporary_directory, "targets.json"))
+        os.remove(os.path.join(self.temporary_directory, "role1.json"))
+        os.remove(os.path.join(self.temporary_directory, "role2.json"))
+
+        # test updating metadata succesfully
+        bundle = MetadataBundle(self.temporary_directory)
+        bundle.root_update_finished()
+
+        with open(os.path.join(remote_dir, "timestamp.json"), "rb") as f:
+            bundle.update_timestamp(f.read())
+        with open(os.path.join(remote_dir, "snapshot.json"), "rb") as f:
+            bundle.update_snapshot(f.read())
+        with open(os.path.join(remote_dir, "targets.json"), "rb") as f:
+            bundle.update_targets(f.read())
+        with open(os.path.join(remote_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:
+            bundle.update_delegated_targets(f.read(), "role2", "role1")
+
+        # test loading the metadata (that should now be locally available)
+        bundle = MetadataBundle(self.temporary_directory)
+        bundle.root_update_finished()
+        self.assertTrue(bundle.load_local_timestamp())
+        self.assertTrue(bundle.load_local_snapshot())
+        self.assertTrue(bundle.load_local_targets())
+        self.assertTrue(bundle.load_local_delegated_targets('role1','targets'))
+        self.assertTrue(bundle.load_local_delegated_targets('role2','role1'))
+
+        # TODO test loading one version, then updating to new versions of each metadata
+
 
 if __name__ == '__main__':
   utils.configure_test_logging(sys.argv)

tuf/client_rework/metadata_bundle.py

@@ -199,7 +199,8 @@ def update_root(self, data: bytes):
         logger.debug("Updating root")
 
         self._load_intermediate_root(data)
-        self.root.to_file(os.path.join(self._path, "root.json"))
+        with open(os.path.join(self._path, "root.json"), "wb") as f:
+            f.write(data)
 
     def root_update_finished(self):
         """Mark root metadata as final."""
@@ -232,7 +233,8 @@ def update_timestamp(self, data: bytes):
         logger.debug("Updating timestamp")
 
         self._load_timestamp(data)
-        self.timestamp.to_file(os.path.join(self._path, "timestamp.json"))
+        with open(os.path.join(self._path, "timestamp.json"), "wb") as f:
+            f.write(data)
 
     def load_local_snapshot(self) -> bool:
         """Load cached snapshot metadata from local storage.
@@ -253,7 +255,8 @@ def update_snapshot(self, data: bytes):
         logger.debug("Updating snapshot")
 
         self._load_snapshot(data)
-        self.snapshot.to_file(os.path.join(self._path, "snapshot.json"))
+        with open(os.path.join(self._path, "snapshot.json"), "wb") as f:
+            f.write(data)
 
     def load_local_targets(self) -> bool:
         """Load cached targets metadata from local storage.
@@ -298,7 +301,8 @@ def update_delegated_targets(
         logger.debug("Updating %s", role_name)
 
         self._load_delegated_targets(data, role_name, delegator_name)
-        self[role_name].to_file(os.path.join(self._path, f"{role_name}.json"))
+        with open(os.path.join(self._path, f"{role_name}.json"), "wb") as f:
+            f.write(data)
 
     def _load_intermediate_root(self, data: bytes):
         """Verifies and loads 'data' as new root metadata.