Ngclient api polish

tests/test_updater_ng.py

@@ -16,7 +16,7 @@
 import tuf.unittest_toolbox as unittest_toolbox
 
 from tests import utils
-from tuf.api.metadata import Metadata
+from tuf.api.metadata import Metadata, TargetFile
 from tuf import exceptions, ngclient
 from securesystemslib.signer import SSlibSigner
 from securesystemslib.interface import import_rsa_privatekey_from_file
@@ -111,11 +111,14 @@ def setUp(self):
 
         self.metadata_url = f"{url_prefix}/metadata/"
         self.targets_url = f"{url_prefix}/targets/"
-        self.destination_directory = self.make_temp_directory()
+        self.dl_dir = self.make_temp_directory()
         # Creating a repository instance.  The test cases will use this client
         # updater to refresh metadata, fetch target files, etc.
-        self.repository_updater = ngclient.Updater(
-            self.client_directory, self.metadata_url, self.targets_url
+        self.updater = ngclient.Updater(
+            repository_dir=self.client_directory,
+            metadata_base_url=self.metadata_url,
+            target_dir=self.dl_dir,
+            target_base_url=self.targets_url
         )
 
     def tearDown(self):
@@ -187,53 +190,40 @@ def consistent_snapshot_modifier(root):
         self._modify_repository_root(
             consistent_snapshot_modifier, bump_version=True
         )
-        self.repository_updater = ngclient.Updater(
-            self.client_directory, self.metadata_url, self.targets_url
+        updater = ngclient.Updater(
+            self.client_directory, self.metadata_url, self.dl_dir, self.targets_url
         )
 
         # All metadata is in local directory already
-        self.repository_updater.refresh()
+        updater.refresh()
         # Make sure that consistent snapshot is enabled
         self.assertTrue(
-            self.repository_updater._trusted_set.root.signed.consistent_snapshot
-        )
-        # Get targetinfo for "file1.txt" listed in targets
-        targetinfo1 = self.repository_updater.get_one_valid_targetinfo(
-            "file1.txt"
-        )
-        # Get targetinfo for "file3.txt" listed in the delegated role1
-        targetinfo3 = self.repository_updater.get_one_valid_targetinfo(
-            "file3.txt"
+            updater._trusted_set.root.signed.consistent_snapshot
         )
 
+        # Get targetinfos, assert cache does not contain the files
+        info1 = updater.get_targetinfo("file1.txt")
+        info3 = updater.get_targetinfo("file3.txt")
+        self.assertIsNone(updater.find_cached_target(info1))
+        self.assertIsNone(updater.find_cached_target(info3))
+
         # Create consistent targets with file path HASH.FILENAME.EXT
-        target1_hash = list(targetinfo1.hashes.values())[0]
-        target3_hash = list(targetinfo3.hashes.values())[0]
+        target1_hash = list(info1.hashes.values())[0]
+        target3_hash = list(info3.hashes.values())[0]
         self._create_consistent_target("file1.txt", target1_hash)
         self._create_consistent_target("file3.txt", target3_hash)
 
-        updated_targets = self.repository_updater.updated_targets(
-            [targetinfo1, targetinfo3], self.destination_directory
-        )
-
-        self.assertListEqual(updated_targets, [targetinfo1, targetinfo3])
-        self.repository_updater.download_target(
-            targetinfo1, self.destination_directory
-        )
-        updated_targets = self.repository_updater.updated_targets(
-            updated_targets, self.destination_directory
-        )
-
-        self.assertListEqual(updated_targets, [targetinfo3])
-
-        self.repository_updater.download_target(
-            targetinfo3, self.destination_directory
-        )
-        updated_targets = self.repository_updater.updated_targets(
-            updated_targets, self.destination_directory
-        )
+        # Download files, assert that cache has correct files
+        updater.download_target(info1)
+        path = updater.find_cached_target(info1)
+        self.assertEqual(path, os.path.join(self.dl_dir, info1.path))
+        self.assertIsNone(updater.find_cached_target(info3))
 
-        self.assertListEqual(updated_targets, [])
+        updater.download_target(info3)
+        path = updater.find_cached_target(info1)
+        self.assertEqual(path, os.path.join(self.dl_dir, info1.path))
+        path = updater.find_cached_target(info3)
+        self.assertEqual(path, os.path.join(self.dl_dir, info3.path))
 
     def test_refresh_and_download(self):
         # Test refresh without consistent targets - targets without hash prefixes.
@@ -244,50 +234,33 @@ def test_refresh_and_download(self):
         os.remove(os.path.join(self.client_directory, "1.root.json"))
 
         # top-level metadata is in local directory already
-        self.repository_updater.refresh()
+        self.updater.refresh()
         self._assert_files(["root", "snapshot", "targets", "timestamp"])
 
-        # Get targetinfo for 'file1.txt' listed in targets
-        targetinfo1 = self.repository_updater.get_one_valid_targetinfo(
-            "file1.txt"
-        )
+        # Get targetinfos, assert that cache does not contain files
+        info1 = self.updater.get_targetinfo("file1.txt")
         self._assert_files(["root", "snapshot", "targets", "timestamp"])
+
         # Get targetinfo for 'file3.txt' listed in the delegated role1
-        targetinfo3 = self.repository_updater.get_one_valid_targetinfo(
-            "file3.txt"
-        )
+        info3 = self.updater.get_targetinfo("file3.txt")
         expected_files = ["role1", "root", "snapshot", "targets", "timestamp"]
         self._assert_files(expected_files)
+        self.assertIsNone(self.updater.find_cached_target(info1))
+        self.assertIsNone(self.updater.find_cached_target(info3))
 
-        updated_targets = self.repository_updater.updated_targets(
-            [targetinfo1, targetinfo3], self.destination_directory
-        )
-
-        self.assertListEqual(updated_targets, [targetinfo1, targetinfo3])
-
-        self.repository_updater.download_target(
-            targetinfo1, self.destination_directory
-        )
-        updated_targets = self.repository_updater.updated_targets(
-            updated_targets, self.destination_directory
+        # Download files, assert that cache has correct files
+        self.updater.download_target(info1)
+        path = self.updater.find_cached_target(info1)
+        self.assertEqual(path, os.path.join(self.dl_dir, info1.path))
+        self.assertIsNone(
+            self.updater.find_cached_target(info3)
         )
 
-        self.assertListEqual(updated_targets, [targetinfo3])
-
-        # Check that duplicates are excluded
-        updated_targets = self.repository_updater.updated_targets(
-            [targetinfo3, targetinfo3], self.destination_directory
-        )
-        self.assertListEqual(updated_targets, [targetinfo3])
-
-        self.repository_updater.download_target(
-            targetinfo3, self.destination_directory
-        )
-        updated_targets = self.repository_updater.updated_targets(
-            updated_targets, self.destination_directory
-        )
-
-        self.assertListEqual(updated_targets, [])
+        self.updater.download_target(info3)
+        path = self.updater.find_cached_target(info1)
+        self.assertEqual(path, os.path.join(self.dl_dir, info1.path))
+        path = self.updater.find_cached_target(info3)
+        self.assertEqual(path, os.path.join(self.dl_dir, info3.path))
 
     def test_refresh_with_only_local_root(self):
         os.remove(os.path.join(self.client_directory, "timestamp.json"))
@@ -298,70 +271,61 @@ def test_refresh_with_only_local_root(self):
         os.remove(os.path.join(self.client_directory, "1.root.json"))
         self._assert_files(["root"])
 
-        self.repository_updater.refresh()
+        self.updater.refresh()
         self._assert_files(["root", "snapshot", "targets", "timestamp"])
 
         # Get targetinfo for 'file3.txt' listed in the delegated role1
-        targetinfo3 = self.repository_updater.get_one_valid_targetinfo(
-            "file3.txt"
-        )
+        targetinfo3 = self.updater.get_targetinfo("file3.txt")
         expected_files = ["role1", "root", "snapshot", "targets", "timestamp"]
         self._assert_files(expected_files)
 
     def test_both_target_urls_not_set(self):
         # target_base_url = None and Updater._target_base_url = None
-        self.repository_updater = ngclient.Updater(
-            self.client_directory, self.metadata_url
-        )
+        updater = ngclient.Updater(self.client_directory, self.metadata_url, self.dl_dir)
+        info = TargetFile(1, {"sha256": ""}, "targetpath")
         with self.assertRaises(ValueError):
-            self.repository_updater.download_target(
-                [], self.destination_directory
-            )
+            updater.download_target(info)
+
+    def test_no_target_dir_no_filepath(self):
+        # filepath = None and Updater.target_dir = None
+        updater = ngclient.Updater(self.client_directory, self.metadata_url)
+        info = TargetFile(1, {"sha256": ""}, "targetpath")
+        with self.assertRaises(ValueError):
+            updater.find_cached_target(info)
+        with self.assertRaises(ValueError):
+            updater.download_target(info)
 
     def test_external_targets_url(self):
-        self.repository_updater.refresh()
-        targetinfo = self.repository_updater.get_one_valid_targetinfo(
-            "file1.txt"
-        )
-        self.repository_updater.download_target(
-            targetinfo, self.destination_directory, self.targets_url
-        )
+        self.updater.refresh()
+        info = self.updater.get_targetinfo("file1.txt")
+
+        self.updater.download_target(info, target_base_url=self.targets_url)
 
     def test_length_hash_mismatch(self):
-        self.repository_updater.refresh()
-        targetinfo = self.repository_updater.get_one_valid_targetinfo(
-            "file1.txt"
-        )
+        self.updater.refresh()
+        targetinfo = self.updater.get_targetinfo("file1.txt")
+
         length = targetinfo.length
         with self.assertRaises(exceptions.RepositoryError):
             targetinfo.length = 44
-            self.repository_updater.download_target(
-                targetinfo, self.destination_directory
-            )
+            self.updater.download_target(targetinfo)
 
         with self.assertRaises(exceptions.RepositoryError):
             targetinfo.length = length
             targetinfo.hashes = {"sha256": "abcd"}
-            self.repository_updater.download_target(
-                targetinfo, self.destination_directory
-            )
+            self.updater.download_target(targetinfo)
 
     def test_updating_root(self):
         # Bump root version, resign and refresh
         self._modify_repository_root(lambda root: None, bump_version=True)
-        self.repository_updater.refresh()
-        self.assertEqual(
-            self.repository_updater._trusted_set.root.signed.version, 2
-        )
+        self.updater.refresh()
+        self.assertEqual(self.updater._trusted_set.root.signed.version, 2)
 
     def test_missing_targetinfo(self):
-        self.repository_updater.refresh()
+        self.updater.refresh()
 
         # Get targetinfo for non-existing file
-        targetinfo = self.repository_updater.get_one_valid_targetinfo(
-            "file33.txt"
-        )
-        self.assertIsNone(targetinfo)
+        self.assertIsNone(self.updater.get_targetinfo("file33.txt"))
 
 
 if __name__ == "__main__":

tests/test_updater_with_simulator.py

@@ -54,6 +54,7 @@ def _run_refresh(self) -> Updater:
         updater = Updater(
             self.metadata_dir,
             "https://example.com/metadata/",
+            self.targets_dir,
             "https://example.com/targets/",
             self.sim
         )
@@ -90,9 +91,13 @@ def test_refresh(self):
     @utils.run_sub_tests_with_dataset(targets)
     def test_targets(self, test_case_data: Tuple[str, bytes, str]):
         targetpath, content, encoded_path = test_case_data
-        # target does not exist yet
+        path = os.path.join(self.targets_dir, encoded_path)
+
         updater = self._run_refresh()
-        self.assertIsNone(updater.get_one_valid_targetinfo(targetpath))
+        # target does not exist yet, configuration is what we expect
+        self.assertIsNone(updater.get_targetinfo(targetpath))
+        self.assertTrue(self.sim.root.consistent_snapshot)
+        self.assertTrue(updater.config.prefix_targets_with_hash)
 
         # Add targets to repository
         self.sim.targets.version += 1
@@ -101,30 +106,25 @@ def test_targets(self, test_case_data: Tuple[str, bytes, str]):
 
         updater = self._run_refresh()
         # target now exists, is not in cache yet
-        file_info = updater.get_one_valid_targetinfo(targetpath)
-        self.assertIsNotNone(file_info)
-        self.assertEqual(
-            updater.updated_targets([file_info], self.targets_dir),
-            [file_info]
-        )
+        info = updater.get_targetinfo(targetpath)
+        self.assertIsNotNone(info)
+        # Test without and with explicit local filepath
+        self.assertIsNone(updater.find_cached_target(info))
+        self.assertIsNone(updater.find_cached_target(info, path))
 
-        # Assert consistent_snapshot is True and downloaded targets have prefix.
-        self.assertTrue(self.sim.root.consistent_snapshot)
-        self.assertTrue(updater.config.prefix_targets_with_hash)
         # download target, assert it is in cache and content is correct
-        local_path = updater.download_target(file_info, self.targets_dir)
-        self.assertEqual(
-            updater.updated_targets([file_info], self.targets_dir), []
-        )
-        self.assertTrue(local_path.startswith(self.targets_dir))
-        with open(local_path, "rb") as f:
-            self.assertEqual(f.read(), content)
-
-        # Assert that the targetpath was URL encoded as expected.
-        encoded_absolute_path = os.path.join(self.targets_dir, encoded_path)
-        self.assertEqual(local_path, encoded_absolute_path)
+        self.assertEqual(path, updater.download_target(info))
+        self.assertEqual(path, updater.find_cached_target(info))
+        self.assertEqual(path, updater.find_cached_target(info, path))
 
+        with open(path, "rb") as f:
+            self.assertEqual(f.read(), content)
 
+        # download using explicit filepath as well
+        os.remove(path)
+        self.assertEqual(path, updater.download_target(info, path))
+        self.assertEqual(path, updater.find_cached_target(info))
+        self.assertEqual(path, updater.find_cached_target(info, path))
 
     def test_fishy_rolenames(self):
         roles_to_filenames = {
@@ -145,7 +145,7 @@ def test_fishy_rolenames(self):
         updater = self._run_refresh()
 
         # trigger updater to fetch the delegated metadata, check filenames
-        updater.get_one_valid_targetinfo("anything")
+        updater.get_targetinfo("anything")
         local_metadata = os.listdir(self.metadata_dir)
         for fname in roles_to_filenames.values():
             self.assertTrue(fname in local_metadata)