Add TestTargetFileSearch class

Extend test_updater_delegation_graphs.py with tests
for targets metadata search.
- create a new test class TestTargetFileSearch which creates
  a single repository and pefrorms multiple file searches in
  subtests.
- group the common functionality in a base class TestDelegations.
- extend the data classes to accomodate for target_files.

Signed-off-by: Teodora Sechkova <[email protected]>

Author: sechkova

tests/test_updater_delegation_graphs.py

@@ -3,15 +3,15 @@
 # Copyright 2021, New York University and the TUF contributors
 # SPDX-License-Identifier: MIT OR Apache-2.0
 
-"""Test updating delegated targets roles with various
-delegation hierarchies"""
+"""Test updating delegated targets roles and searching for
+target files with various delegation graphs"""
 
 import os
 import sys
 import tempfile
 import unittest
 from dataclasses import astuple, dataclass, field
-from typing import Iterable, List, Optional
+from typing import Iterable, List, Optional, Tuple
 from unittest.mock import call, patch
 
 from tests import utils
@@ -36,18 +36,25 @@ class TestDelegation:
     path_hash_prefixes: Optional[List[str]] = None
 
 
+@dataclass
+class TestTarget:
+    rolename: str
+    content: bytes
+    targetpath: str
+
+
 @dataclass
 class DelegationsTestCase:
-    """Describes a delegations graph as a list of delegations
-    and the expected order of traversal as 'visited_order'."""
+    """A delegations graph as lists of delegations and target files
+    and the expected order of traversal as a list of role names."""
 
     delegations: List[TestDelegation]
-    visited_order: List[str]
+    target_files: List[TestTarget] = field(default_factory=list)
+    visited_order: List[str] = field(default_factory=list)
 
 
-class TestDelegationsGraphs(unittest.TestCase):
-    """Test creating delegations graphs with different complexity
-    and successfully updating the delegated roles metadata"""
+class TestDelegations(unittest.TestCase):
+    """Base class for delegation tests"""
 
     # set dump_dir to trigger repository state dumps
     dump_dir: Optional[str] = None
@@ -59,73 +66,76 @@ def setUp(self) -> None:
         self.targets_dir = os.path.join(self.temp_dir.name, "targets")
         os.mkdir(self.metadata_dir)
         os.mkdir(self.targets_dir)
+        self.sim: RepositorySimulator
 
     def tearDown(self) -> None:
         self.temp_dir.cleanup()
 
-    def setup_subtest(
-        self, delegations: List[TestDelegation]
-    ) -> RepositorySimulator:
-        sim = self._init_repo(delegations)
-
+    def setup_subtest(self) -> None:
         self.subtest_count += 1
         if self.dump_dir is not None:
             # create subtest dumpdir
             name = f"{self.id().split('.')[-1]}-{self.subtest_count}"
-            sim.dump_dir = os.path.join(self.dump_dir, name)
-            os.mkdir(sim.dump_dir)
+            self.sim.dump_dir = os.path.join(self.dump_dir, name)
+            os.mkdir(self.sim.dump_dir)
             # dump the repo simulator metadata
-            sim.write()
-
-        return sim
+            self.sim.write()
 
     def teardown_subtest(self) -> None:
-        # clean up after each subtest
         utils.cleanup_dir(self.metadata_dir)
 
-    def _init_updater(self, sim: RepositorySimulator) -> Updater:
-        """Create a new Updater instance"""
-        return Updater(
-            self.metadata_dir,
-            "https://example.com/metadata/",
-            self.targets_dir,
-            "https://example.com/targets/",
-            sim,
-        )
+    def _init_repo(self, test_case: DelegationsTestCase) -> None:
+        """Create a new RepositorySimulator instance and
+        populate it with delegations and target files"""
 
-    def _init_repo(
-        self, delegations: List[TestDelegation]
-    ) -> RepositorySimulator:
-        """Create a new RepositorySimulator instance with 'delegations'"""
-        sim = RepositorySimulator()
+        self.sim = RepositorySimulator()
         spec_version = ".".join(SPECIFICATION_VERSION)
-
-        for d in delegations:
-            if d.rolename in sim.md_delegates:
-                targets = sim.md_delegates[d.rolename].signed
+        for d in test_case.delegations:
+            if d.rolename in self.sim.md_delegates:
+                targets = self.sim.md_delegates[d.rolename].signed
             else:
-                targets = Targets(1, spec_version, sim.safe_expiry, {}, None)
-
+                targets = Targets(
+                    1, spec_version, self.sim.safe_expiry, {}, None
+                )
             # unpack 'd' but skip "delegator"
             role = DelegatedRole(*astuple(d)[1:])
-            sim.add_delegation(d.delegator, role, targets)
-        sim.update_snapshot()
+            self.sim.add_delegation(d.delegator, role, targets)
+
+        for target in test_case.target_files:
+            self.sim.add_target(*astuple(target))
 
+        if test_case.target_files:
+            self.sim.targets.version += 1
+        self.sim.update_snapshot()
+
+    def _init_updater(self) -> Updater:
+        """Create a new Updater instance"""
         # Init trusted root for Updater
         with open(os.path.join(self.metadata_dir, "root.json"), "bw") as f:
