336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""Unit tests for the ChangeDetector class."""
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.incremental.change_detector import ChangeDetector
|
|
from iac_reverse.models import (
|
|
ChangeSummary,
|
|
ChangeType,
|
|
CpuArchitecture,
|
|
DiscoveredResource,
|
|
PlatformCategory,
|
|
ProviderType,
|
|
ScanResult,
|
|
)
|
|
|
|
|
|
def _make_resource(
|
|
unique_id: str = "res-1",
|
|
name: str = "test-resource",
|
|
resource_type: str = "kubernetes_deployment",
|
|
attributes: dict | None = None,
|
|
) -> DiscoveredResource:
|
|
"""Create a sample DiscoveredResource for testing."""
|
|
return DiscoveredResource(
|
|
resource_type=resource_type,
|
|
unique_id=unique_id,
|
|
name=name,
|
|
provider=ProviderType.KUBERNETES,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture=CpuArchitecture.AMD64,
|
|
endpoint="https://k8s-api.internal.lab:6443",
|
|
attributes=attributes if attributes is not None else {"replicas": 3},
|
|
)
|
|
|
|
|
|
def _make_scan_result(
|
|
resources: list[DiscoveredResource] | None = None,
|
|
) -> ScanResult:
|
|
"""Create a sample ScanResult for testing."""
|
|
return ScanResult(
|
|
resources=resources if resources is not None else [],
|
|
warnings=[],
|
|
errors=[],
|
|
scan_timestamp="2024-01-15T10:30:00Z",
|
|
profile_hash="abc123",
|
|
)
|
|
|
|
|
|
class TestNoChanges:
|
|
"""Tests for identical scans producing no changes."""
|
|
|
|
def test_identical_scans_produce_no_changes(self) -> None:
|
|
"""Comparing identical scans returns empty change summary."""
|
|
resource = _make_resource(unique_id="res-1", attributes={"replicas": 3})
|
|
current = _make_scan_result(resources=[resource])
|
|
previous = _make_scan_result(resources=[resource])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.added_count == 0
|
|
assert summary.removed_count == 0
|
|
assert summary.modified_count == 0
|
|
assert summary.changes == []
|
|
|
|
def test_empty_scans_produce_no_changes(self) -> None:
|
|
"""Comparing two empty scans returns empty change summary."""
|
|
current = _make_scan_result(resources=[])
|
|
previous = _make_scan_result(resources=[])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.added_count == 0
|
|
assert summary.removed_count == 0
|
|
assert summary.modified_count == 0
|
|
assert summary.changes == []
|
|
|
|
|
|
class TestAddedResources:
|
|
"""Tests for detecting added resources."""
|
|
|
|
def test_new_resource_detected_as_added(self) -> None:
|
|
"""A resource in current but not in previous is classified as ADDED."""
|
|
resource = _make_resource(unique_id="new-res", name="new-service")
|
|
current = _make_scan_result(resources=[resource])
|
|
previous = _make_scan_result(resources=[])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.added_count == 1
|
|
assert summary.removed_count == 0
|
|
assert summary.modified_count == 0
|
|
assert len(summary.changes) == 1
|
|
|
|
change = summary.changes[0]
|
|
assert change.resource_id == "new-res"
|
|
assert change.resource_name == "new-service"
|
|
assert change.change_type == ChangeType.ADDED
|
|
assert change.changed_attributes is None
|
|
|
|
def test_multiple_added_resources(self) -> None:
|
|
"""Multiple new resources are all classified as ADDED."""
|
|
res1 = _make_resource(unique_id="res-1", name="service-1")
|
|
res2 = _make_resource(unique_id="res-2", name="service-2")
|
|
current = _make_scan_result(resources=[res1, res2])
|
|
previous = _make_scan_result(resources=[])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.added_count == 2
|
|
added_ids = {c.resource_id for c in summary.changes}
|
|
assert added_ids == {"res-1", "res-2"}
|
|
|
|
|
|
class TestRemovedResources:
|
|
"""Tests for detecting removed resources."""
|
|
|
|
def test_missing_resource_detected_as_removed(self) -> None:
|
|
"""A resource in previous but not in current is classified as REMOVED."""
|
|
resource = _make_resource(unique_id="old-res", name="old-service")
|
|
current = _make_scan_result(resources=[])
|
|
previous = _make_scan_result(resources=[resource])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.added_count == 0
|
|
assert summary.removed_count == 1
|
|
assert summary.modified_count == 0
|
|
assert len(summary.changes) == 1
|
|
|
|
change = summary.changes[0]
|
|
assert change.resource_id == "old-res"
|
|
assert change.resource_name == "old-service"
|
|
assert change.change_type == ChangeType.REMOVED
|
|
assert change.changed_attributes is None
|
|
|
|
def test_multiple_removed_resources(self) -> None:
|
|
"""Multiple missing resources are all classified as REMOVED."""
|
|
res1 = _make_resource(unique_id="res-1", name="service-1")
|
|
res2 = _make_resource(unique_id="res-2", name="service-2")
|
|
current = _make_scan_result(resources=[])
|
|
previous = _make_scan_result(resources=[res1, res2])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.removed_count == 2
|
|
removed_ids = {c.resource_id for c in summary.changes}
|
|
assert removed_ids == {"res-1", "res-2"}
|
|
|
|
|
|
class TestModifiedResources:
|
|
"""Tests for detecting modified resources."""
|
|
|
|
def test_changed_attributes_detected_as_modified(self) -> None:
|
|
"""A resource with changed attributes is classified as MODIFIED."""
|
|
prev_resource = _make_resource(
|
|
unique_id="res-1", attributes={"replicas": 3, "image": "nginx:1.24"}
|
|
)
|
|
curr_resource = _make_resource(
|
|
unique_id="res-1", attributes={"replicas": 5, "image": "nginx:1.24"}
|
|
)
|
|
current = _make_scan_result(resources=[curr_resource])
|
|
previous = _make_scan_result(resources=[prev_resource])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.added_count == 0
|
|
assert summary.removed_count == 0
|
|
assert summary.modified_count == 1
|
|
assert len(summary.changes) == 1
|
|
|
|
change = summary.changes[0]
|
|
assert change.resource_id == "res-1"
|
|
assert change.change_type == ChangeType.MODIFIED
|
|
assert change.changed_attributes == {"replicas": {"old": 3, "new": 5}}
|
|
|
|
def test_added_attribute_detected_as_modified(self) -> None:
|
|
"""A resource with a new attribute key is classified as MODIFIED."""
|
|
prev_resource = _make_resource(
|
|
unique_id="res-1", attributes={"replicas": 3}
|
|
)
|
|
curr_resource = _make_resource(
|
|
unique_id="res-1", attributes={"replicas": 3, "image": "nginx:1.25"}
|
|
)
|
|
current = _make_scan_result(resources=[curr_resource])
|
|
previous = _make_scan_result(resources=[prev_resource])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.modified_count == 1
|
|
change = summary.changes[0]
|
|
assert change.changed_attributes == {
|
|
"image": {"old": None, "new": "nginx:1.25"}
|
|
}
|
|
|
|
def test_removed_attribute_detected_as_modified(self) -> None:
|
|
"""A resource with a removed attribute key is classified as MODIFIED."""
|
|
prev_resource = _make_resource(
|
|
unique_id="res-1", attributes={"replicas": 3, "image": "nginx:1.25"}
|
|
)
|
|
curr_resource = _make_resource(
|
|
unique_id="res-1", attributes={"replicas": 3}
|
|
)
|
|
current = _make_scan_result(resources=[curr_resource])
|
|
previous = _make_scan_result(resources=[prev_resource])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.modified_count == 1
|
|
change = summary.changes[0]
|
|
assert change.changed_attributes == {
|
|
"image": {"old": "nginx:1.25", "new": None}
|
|
}
|
|
|
|
|
|
class TestMixedChanges:
|
|
"""Tests for scans with a mix of added, removed, and modified resources."""
|
|
|
|
def test_mixed_changes_detected_correctly(self) -> None:
|
|
"""A scan with added, removed, and modified resources is classified correctly."""
|
|
# Shared resource (modified)
|
|
prev_shared = _make_resource(
|
|
unique_id="shared", name="shared-svc", attributes={"replicas": 2}
|
|
)
|
|
curr_shared = _make_resource(
|
|
unique_id="shared", name="shared-svc", attributes={"replicas": 4}
|
|
)
|
|
|
|
# Removed resource
|
|
removed = _make_resource(unique_id="old-res", name="old-svc")
|
|
|
|
# Added resource
|
|
added = _make_resource(unique_id="new-res", name="new-svc")
|
|
|
|
previous = _make_scan_result(resources=[prev_shared, removed])
|
|
current = _make_scan_result(resources=[curr_shared, added])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.added_count == 1
|
|
assert summary.removed_count == 1
|
|
assert summary.modified_count == 1
|
|
assert len(summary.changes) == 3
|
|
|
|
change_map = {c.resource_id: c for c in summary.changes}
|
|
assert change_map["new-res"].change_type == ChangeType.ADDED
|
|
assert change_map["old-res"].change_type == ChangeType.REMOVED
|
|
assert change_map["shared"].change_type == ChangeType.MODIFIED
|
|
|
|
|
|
class TestFirstScan:
|
|
"""Tests for first scan (no previous snapshot)."""
|
|
|
|
def test_first_scan_treats_all_as_added(self) -> None:
|
|
"""When previous is None, all current resources are classified as ADDED."""
|
|
res1 = _make_resource(unique_id="res-1", name="service-1")
|
|
res2 = _make_resource(unique_id="res-2", name="service-2")
|
|
current = _make_scan_result(resources=[res1, res2])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous=None)
|
|
|
|
assert summary.added_count == 2
|
|
assert summary.removed_count == 0
|
|
assert summary.modified_count == 0
|
|
assert len(summary.changes) == 2
|
|
assert all(c.change_type == ChangeType.ADDED for c in summary.changes)
|
|
|
|
def test_first_scan_empty_produces_empty_summary(self) -> None:
|
|
"""First scan with no resources produces empty change summary."""
|
|
current = _make_scan_result(resources=[])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous=None)
|
|
|
|
assert summary.added_count == 0
|
|
assert summary.removed_count == 0
|
|
assert summary.modified_count == 0
|
|
assert summary.changes == []
|
|
|
|
|
|
class TestChangeSummaryCounts:
|
|
"""Tests that ChangeSummary counts are always correct."""
|
|
|
|
def test_counts_match_change_list(self) -> None:
|
|
"""The counts in ChangeSummary always match the actual changes list."""
|
|
# Set up a scenario with 2 added, 1 removed, 1 modified
|
|
prev_mod = _make_resource(
|
|
unique_id="mod-1", name="mod-svc", attributes={"port": 80}
|
|
)
|
|
curr_mod = _make_resource(
|
|
unique_id="mod-1", name="mod-svc", attributes={"port": 8080}
|
|
)
|
|
removed = _make_resource(unique_id="rem-1", name="rem-svc")
|
|
added1 = _make_resource(unique_id="add-1", name="add-svc-1")
|
|
added2 = _make_resource(unique_id="add-2", name="add-svc-2")
|
|
|
|
previous = _make_scan_result(resources=[prev_mod, removed])
|
|
current = _make_scan_result(resources=[curr_mod, added1, added2])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
# Verify counts match actual changes
|
|
actual_added = [c for c in summary.changes if c.change_type == ChangeType.ADDED]
|
|
actual_removed = [c for c in summary.changes if c.change_type == ChangeType.REMOVED]
|
|
actual_modified = [c for c in summary.changes if c.change_type == ChangeType.MODIFIED]
|
|
|
|
assert summary.added_count == len(actual_added) == 2
|
|
assert summary.removed_count == len(actual_removed) == 1
|
|
assert summary.modified_count == len(actual_modified) == 1
|
|
|
|
def test_resource_type_preserved_in_changes(self) -> None:
|
|
"""ResourceChange objects preserve the resource_type from the resource."""
|
|
resource = _make_resource(
|
|
unique_id="res-1",
|
|
resource_type="docker_service",
|
|
name="my-service",
|
|
)
|
|
current = _make_scan_result(resources=[resource])
|
|
previous = _make_scan_result(resources=[])
|
|
|
|
detector = ChangeDetector()
|
|
summary = detector.compare(current, previous)
|
|
|
|
assert summary.changes[0].resource_type == "docker_service"
|