344 lines
12 KiB
Python
344 lines
12 KiB
Python
"""Unit tests for the ResourceMerger."""
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.models import (
|
|
CpuArchitecture,
|
|
DiscoveredResource,
|
|
PlatformCategory,
|
|
ProviderType,
|
|
ScanResult,
|
|
)
|
|
from iac_reverse.generator import ResourceMerger
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def make_resource(
|
|
name: str = "nginx",
|
|
resource_type: str = "kubernetes_deployment",
|
|
unique_id: str = "default/deployments/nginx",
|
|
provider: ProviderType = ProviderType.KUBERNETES,
|
|
platform_category: PlatformCategory = PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture: CpuArchitecture = CpuArchitecture.AARCH64,
|
|
attributes: dict | None = None,
|
|
) -> 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=architecture,
|
|
endpoint="https://api.local:6443",
|
|
attributes=attributes or {},
|
|
raw_references=[],
|
|
)
|
|
|
|
|
|
def make_scan_result(resources: list[DiscoveredResource]) -> ScanResult:
|
|
"""Create a ScanResult wrapping the given resources."""
|
|
return ScanResult(
|
|
resources=resources,
|
|
warnings=[],
|
|
errors=[],
|
|
scan_timestamp="2024-01-15T10:30:00Z",
|
|
profile_hash="abc123",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResourceMergerSingleResult:
|
|
"""Single scan result passes through unchanged."""
|
|
|
|
def test_single_scan_result_passes_through_unchanged(self):
|
|
"""Resources from a single scan result are returned as-is."""
|
|
resources = [
|
|
make_resource(name="nginx", unique_id="k8s/nginx"),
|
|
make_resource(name="redis", unique_id="k8s/redis"),
|
|
]
|
|
scan_result = make_scan_result(resources)
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([scan_result])
|
|
|
|
assert len(merged) == 2
|
|
assert merged[0].name == "nginx"
|
|
assert merged[1].name == "redis"
|
|
|
|
def test_empty_scan_result_returns_empty_list(self):
|
|
"""An empty scan result produces an empty merged list."""
|
|
scan_result = make_scan_result([])
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([scan_result])
|
|
|
|
assert merged == []
|
|
|
|
|
|
class TestResourceMergerNoConflicts:
|
|
"""Two scan results with no conflicts merge cleanly."""
|
|
|
|
def test_two_results_different_names_merge_cleanly(self):
|
|
"""Resources with different names from different providers merge without prefixing."""
|
|
k8s_resources = [
|
|
make_resource(
|
|
name="nginx",
|
|
unique_id="k8s/nginx",
|
|
provider=ProviderType.KUBERNETES,
|
|
),
|
|
]
|
|
docker_resources = [
|
|
make_resource(
|
|
name="redis",
|
|
unique_id="docker/redis",
|
|
provider=ProviderType.DOCKER_SWARM,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
),
|
|
]
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([
|
|
make_scan_result(k8s_resources),
|
|
make_scan_result(docker_resources),
|
|
])
|
|
|
|
assert len(merged) == 2
|
|
names = {r.name for r in merged}
|
|
assert names == {"nginx", "redis"}
|
|
|
|
def test_same_name_same_provider_no_conflict(self):
|
|
"""Resources with the same name from the same provider are not conflicts."""
|
|
resources = [
|
|
make_resource(name="nginx", unique_id="k8s/ns1/nginx"),
|
|
make_resource(name="nginx", unique_id="k8s/ns2/nginx"),
|
|
]
|
|
scan_result = make_scan_result(resources)
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([scan_result])
|
|
|
|
assert len(merged) == 2
|
|
# Both keep original name since they're from the same provider
|
|
assert all(r.name == "nginx" for r in merged)
|
|
|
|
|
|
class TestResourceMergerConflictResolution:
|
|
"""Two scan results with name conflicts get provider-prefixed names."""
|
|
|
|
def test_conflicting_names_get_provider_prefix(self):
|
|
"""Resources with the same name from different providers get prefixed."""
|
|
k8s_resources = [
|
|
make_resource(
|
|
name="nginx",
|
|
unique_id="k8s/nginx",
|
|
provider=ProviderType.KUBERNETES,
|
|
),
|
|
]
|
|
docker_resources = [
|
|
make_resource(
|
|
name="nginx",
|
|
unique_id="docker/nginx",
|
|
provider=ProviderType.DOCKER_SWARM,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
),
|
|
]
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([
|
|
make_scan_result(k8s_resources),
|
|
make_scan_result(docker_resources),
|
|
])
|
|
|
|
assert len(merged) == 2
|
|
names = {r.name for r in merged}
|
|
assert "kubernetes_nginx" in names
|
|
assert "docker_swarm_nginx" in names
|
|
|
|
def test_non_conflicting_names_unchanged_alongside_conflicts(self):
|
|
"""Non-conflicting resources keep their original names even when conflicts exist."""
|
|
k8s_resources = [
|
|
make_resource(name="nginx", unique_id="k8s/nginx", provider=ProviderType.KUBERNETES),
|
|
make_resource(name="postgres", unique_id="k8s/postgres", provider=ProviderType.KUBERNETES),
|
|
]
|
|
docker_resources = [
|
|
make_resource(
|
|
name="nginx",
|
|
unique_id="docker/nginx",
|
|
provider=ProviderType.DOCKER_SWARM,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
),
|
|
]
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([
|
|
make_scan_result(k8s_resources),
|
|
make_scan_result(docker_resources),
|
|
])
|
|
|
|
assert len(merged) == 3
|
|
names = {r.name for r in merged}
|
|
assert "kubernetes_nginx" in names
|
|
assert "docker_swarm_nginx" in names
|
|
assert "postgres" in names
|
|
|
|
|
|
class TestResourceMergerPreservesAttributes:
|
|
"""Provider-specific attributes are preserved."""
|
|
|
|
def test_attributes_preserved_after_merge(self):
|
|
"""Provider-specific attributes remain unchanged after merging."""
|
|
k8s_attrs = {"namespace": "default", "replicas": 3, "image": "nginx:1.25"}
|
|
docker_attrs = {"mode": "replicated", "replicas": 2, "network": "overlay"}
|
|
|
|
k8s_resources = [
|
|
make_resource(
|
|
name="nginx",
|
|
unique_id="k8s/nginx",
|
|
provider=ProviderType.KUBERNETES,
|
|
attributes=k8s_attrs,
|
|
),
|
|
]
|
|
docker_resources = [
|
|
make_resource(
|
|
name="nginx",
|
|
unique_id="docker/nginx",
|
|
provider=ProviderType.DOCKER_SWARM,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
attributes=docker_attrs,
|
|
),
|
|
]
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([
|
|
make_scan_result(k8s_resources),
|
|
make_scan_result(docker_resources),
|
|
])
|
|
|
|
k8s_merged = next(r for r in merged if r.name == "kubernetes_nginx")
|
|
docker_merged = next(r for r in merged if r.name == "docker_swarm_nginx")
|
|
|
|
assert k8s_merged.attributes == k8s_attrs
|
|
assert docker_merged.attributes == docker_attrs
|
|
|
|
def test_provider_and_metadata_preserved(self):
|
|
"""Provider type, platform category, and architecture are preserved."""
|
|
k8s_resources = [
|
|
make_resource(
|
|
name="app",
|
|
unique_id="k8s/app",
|
|
provider=ProviderType.KUBERNETES,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture=CpuArchitecture.AARCH64,
|
|
),
|
|
]
|
|
harvester_resources = [
|
|
make_resource(
|
|
name="app",
|
|
unique_id="harvester/app",
|
|
provider=ProviderType.HARVESTER,
|
|
platform_category=PlatformCategory.HCI,
|
|
architecture=CpuArchitecture.AMD64,
|
|
),
|
|
]
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([
|
|
make_scan_result(k8s_resources),
|
|
make_scan_result(harvester_resources),
|
|
])
|
|
|
|
k8s_merged = next(r for r in merged if r.name == "kubernetes_app")
|
|
harvester_merged = next(r for r in merged if r.name == "harvester_app")
|
|
|
|
assert k8s_merged.provider == ProviderType.KUBERNETES
|
|
assert k8s_merged.platform_category == PlatformCategory.CONTAINER_ORCHESTRATION
|
|
assert k8s_merged.architecture == CpuArchitecture.AARCH64
|
|
|
|
assert harvester_merged.provider == ProviderType.HARVESTER
|
|
assert harvester_merged.platform_category == PlatformCategory.HCI
|
|
assert harvester_merged.architecture == CpuArchitecture.AMD64
|
|
|
|
|
|
class TestResourceMergerBothAppearInOutput:
|
|
"""Resources from different providers with same name both appear in output."""
|
|
|
|
def test_both_conflicting_resources_present(self):
|
|
"""Both resources with the same name from different providers appear in output."""
|
|
k8s_resources = [
|
|
make_resource(
|
|
name="webserver",
|
|
unique_id="k8s/webserver",
|
|
provider=ProviderType.KUBERNETES,
|
|
attributes={"replicas": 3},
|
|
),
|
|
]
|
|
docker_resources = [
|
|
make_resource(
|
|
name="webserver",
|
|
unique_id="docker/webserver",
|
|
provider=ProviderType.DOCKER_SWARM,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
attributes={"mode": "global"},
|
|
),
|
|
]
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([
|
|
make_scan_result(k8s_resources),
|
|
make_scan_result(docker_resources),
|
|
])
|
|
|
|
assert len(merged) == 2
|
|
unique_ids = {r.unique_id for r in merged}
|
|
assert "k8s/webserver" in unique_ids
|
|
assert "docker/webserver" in unique_ids
|
|
|
|
def test_three_providers_same_name_all_appear(self):
|
|
"""Three providers with the same resource name all appear with prefixes."""
|
|
k8s_resources = [
|
|
make_resource(
|
|
name="monitor",
|
|
unique_id="k8s/monitor",
|
|
provider=ProviderType.KUBERNETES,
|
|
),
|
|
]
|
|
docker_resources = [
|
|
make_resource(
|
|
name="monitor",
|
|
unique_id="docker/monitor",
|
|
provider=ProviderType.DOCKER_SWARM,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
),
|
|
]
|
|
harvester_resources = [
|
|
make_resource(
|
|
name="monitor",
|
|
unique_id="harvester/monitor",
|
|
provider=ProviderType.HARVESTER,
|
|
platform_category=PlatformCategory.HCI,
|
|
architecture=CpuArchitecture.AMD64,
|
|
),
|
|
]
|
|
|
|
merger = ResourceMerger()
|
|
merged = merger.merge([
|
|
make_scan_result(k8s_resources),
|
|
make_scan_result(docker_resources),
|
|
make_scan_result(harvester_resources),
|
|
])
|
|
|
|
assert len(merged) == 3
|
|
names = {r.name for r in merged}
|
|
assert "kubernetes_monitor" in names
|
|
assert "docker_swarm_monitor" in names
|
|
assert "harvester_monitor" in names
|