469 lines
17 KiB
Python
469 lines
17 KiB
Python
"""Unit tests for unmapped resource handling in StateBuilder.
|
|
|
|
Tests that resources with missing provider-assigned identifiers or
|
|
unrecognized resource types are excluded from the state file, warnings
|
|
are logged, and the unmapped resources list is correctly populated.
|
|
"""
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.models import (
|
|
CodeGenerationResult,
|
|
CpuArchitecture,
|
|
DependencyGraph,
|
|
DiscoveredResource,
|
|
GeneratedFile,
|
|
PlatformCategory,
|
|
ProviderType,
|
|
ResourceRelationship,
|
|
)
|
|
from iac_reverse.state_builder import StateBuilder
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers / Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def make_resource(
|
|
resource_type: str = "kubernetes_deployment",
|
|
unique_id: str = "apps/v1/deployments/default/nginx",
|
|
name: str = "nginx",
|
|
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 {"replicas": 3, "image": "nginx:1.25"},
|
|
raw_references=[],
|
|
)
|
|
|
|
|
|
def make_code_generation_result() -> CodeGenerationResult:
|
|
"""Create a minimal CodeGenerationResult for testing."""
|
|
return CodeGenerationResult(
|
|
resource_files=[
|
|
GeneratedFile(
|
|
filename="kubernetes_deployment.tf",
|
|
content='resource "kubernetes_deployment" "nginx" {}',
|
|
resource_count=1,
|
|
)
|
|
],
|
|
variables_file=GeneratedFile(
|
|
filename="variables.tf", content="", resource_count=0
|
|
),
|
|
provider_file=GeneratedFile(
|
|
filename="providers.tf", content="", resource_count=0
|
|
),
|
|
)
|
|
|
|
|
|
def make_dependency_graph(
|
|
resources: list[DiscoveredResource],
|
|
relationships: list[ResourceRelationship] | None = None,
|
|
) -> DependencyGraph:
|
|
"""Create a DependencyGraph from resources and optional relationships."""
|
|
return DependencyGraph(
|
|
resources=resources,
|
|
relationships=relationships or [],
|
|
topological_order=[r.unique_id for r in resources],
|
|
cycles=[],
|
|
unresolved_references=[],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Resource with empty unique_id is excluded from state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEmptyUniqueIdExcluded:
|
|
"""Resources with empty unique_id are excluded from state."""
|
|
|
|
def test_empty_string_unique_id_excluded(self):
|
|
"""A resource with empty string unique_id produces no state entry."""
|
|
resource = make_resource(unique_id="")
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 0
|
|
|
|
def test_whitespace_only_unique_id_excluded(self):
|
|
"""A resource with whitespace-only unique_id produces no state entry."""
|
|
resource = make_resource(unique_id=" ")
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 0
|
|
|
|
def test_tabs_and_newlines_unique_id_excluded(self):
|
|
"""A resource with tabs/newlines-only unique_id is excluded."""
|
|
resource = make_resource(unique_id="\t\n")
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Resource with None-like identifier is excluded
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNoneLikeIdentifierExcluded:
|
|
"""Resources with None-like identifiers are excluded from state."""
|
|
|
|
def test_none_unique_id_excluded(self):
|
|
"""A resource with None unique_id produces no state entry."""
|
|
resource = make_resource()
|
|
# Manually set unique_id to None (bypassing type hints)
|
|
resource.unique_id = None # type: ignore[assignment]
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 0
|
|
|
|
def test_empty_string_is_falsy_excluded(self):
|
|
"""Empty string is falsy and should be excluded."""
|
|
resource = make_resource(unique_id="")
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Unrecognized resource type is excluded
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUnrecognizedResourceTypeExcluded:
|
|
"""Resources with unrecognized resource types are excluded."""
|
|
|
|
def test_unknown_resource_type_excluded(self):
|
|
"""A resource with an unrecognized type produces no state entry."""
|
|
resource = make_resource(
|
|
resource_type="totally_unknown_type",
|
|
unique_id="some/valid/id",
|
|
name="mystery",
|
|
)
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 0
|
|
|
|
def test_misspelled_resource_type_excluded(self):
|
|
"""A misspelled resource type is excluded."""
|
|
resource = make_resource(
|
|
resource_type="kuberntes_deployment", # typo
|
|
unique_id="deploy/nginx",
|
|
name="nginx",
|
|
)
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Warning is logged for unmapped resources
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWarningLogged:
|
|
"""Warnings are logged for unmapped resources."""
|
|
|
|
def test_warning_logged_for_empty_unique_id(self, caplog):
|
|
"""A warning is logged when a resource has empty unique_id."""
|
|
resource = make_resource(
|
|
unique_id="",
|
|
name="orphan-resource",
|
|
resource_type="kubernetes_deployment",
|
|
)
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
with caplog.at_level(logging.WARNING):
|
|
builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert any("orphan-resource" in record.message for record in caplog.records)
|
|
assert any(
|
|
"missing provider-assigned resource identifier" in record.message
|
|
for record in caplog.records
|
|
)
|
|
|
|
def test_warning_logged_for_unrecognized_type(self, caplog):
|
|
"""A warning is logged when a resource has unrecognized type."""
|
|
resource = make_resource(
|
|
resource_type="alien_resource",
|
|
unique_id="some/id",
|
|
name="alien",
|
|
)
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
with caplog.at_level(logging.WARNING):
|
|
builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert any("alien" in record.message for record in caplog.records)
|
|
assert any(
|
|
"not recognized" in record.message for record in caplog.records
|
|
)
|
|
|
|
def test_warning_includes_resource_type_and_name(self, caplog):
|
|
"""Warning message identifies the resource by type and name."""
|
|
resource = make_resource(
|
|
resource_type="unknown_type",
|
|
unique_id="id/123",
|
|
name="my-resource",
|
|
)
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
with caplog.at_level(logging.WARNING):
|
|
builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
# The warning should contain the resource identifier (type.name)
|
|
assert any(
|
|
"unknown_type.my-resource" in record.message
|
|
for record in caplog.records
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Mapped resources still produce valid state entries alongside unmapped
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMappedAlongsideUnmapped:
|
|
"""Mapped resources produce valid entries even when unmapped ones exist."""
|
|
|
|
def test_valid_resource_produces_entry_alongside_unmapped(self):
|
|
"""A valid resource still gets a state entry when others are unmapped."""
|
|
valid_resource = make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="deploy/nginx",
|
|
name="nginx",
|
|
)
|
|
unmapped_resource = make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="", # empty - unmappable
|
|
name="orphan-svc",
|
|
)
|
|
graph = make_dependency_graph([valid_resource, unmapped_resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 1
|
|
assert state.resources[0].resource_type == "kubernetes_deployment"
|
|
assert state.resources[0].provider_id == "deploy/nginx"
|
|
|
|
def test_multiple_valid_resources_with_one_unmapped(self):
|
|
"""Multiple valid resources produce entries; unmapped one is excluded."""
|
|
resources = [
|
|
make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="deploy/nginx",
|
|
name="nginx",
|
|
),
|
|
make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc/redis",
|
|
name="redis",
|
|
),
|
|
make_resource(
|
|
resource_type="unknown_type",
|
|
unique_id="id/mystery",
|
|
name="mystery",
|
|
),
|
|
]
|
|
graph = make_dependency_graph(resources)
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(state.resources) == 2
|
|
types = {e.resource_type for e in state.resources}
|
|
assert "kubernetes_deployment" in types
|
|
assert "kubernetes_service" in types
|
|
assert "unknown_type" not in types
|
|
|
|
def test_valid_entries_have_correct_attributes(self):
|
|
"""Valid entries retain full attributes even when unmapped exist."""
|
|
valid_resource = make_resource(
|
|
resource_type="docker_service",
|
|
unique_id="svc/web",
|
|
name="web",
|
|
attributes={"replicas": 2, "image": "web:latest"},
|
|
provider=ProviderType.DOCKER_SWARM,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
)
|
|
unmapped_resource = make_resource(
|
|
resource_type="docker_service",
|
|
unique_id="",
|
|
name="broken",
|
|
provider=ProviderType.DOCKER_SWARM,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
)
|
|
graph = make_dependency_graph([valid_resource, unmapped_resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
state = builder.build(code_result, graph, provider_version="2.0.0")
|
|
|
|
assert len(state.resources) == 1
|
|
assert state.resources[0].attributes == {"replicas": 2, "image": "web:latest"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Unmapped resources list contains correct entries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUnmappedResourcesList:
|
|
"""The unmapped_resources property contains correct entries."""
|
|
|
|
def test_unmapped_list_empty_when_all_mapped(self):
|
|
"""When all resources are mappable, unmapped list is empty."""
|
|
resource = make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="deploy/nginx",
|
|
name="nginx",
|
|
)
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert builder.unmapped_resources == []
|
|
|
|
def test_unmapped_list_contains_empty_id_resource(self):
|
|
"""Resource with empty unique_id appears in unmapped list."""
|
|
resource = make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="",
|
|
name="orphan",
|
|
)
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(builder.unmapped_resources) == 1
|
|
identifier, reason = builder.unmapped_resources[0]
|
|
assert "kubernetes_deployment.orphan" == identifier
|
|
assert "missing provider-assigned resource identifier" in reason
|
|
|
|
def test_unmapped_list_contains_unrecognized_type_resource(self):
|
|
"""Resource with unrecognized type appears in unmapped list."""
|
|
resource = make_resource(
|
|
resource_type="alien_widget",
|
|
unique_id="widget/123",
|
|
name="my-widget",
|
|
)
|
|
graph = make_dependency_graph([resource])
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(builder.unmapped_resources) == 1
|
|
identifier, reason = builder.unmapped_resources[0]
|
|
assert "alien_widget.my-widget" == identifier
|
|
assert "not recognized" in reason
|
|
|
|
def test_unmapped_list_contains_multiple_entries(self):
|
|
"""Multiple unmapped resources all appear in the list."""
|
|
resources = [
|
|
make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="",
|
|
name="no-id",
|
|
),
|
|
make_resource(
|
|
resource_type="fake_type",
|
|
unique_id="fake/id",
|
|
name="fake",
|
|
),
|
|
make_resource(
|
|
resource_type="kubernetes_service",
|
|
unique_id="svc/valid",
|
|
name="valid",
|
|
),
|
|
]
|
|
graph = make_dependency_graph(resources)
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
builder.build(code_result, graph, provider_version="1.0.0")
|
|
|
|
assert len(builder.unmapped_resources) == 2
|
|
identifiers = [entry[0] for entry in builder.unmapped_resources]
|
|
assert "kubernetes_deployment.no-id" in identifiers
|
|
assert "fake_type.fake" in identifiers
|
|
|
|
def test_unmapped_list_resets_on_new_build(self):
|
|
"""The unmapped list is reset on each new build call."""
|
|
unmapped_resource = make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="",
|
|
name="orphan",
|
|
)
|
|
valid_resource = make_resource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id="deploy/nginx",
|
|
name="nginx",
|
|
)
|
|
code_result = make_code_generation_result()
|
|
|
|
builder = StateBuilder()
|
|
|
|
# First build with unmapped resource
|
|
graph1 = make_dependency_graph([unmapped_resource])
|
|
builder.build(code_result, graph1, provider_version="1.0.0")
|
|
assert len(builder.unmapped_resources) == 1
|
|
|
|
# Second build with valid resource - unmapped list should reset
|
|
graph2 = make_dependency_graph([valid_resource])
|
|
builder.build(code_result, graph2, provider_version="1.0.0")
|
|
assert len(builder.unmapped_resources) == 0
|