Add test_updater_deleagation_graphs.py

Add tests creating delegations graphs with different complexity
and successfully updating the delegated roles metadata.

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

Author: sechkova

tests/test_updater_delegation_graphs.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python
+
+# 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"""
+
+import os
+import sys
+import tempfile
+import unittest
+from dataclasses import dataclass
+from typing import Any, Dict, Iterable, List, Optional
+from unittest.mock import call, patch
+
+from tests import utils
+from tests.repository_simulator import RepositorySimulator
+from tuf.api.metadata import (
+    SPECIFICATION_VERSION,
+    TOP_LEVEL_ROLE_NAMES,
+    DelegatedRole,
+    Targets,
+)
+from tuf.ngclient import Updater
+
+
+@dataclass
+class Delegation:
+    def __init__(
+        self,
+        delegator: str,
+        rolename: str,
+        keyids: Optional[List[str]] = None,
+        threshold: int = 1,
+        terminating: bool = False,
+        paths: Optional[List[str]] = ["*"],
+        path_hash_prefixes: Optional[List[str]] = None,
+    ):
+        self.delegator = delegator
+        keyids = keyids or []
+        self.role = DelegatedRole(
+            rolename, keyids, threshold, terminating, paths, path_hash_prefixes
+        )
+
+
+class TestDelegationsGraphs(unittest.TestCase):
+    """Test creating delegations graphs with different complexity
+    and successfully updating the delegated roles metadata"""
+
+    def setUp(self) -> None:
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.metadata_dir = os.path.join(self.temp_dir.name, "metadata")
+        self.targets_dir = os.path.join(self.temp_dir.name, "targets")
+        os.mkdir(self.metadata_dir)
+        os.mkdir(self.targets_dir)
+
+    def tearDown(self) -> None:
+        self.temp_dir.cleanup()
+
+    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, delegations: List[Delegation]) -> RepositorySimulator:
+        """Create a new RepositorySimulator instance with 'delegations'"""
+        sim = RepositorySimulator()
+        spec_version = ".".join(SPECIFICATION_VERSION)
+
+        for delegation in delegations:
+            if delegation.role.name in sim.md_delegates:
+                targets = sim.md_delegates[delegation.role.name].signed
+            else:
+                targets = Targets(1, spec_version, sim.safe_expiry, {}, None)
+            sim.add_delegation(delegation.delegator, delegation.role, targets)
+        sim.update_snapshot()
+
+        # Init trusted root for Updater
+        with open(os.path.join(self.metadata_dir, "root.json"), "bw") as f:
+            root = sim.download_bytes(
+                "https://example.com/metadata/1.root.json", 100000
+            )
+            f.write(root)
+
+        return 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)
+
+    graphs: utils.DataSet = {
+        "single delegation": {
+            "delegations": [Delegation("targets", "A")],
+            "visited roles": ["A"],
+        },
+        "1-level tree": {
+            "delegations": [
+                Delegation("targets", "A"),
+                Delegation("targets", "B"),
+            ],
+            "visited roles": ["A", "B"],
+        },
+        "2-level tree-1": {
+            "delegations": [
+                Delegation("targets", "A"),
+                Delegation("targets", "B"),
+                Delegation("B", "C"),
+            ],
+            "visited roles": ["A", "B", "C"],
+        },
+        "2-level tree-2": {
+            "delegations": [
+                Delegation("targets", "A"),
+                Delegation("targets", "B"),
+                Delegation("A", "C"),
+                Delegation("A", "D"),
+            ],
+            "visited roles": ["A", "C", "D", "B"],
+        },
+        "2-level tree terminating": {
+            "delegations": [
+                Delegation("targets", "A"),
+                Delegation("targets", "B"),
+                Delegation("A", "C", terminating=True),
+                Delegation("A", "D"),
+            ],
+            "visited roles": ["A", "C"],
+        },
+        "3-level tree": {
+            "delegations": [
+                Delegation("targets", "A"),
+                Delegation("targets", "B"),
+                Delegation("A", "C"),
+                Delegation("C", "D"),
+            ],
+            "visited roles": ["A", "C", "D", "B"],
+        },
+        "3-level tree terminating": {
+            "delegations": [
+                Delegation("targets", "A"),
+                Delegation("targets", "B"),
+                Delegation("A", "C", terminating=True),
+                Delegation("C", "D"),
+            ],
+            "visited roles": ["A", "C", "D"],
+        },
+        "cyclic graph": {
+            "delegations": [
+                Delegation("targets", "A"),
+                Delegation("targets", "B"),
+                Delegation("B", "C"),
+                Delegation("C", "D"),
+                Delegation("D", "B"),
+            ],
+            "visited roles": ["A", "B", "C", "D"],
+        },
+        "2-level tree pathpattern": {
+            "delegations": [
+                Delegation("targets", "A", paths=["*.py"]),
+                Delegation("targets", "B"),
+                Delegation("A", "C"),
+                Delegation("A", "D"),
+            ],
+            "visited roles": ["B"],
+        },
+        "3-level tree pathpattern": {
+            "delegations": [
+                Delegation("targets", "A"),
+                Delegation("targets", "B"),
+                Delegation("A", "C", paths=["*.py"]),
+                Delegation("C", "D"),
+            ],
+            "visited roles": ["A", "B"],
+        },
+    }
+
+    @utils.run_sub_tests_with_dataset(graphs)
+    def test_graph_traversal(
+        self, test_case_data: Dict[str, List[Any]]
+    ) -> None:
+        delegations: List[Delegation] = test_case_data["delegations"]
+        visited_roles: List[str] = test_case_data["visited roles"]
+        expected_files = [*TOP_LEVEL_ROLE_NAMES, *visited_roles]
+        expected_calls = [call(role, 1) for role in visited_roles]
+
+        sim = self._init_repo(delegations)
+        updater = self._init_updater(sim)
+        # Call explicitly refresh to simplify the expected_calls list
+        updater.refresh()
+        # Check that metadata dir contains only top-level roles
+        self._assert_files_exist(TOP_LEVEL_ROLE_NAMES)
+
+        with patch.object(
+            sim, "_fetch_metadata", wraps=sim._fetch_metadata
+        ) as wrapped_fetch:
+            # Looking for a non-existing targetpath forces updater
+            # to visit all possible delegated roles
+            targetfile = updater.get_targetinfo("missingpath")
+
+            self.assertIsNone(targetfile)
+            self.assertListEqual(wrapped_fetch.call_args_list, expected_calls)
+            self._assert_files_exist(expected_files)
+
+        utils.cleanup_dir(self.metadata_dir)
+
+
+if __name__ == "__main__":
+
+    utils.configure_test_logging(sys.argv)
+    unittest.main()