Transition to the newest version of TUF

taf/repository_tool.py

@@ -183,13 +183,6 @@ def __init__(self, path, name="default"):
 
     _framework_files = ["repositories.json", "test-auth-repo"]
 
-    @property
-    def targets_path(self):
-        return self.path / TARGETS_DIRECTORY_NAME
-
-    @property
-    def metadata_path(self):
-        return self.path / METADATA_DIRECTORY_NAME
 
     _tuf_repository = None
 
@@ -384,13 +377,6 @@ def add_existing_target(self, file_path, targets_role="targets", custom=None):
         targets_obj = self._role_obj(targets_role)
         self._add_target(targets_obj, file_path, custom)
 
-    def get_all_roles(self):
-        """
-        Return a list of all defined roles, main roles combined with delegated targets roles
-        """
-        all_target_roles = self.get_all_targets_roles()
-        all_roles = ["root", "targets", "snapshot", "timestamp"] + all_target_roles
-        return all_roles
 
     def get_all_target_files_state(self):
         """Create dictionaries of added/modified and removed files by comparing current
@@ -548,24 +534,6 @@ def modify_targets(self, added_data=None, removed_data=None):
 
         return targets_role
 
-    def all_target_files(self):
-        """
-        Return a set of relative paths of all files inside the targets
-        directory
-        """
-        targets = []
-        for root, _, filenames in os.walk(str(self.targets_path)):
-            for filename in filenames:
-                filepath = Path(root) / filename
-                if filepath.is_file():
-                    targets.append(
-                        str(
-                            Path(
-                                os.path.relpath(str(filepath), str(self.targets_path))
-                            ).as_posix()
-                        )
-                    )
-        return set(targets)
 
     def get_target_file_custom_data(self, target_path):
         """
@@ -821,25 +789,6 @@ def get_role_keys(self, role, parent_role=None):
             pass
         return self.get_delegated_role_property("keyids", role, parent_role)
 