-            root = sim.download_bytes(
+            root = self.sim.download_bytes(
                 "https://example.com/metadata/1.root.json", 100000
             )
             f.write(root)
 
-        return sim
+        return Updater(
+            self.metadata_dir,
+            "https://example.com/metadata/",
+            self.targets_dir,
+            "https://example.com/targets/",
+            self.sim,
+        )
 
     def _assert_files_exist(self, roles: Iterable[str]) -> None:
         """Assert that local metadata files exist for 'roles'"""
         expected_files = sorted([f"{role}.json" for role in roles])
         local_metadata_files = sorted(os.listdir(self.metadata_dir))
         self.assertListEqual(local_metadata_files, expected_files)
 
+
+class TestDelegationsGraphs(TestDelegations):
+    """Test creating delegations graphs with different complexity
+    and successfully updating the delegated roles metadata"""
+
     graphs: utils.DataSet = {
         "basic delegation": DelegationsTestCase(
             delegations=[TestDelegation("targets", "A")],
@@ -251,8 +261,10 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None:
             exp_files = [*TOP_LEVEL_ROLE_NAMES, *test_data.visited_order]
             exp_calls = [call(role, 1) for role in test_data.visited_order]
 
-            sim = self.setup_subtest(test_data.delegations)
-            updater = self._init_updater(sim)
+            self._init_repo(test_data)
+            self.setup_subtest()
+
+            updater = self._init_updater()
             # restrict the max number of delegations to simplify the test
             updater.config.max_delegations = 4
             # Call explicitly refresh to simplify the expected_calls list
@@ -261,7 +273,7 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None:
             self._assert_files_exist(TOP_LEVEL_ROLE_NAMES)
 
             with patch.object(
-                sim, "_fetch_metadata", wraps=sim._fetch_metadata
+                self.sim, "_fetch_metadata", wraps=self.sim._fetch_metadata
             ) as wrapped_fetch:
                 # Looking for a non-existing targetpath forces updater
                 # to visit all possible delegated roles
@@ -276,10 +288,76 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None:
             self.teardown_subtest()
 
 
+class TestTargetFileSearch(TestDelegations):
+    """
+    Create a single repository with the following delegations:
+
+      targets
+    *py / \*
+       A   B
+     *txt / \*
+         C   D
+
+    Test that Updater successfully finds the target files metadata,
+    traversing the delegations as expected.
+    """
+
+    delegations_tree = DelegationsTestCase(
+        delegations=[
+            TestDelegation("targets", "A", paths=["*.py"]),
+            TestDelegation("targets", "B", paths=["*"]),
+            TestDelegation("B", "C", paths=["*.txt"]),
+            TestDelegation("B", "D", paths=["*"]),
+        ],
+        target_files=[
+            TestTarget("targets", b"content", "file"),
+            TestTarget("A", b"content", "file.py"),
+            TestTarget("C", b"content", "file.txt"),
+            TestTarget("D", b"content", "file.ext"),
+        ],
+    )
+
+    def setUp(self) -> None:
+        super().setUp()
+        self._init_repo(self.delegations_tree)
+
+    targets: utils.DataSet = {
+        "file": ("file", []),
+        "file.py": ("file.py", ["A"]),
+        "file.txt": ("file.txt", ["B", "C"]),
+        "file.ext": ("file.ext", ["B", "D"]),
+    }
+
+    @utils.run_sub_tests_with_dataset(targets)
+    def test_targetfile_search(self, test_data: Tuple[str, List[str]]) -> None:
+        try:
+            self.setup_subtest()
+            targetpath, visited_order = test_data
+            exp_files = [*TOP_LEVEL_ROLE_NAMES, *visited_order]
+            exp_calls = [call(role, 1) for role in visited_order]
+            updater = self._init_updater()
+            # Call explicitly refresh to simplify the expected_calls list
+            updater.refresh()
+            with patch.object(
+                self.sim, "_fetch_metadata", wraps=self.sim._fetch_metadata
+            ) as wrapped_fetch:
+                target = updater.get_targetinfo(targetpath)
+                # Confirm that the expected TargetFile is found
+                assert target is not None
+                exp_target = self.sim.target_files[targetpath].target_file
+                self.assertDictEqual(target.to_dict(), exp_target.to_dict())
+                # Check that the delegated roles were visited in the expected
+                # order and the corresponding metadata files were persisted
+                self.assertListEqual(wrapped_fetch.call_args_list, exp_calls)
+                self._assert_files_exist(exp_files)
+        finally:
+            self.teardown_subtest()
+
+
 if __name__ == "__main__":
     if "--dump" in sys.argv:
-        TestDelegationsGraphs.dump_dir = tempfile.mkdtemp()
-        print(f"Repository Simulator dumps in {TestDelegationsGraphs.dump_dir}")
+        TestDelegations.dump_dir = tempfile.mkdtemp()
+        print(f"Repository Simulator dumps in {TestDelegations.dump_dir}")
         sys.argv.remove("--dump")
 
     utils.configure_test_logging(sys.argv)