diff --git a/tests/test_metadata_bundle.py b/tests/test_metadata_bundle.py index be30e6ba..b566a7a2 100644 --- a/tests/test_metadata_bundle.py +++ b/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) diff --git a/tuf/client_rework/metadata_bundle.py b/tuf/client_rework/metadata_bundle.py index 3ea73809..91f0b39d 100644 --- a/tuf/client_rework/metadata_bundle.py +++ b/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.