-    def get_role_paths(self, role, parent_role=None):
-        """Get paths of the given role
-
-        Args:
-        - role(str): TUF role (root, targets, timestamp, snapshot or delegated one)
-        - parent_role(str): Name of the parent role of the delegated role. If not specified,
-                            it will be set automatically, but this might be slow if there
-                            are many delegations.
-
-        Returns:
-        Defined delegated paths of delegate target role or * in case of targets
-
-        Raises:
-        - securesystemslib.exceptions.FormatError: If the arguments are improperly formatted.
-        - securesystemslib.exceptions.UnknownRoleError: If 'rolename' has not been delegated by this
-        """
-        if role == "targets":
-            return "*"
-        return self.get_delegated_role_property("paths", role, parent_role)
 
     def get_role_repositories(self, role, parent_role=None):
         """Get repositories of the given role

taf/tests/test_repository/test_repo.py

@@ -30,28 +30,35 @@ def test_get_expiration_date():
     assert tuf_repo.get_expiration_date("delegated_role1") ==  datetime.datetime(2020, 2, 5, 18, 14, 2, tzinfo=datetime.timezone.utc)
 
 
-def test_get_all_roles_no_delegations():
+def test_get_all_target_roles_no_delegations():
     test_group_dir = TEST_DATA_REPOS_PATH / "test-repository-tool/test-happy-path-pkcs1v15" / "taf"
     tuf_repo = MetadataRepository(test_group_dir)
     assert tuf_repo.get_all_targets_roles() == ["targets"]
 
 
-def test_get_all_roles_with_delegations():
+def test_get_all_target_roles_with_delegations():
     test_group_dir = TEST_DATA_REPOS_PATH / "test-repository-tool/test-delegated-roles-pkcs1v15" / "taf"
     tuf_repo = MetadataRepository(test_group_dir)
     actual = tuf_repo.get_all_targets_roles()
     assert len(actual) == 4
     assert set(actual) == {"targets", "delegated_role1", "delegated_role2", "inner_delegated_role"}
 
 
+def test_get_all_roles_with_delegations():
+    test_group_dir = TEST_DATA_REPOS_PATH / "test-repository-tool/test-delegated-roles-pkcs1v15" / "taf"
+    tuf_repo = MetadataRepository(test_group_dir)
+    actual = tuf_repo.get_all_roles()
+    assert len(actual) == 7
+    assert set(actual) == {"root", "snapshot", "timestamp", "targets", "delegated_role1", "delegated_role2", "inner_delegated_role"}
+
 def test_find_delegated_roles_parent():
     test_group_dir = TEST_DATA_REPOS_PATH / "test-repository-tool/test-delegated-roles-pkcs1v15" / "taf"
     tuf_repo = MetadataRepository(test_group_dir)
     assert tuf_repo.find_delegated_roles_parent("delegated_role1") == "targets"
     assert tuf_repo.find_delegated_roles_parent("delegated_role2") == "targets"
     assert tuf_repo.find_delegated_roles_parent("inner_delegated_role") == "delegated_role2"
 
-def test_chec_if_role_exists():
+def test_check_if_role_exists():
     test_group_dir = TEST_DATA_REPOS_PATH / "test-repository-tool/test-delegated-roles-pkcs1v15" / "taf"
     tuf_repo = MetadataRepository(test_group_dir)
     assert tuf_repo.check_if_role_exists("targets")
@@ -67,3 +74,21 @@ def test_check_roles_expiration_dates():
     assert "targets" in expired_dict
     assert "delegated_role1" in expired_dict
     assert not len(will_expire_dict)
+
+def test_all_target_files():
+    test_group_dir = TEST_DATA_REPOS_PATH / "test-repository-tool/test-happy-path-pkcs1v15" / "taf"
+    tuf_repo = MetadataRepository(test_group_dir)
+    actual = tuf_repo.all_target_files()
+    assert len(actual) == 3
+    assert actual == {'branch', 'dummy/target_dummy_repo', 'repositories.json'}
+
+def test_get_role_paths():
+    test_group_dir = TEST_DATA_REPOS_PATH / "test-repository-tool/test-delegated-roles-pkcs1v15" / "taf"
+    tuf_repo = MetadataRepository(test_group_dir)
+    actual = tuf_repo.get_role_paths("delegated_role1")
+    assert actual == ["dir1/*"]
+    actual = tuf_repo.get_role_paths("delegated_role2")
+    assert actual == ["dir2/*"]
+    actual = tuf_repo.get_role_paths("inner_delegated_role")
+    assert actual == ["dir2/inner_delegated_role.txt"]
+

taf/tuf/repository.py

@@ -19,13 +19,15 @@
     Targets,
     TargetFile,
     Timestamp,
+    DelegatedRole,
 )
 from taf.exceptions import TAFError
 from tuf.repository import Repository
 
 logger = logging.getLogger(__name__)
 
 METADATA_DIRECTORY_NAME = "metadata"
+TARGETS_DIRECTORY_NAME = "targets"
 
 MAIN_ROLES = ["root", "targets", "snapshot", "timestamp"]
 
@@ -59,6 +61,10 @@ def __init__(self, path: Path) -> None:
     def metadata_path(self) -> Path:
         return self._path / METADATA_DIRECTORY_NAME
 
+    @property
+    def targets_path(self):
+        return self._path / TARGETS_DIRECTORY_NAME
+
     @property
     def targets_infos(self) -> Dict[str, MetaFile]:
         # tracks targets and root metadata changes, needed in `do_snapshot`
@@ -69,6 +75,24 @@ def snapshot_info(self) -> MetaFile:
         # tracks snapshot metadata changes, needed in `do_timestamp`
         return self._snapshot_info
 
+
+    def all_target_files(self):
+        """
+        Return a set of relative paths of all files inside the targets
+        directory
+        """
+        targets = []
+        # Assume self.targets_path is a Path object, or convert it if necessary
+        base_path = Path(self.targets_path)
+
+        for filepath in base_path.rglob('*'):
+            if filepath.is_file():
+                # Get the relative path to the base directory and convert it to a POSIX path
+                relative_path = filepath.relative_to(base_path).as_posix()
+                targets.append(relative_path)
+
+        return set(targets)
+
     def add_target_files(self, target_files: List[TargetFile]) -> None:
         """Add target files to top-level targets metadata."""
         with self.edit_targets() as targets:
@@ -206,6 +230,7 @@ def create(self, signers: Dict[str, Dict[str, Signer]]):
             signed.version = 0  # `close` will bump to initial valid verison 1
             self.close(signed.type, Metadata(signed))
 
+
     def find_delegated_roles_parent(self, delegated_role, parent=None):
         if parent is None:
             parent = "targets"
@@ -254,11 +279,20 @@ def _role_obj(self, role, parent=None):
             for delegation in delegations_data:
                 if delegation["name"] == role:
                     try:
-                        return Role.from_dict(delegation)
+                        return DelegatedRole.from_dict(delegation)
                     except (KeyError, ValueError):
                         raise TAFError(f"{delegation}.json is invalid")
             return None
 
+
+    def get_all_roles(self):
+        """
+        Return a list of all defined roles, main roles combined with delegated targets roles
+        """
+        all_target_roles = self.get_all_targets_roles()
+        all_roles = ["root", "snapshot", "timestamp"] + all_target_roles
+        return all_roles
+
     def get_all_targets_roles(self):
         """
         Return a list containing names of all target roles
@@ -301,7 +335,28 @@ def get_role_threshold(self, role: str, parent: Optional[str]=None ) -> int:
             raise TAFError(f"Role {role} does not exist")
         return role_obj.threshold
 
+    def get_role_paths(self, role, parent_role=None):
+        """Get paths of the given role
 
+        Args:
+        - role(str): TUF role (root, targets, timestamp, snapshot or delegated one)
+        - parent_role(str): Name of the parent role of the delegated role. If not specified,
+                            it will be set automatically, but this might be slow if there
+                            are many delegations.
+
+        Returns:
+        Defined delegated paths of delegate target role or * in case of targets
+
+        Raises:
+        - securesystemslib.exceptions.FormatError: If the arguments are improperly formatted.
+        - securesystemslib.exceptions.UnknownRoleError: If 'rolename' has not been delegated by this
+        """
+        if role == "targets":
+            return "*"
+        role = self._role_obj(role)
+        if role is None:
+            raise TAFError(f"Role {role} does not exist")
+        return role.paths
 
     def set_metadata_expiration_date(self, role, start_date=None, interval=None):
         """Set expiration date of the provided role.