538 lines
19 KiB
Python
538 lines
19 KiB
Python
"""Unit tests for cycle detection and resolution in the DependencyResolver."""
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.models import (
|
|
CpuArchitecture,
|
|
CycleReport,
|
|
DiscoveredResource,
|
|
PlatformCategory,
|
|
ProviderType,
|
|
ScanResult,
|
|
)
|
|
from iac_reverse.resolver import DependencyResolver
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def make_resource(
|
|
resource_type: str = "kubernetes_deployment",
|
|
unique_id: str = "default/deployments/nginx",
|
|
name: str = "nginx",
|
|
raw_references: list[str] | None = None,
|
|
attributes: dict | None = None,
|
|
provider: ProviderType = ProviderType.KUBERNETES,
|
|
platform_category: PlatformCategory = PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
) -> DiscoveredResource:
|
|
"""Create a sample DiscoveredResource for testing."""
|
|
return DiscoveredResource(
|
|
resource_type=resource_type,
|
|
unique_id=unique_id,
|
|
name=name,
|
|
provider=provider,
|
|
platform_category=platform_category,
|
|
architecture=CpuArchitecture.AARCH64,
|
|
endpoint="https://k8s-api.local:6443",
|
|
attributes=attributes or {},
|
|
raw_references=raw_references or [],
|
|
)
|
|
|
|
|
|
def make_scan_result(resources: list[DiscoveredResource]) -> ScanResult:
|
|
"""Create a ScanResult from a list of resources."""
|
|
return ScanResult(
|
|
resources=resources,
|
|
warnings=[],
|
|
errors=[],
|
|
scan_timestamp="2024-01-15T10:30:00Z",
|
|
profile_hash="test-hash",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Simple A -> B -> A cycle detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSimpleTwoNodeCycle:
|
|
"""Tests for a simple two-node cycle: A -> B -> A."""
|
|
|
|
def test_two_node_cycle_detected(self):
|
|
"""A simple A -> B -> A cycle is detected."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-a",
|
|
name="service-a",
|
|
raw_references=["svc-b"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-b",
|
|
name="service-b",
|
|
raw_references=["svc-a"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
# Should detect exactly one cycle
|
|
assert len(graph.cycles) == 1
|
|
# The cycle should contain both resource IDs
|
|
cycle = graph.cycles[0]
|
|
assert set(cycle) == {"svc-a", "svc-b"}
|
|
|
|
def test_two_node_cycle_has_cycle_report(self):
|
|
"""A two-node cycle produces a CycleReport with resolution suggestion."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-a",
|
|
name="service-a",
|
|
raw_references=["svc-b"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-b",
|
|
name="service-b",
|
|
raw_references=["svc-a"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert len(graph.cycle_reports) == 1
|
|
report = graph.cycle_reports[0]
|
|
assert isinstance(report, CycleReport)
|
|
assert "data source" in report.resolution_strategy.lower()
|
|
|
|
def test_two_node_cycle_still_produces_topological_order(self):
|
|
"""Despite a cycle, a topological order is still produced."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-a",
|
|
name="service-a",
|
|
raw_references=["svc-b"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-b",
|
|
name="service-b",
|
|
raw_references=["svc-a"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
# Topological order should contain both resources
|
|
assert len(graph.topological_order) == 2
|
|
assert set(graph.topological_order) == {"svc-a", "svc-b"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Multi-node cycle (A -> B -> C -> A)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMultiNodeCycle:
|
|
"""Tests for a multi-node cycle: A -> B -> C -> A."""
|
|
|
|
def test_three_node_cycle_detected(self):
|
|
"""A three-node cycle A -> B -> C -> A is detected."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-a",
|
|
name="service-a",
|
|
raw_references=["svc-c"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-b",
|
|
name="service-b",
|
|
raw_references=["svc-a"],
|
|
)
|
|
resource_c = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-c",
|
|
name="service-c",
|
|
raw_references=["svc-b"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b, resource_c])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
# Should detect exactly one cycle containing all three nodes
|
|
assert len(graph.cycles) == 1
|
|
cycle = graph.cycles[0]
|
|
assert set(cycle) == {"svc-a", "svc-b", "svc-c"}
|
|
|
|
def test_three_node_cycle_produces_topological_order(self):
|
|
"""A three-node cycle still produces a valid topological order."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-a",
|
|
name="service-a",
|
|
raw_references=["svc-c"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-b",
|
|
name="service-b",
|
|
raw_references=["svc-a"],
|
|
)
|
|
resource_c = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-c",
|
|
name="service-c",
|
|
raw_references=["svc-b"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b, resource_c])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert len(graph.topological_order) == 3
|
|
assert set(graph.topological_order) == {"svc-a", "svc-b", "svc-c"}
|
|
|
|
def test_three_node_cycle_report_has_all_nodes(self):
|
|
"""The cycle report for a 3-node cycle lists all involved resources."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-a",
|
|
name="service-a",
|
|
raw_references=["svc-c"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-b",
|
|
name="service-b",
|
|
raw_references=["svc-a"],
|
|
)
|
|
resource_c = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-c",
|
|
name="service-c",
|
|
raw_references=["svc-b"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b, resource_c])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert len(graph.cycle_reports) == 1
|
|
report = graph.cycle_reports[0]
|
|
assert set(report.cycle) == {"svc-a", "svc-b", "svc-c"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Graph with both cycles and acyclic portions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMixedCyclicAndAcyclicGraph:
|
|
"""Tests for graphs containing both cyclic and acyclic portions."""
|
|
|
|
def test_cycle_detected_alongside_acyclic_resources(self):
|
|
"""Cycles are detected even when acyclic resources exist in the graph."""
|
|
# Acyclic chain: D -> E (E depends on D)
|
|
resource_d = make_resource(
|
|
resource_type="kubernetes_namespace",
|
|
unique_id="ns/production",
|
|
name="production",
|
|
)
|
|
resource_e = make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="prod/deployments/app",
|
|
name="app",
|
|
raw_references=["ns/production"],
|
|
)
|
|
|
|
# Cycle: A -> B -> A
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-x",
|
|
name="service-x",
|
|
raw_references=["svc-y"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-y",
|
|
name="service-y",
|
|
raw_references=["svc-x"],
|
|
)
|
|
|
|
scan_result = make_scan_result(
|
|
[resource_d, resource_e, resource_a, resource_b]
|
|
)
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
# Should detect the cycle
|
|
assert len(graph.cycles) == 1
|
|
assert set(graph.cycles[0]) == {"svc-x", "svc-y"}
|
|
|
|
def test_acyclic_ordering_preserved_despite_cycle_elsewhere(self):
|
|
"""Acyclic portions maintain correct topological ordering."""
|
|
# Acyclic chain: namespace -> deployment
|
|
namespace = make_resource(
|
|
resource_type="kubernetes_namespace",
|
|
unique_id="ns/prod",
|
|
name="prod",
|
|
)
|
|
deployment = make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="prod/deploy/app",
|
|
name="app",
|
|
raw_references=["ns/prod"],
|
|
)
|
|
|
|
# Cycle: X -> Y -> X
|
|
svc_x = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-x",
|
|
name="svc-x",
|
|
raw_references=["svc-y"],
|
|
)
|
|
svc_y = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-y",
|
|
name="svc-y",
|
|
raw_references=["svc-x"],
|
|
)
|
|
|
|
scan_result = make_scan_result([namespace, deployment, svc_x, svc_y])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
order = graph.topological_order
|
|
# The acyclic relationship should still be respected
|
|
assert order.index("ns/prod") < order.index("prod/deploy/app")
|
|
|
|
def test_all_resources_present_in_topological_order(self):
|
|
"""All resources (cyclic and acyclic) appear in the topological order."""
|
|
namespace = make_resource(
|
|
resource_type="kubernetes_namespace",
|
|
unique_id="ns/default",
|
|
name="default",
|
|
)
|
|
deployment = make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="default/deploy/web",
|
|
name="web",
|
|
raw_references=["ns/default"],
|
|
)
|
|
svc_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="cycle-a",
|
|
name="cycle-a",
|
|
raw_references=["cycle-b"],
|
|
)
|
|
svc_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="cycle-b",
|
|
name="cycle-b",
|
|
raw_references=["cycle-a"],
|
|
)
|
|
|
|
scan_result = make_scan_result([namespace, deployment, svc_a, svc_b])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert len(graph.topological_order) == 4
|
|
assert set(graph.topological_order) == {
|
|
"ns/default",
|
|
"default/deploy/web",
|
|
"cycle-a",
|
|
"cycle-b",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Resolution suggestion identifies correct relationship to break
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResolutionSuggestions:
|
|
"""Tests that resolution suggestions prefer breaking 'reference' over
|
|
'dependency' over 'parent-child'."""
|
|
|
|
def test_prefers_breaking_reference_over_dependency(self):
|
|
"""When a cycle has both reference and dependency edges, suggests breaking reference."""
|
|
# Create a cycle where one edge is "dependency" and one is "reference"
|
|
# IIS site -> app pool (dependency), app pool -> IIS site (reference)
|
|
app_pool = make_resource(
|
|
resource_type="windows_iis_app_pool",
|
|
unique_id="pool-a",
|
|
name="pool-a",
|
|
raw_references=["site-a"], # This creates a "reference" relationship
|
|
provider=ProviderType.WINDOWS,
|
|
platform_category=PlatformCategory.WINDOWS,
|
|
)
|
|
iis_site = make_resource(
|
|
resource_type="windows_iis_site",
|
|
unique_id="site-a",
|
|
name="site-a",
|
|
raw_references=["pool-a"], # This creates a "dependency" relationship
|
|
provider=ProviderType.WINDOWS,
|
|
platform_category=PlatformCategory.WINDOWS,
|
|
)
|
|
|
|
scan_result = make_scan_result([app_pool, iis_site])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert len(graph.cycle_reports) == 1
|
|
report = graph.cycle_reports[0]
|
|
# Should suggest breaking the "reference" relationship (pool -> site)
|
|
assert report.break_relationship_type == "reference"
|
|
|
|
def test_prefers_breaking_dependency_over_parent_child(self):
|
|
"""When a cycle has dependency and parent-child edges, suggests breaking dependency."""
|
|
# Create a cycle: namespace -> deployment (parent-child back-ref),
|
|
# deployment -> namespace (dependency)
|
|
# We need to craft this carefully:
|
|
# - deployment references namespace -> classified as "parent-child" (namespace is in _NAMESPACE_RESOURCE_TYPES)
|
|
# - namespace references deployment -> classified as "reference" (deployment is not special)
|
|
# Actually let's use a different setup to get dependency vs parent-child
|
|
|
|
# Use harvester: VM -> network (dependency), network -> VM (reference)
|
|
network = make_resource(
|
|
resource_type="harvester_network",
|
|
unique_id="net-a",
|
|
name="net-a",
|
|
raw_references=["vm-a"], # reference (VM is not a namespace type)
|
|
provider=ProviderType.HARVESTER,
|
|
platform_category=PlatformCategory.HCI,
|
|
)
|
|
vm = make_resource(
|
|
resource_type="harvester_virtualmachine",
|
|
unique_id="vm-a",
|
|
name="vm-a",
|
|
raw_references=["net-a"], # dependency (VM depends on network)
|
|
provider=ProviderType.HARVESTER,
|
|
platform_category=PlatformCategory.HCI,
|
|
)
|
|
|
|
scan_result = make_scan_result([network, vm])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert len(graph.cycle_reports) == 1
|
|
report = graph.cycle_reports[0]
|
|
# The network -> VM edge is "reference", VM -> network is "parent-child"
|
|
# (because harvester_network is in _NAMESPACE_RESOURCE_TYPES)
|
|
# So it should prefer breaking "reference"
|
|
assert report.break_relationship_type == "reference"
|
|
|
|
def test_resolution_strategy_mentions_data_source(self):
|
|
"""Resolution strategy suggests data source lookup as alternative."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-a",
|
|
name="service-a",
|
|
raw_references=["svc-b"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-b",
|
|
name="service-b",
|
|
raw_references=["svc-a"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert len(graph.cycle_reports) == 1
|
|
report = graph.cycle_reports[0]
|
|
assert "data source" in report.resolution_strategy.lower()
|
|
|
|
def test_suggested_break_edge_is_in_cycle(self):
|
|
"""The suggested edge to break is actually part of the cycle."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-a",
|
|
name="service-a",
|
|
raw_references=["svc-b"],
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc-b",
|
|
name="service-b",
|
|
raw_references=["svc-a"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
report = graph.cycle_reports[0]
|
|
# The suggested break edge nodes should be in the cycle
|
|
source, target = report.suggested_break
|
|
assert source in report.cycle
|
|
assert target in report.cycle
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: No cycles in acyclic graph
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNoCycles:
|
|
"""Tests that acyclic graphs report no cycles."""
|
|
|
|
def test_linear_chain_has_no_cycles(self):
|
|
"""A simple linear chain has no cycles."""
|
|
resource_a = make_resource(
|
|
resource_type="kubernetes_namespace",
|
|
unique_id="ns/default",
|
|
name="default",
|
|
)
|
|
resource_b = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="default/svc/app",
|
|
name="app",
|
|
raw_references=["ns/default"],
|
|
)
|
|
resource_c = make_resource(
|
|
resource_type="kubernetes_ingress",
|
|
unique_id="default/ingress/app",
|
|
name="app-ingress",
|
|
raw_references=["default/svc/app"],
|
|
)
|
|
|
|
scan_result = make_scan_result([resource_a, resource_b, resource_c])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert graph.cycles == []
|
|
assert graph.cycle_reports == []
|
|
|
|
def test_empty_graph_has_no_cycles(self):
|
|
"""An empty graph has no cycles."""
|
|
scan_result = make_scan_result([])
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert graph.cycles == []
|
|
assert graph.cycle_reports == []
|
|
|
|
def test_standalone_resources_have_no_cycles(self):
|
|
"""Resources with no references have no cycles."""
|
|
resources = [
|
|
make_resource(unique_id=f"res-{i}", name=f"res-{i}")
|
|
for i in range(5)
|
|
]
|
|
|
|
scan_result = make_scan_result(resources)
|
|
resolver = DependencyResolver(scan_result)
|
|
graph = resolver.resolve()
|
|
|
|
assert graph.cycles == []
|
|
assert graph.cycle_reports == []
|