514 lines
18 KiB
Python
514 lines
18 KiB
Python
"""Unit tests for the IncrementalUpdater class."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.incremental.incremental_updater import IncrementalUpdater
|
|
from iac_reverse.models import ChangeSummary, ChangeType, ResourceChange
|
|
|
|
|
|
def _make_change(
|
|
resource_id: str = "res-1",
|
|
resource_type: str = "kubernetes_deployment",
|
|
resource_name: str = "nginx",
|
|
change_type: ChangeType = ChangeType.ADDED,
|
|
changed_attributes: dict | None = None,
|
|
) -> ResourceChange:
|
|
"""Create a ResourceChange for testing."""
|
|
return ResourceChange(
|
|
resource_id=resource_id,
|
|
resource_type=resource_type,
|
|
resource_name=resource_name,
|
|
change_type=change_type,
|
|
changed_attributes=changed_attributes,
|
|
)
|
|
|
|
|
|
def _make_summary(changes: list[ResourceChange]) -> ChangeSummary:
|
|
"""Create a ChangeSummary from a list of changes."""
|
|
added = sum(1 for c in changes if c.change_type == ChangeType.ADDED)
|
|
removed = sum(1 for c in changes if c.change_type == ChangeType.REMOVED)
|
|
modified = sum(1 for c in changes if c.change_type == ChangeType.MODIFIED)
|
|
return ChangeSummary(
|
|
added_count=added,
|
|
removed_count=removed,
|
|
modified_count=modified,
|
|
changes=changes,
|
|
)
|
|
|
|
|
|
def _write_tf_file(output_dir: Path, resource_type: str, content: str) -> Path:
|
|
"""Write a .tf file to the output directory."""
|
|
tf_file = output_dir / f"{resource_type}.tf"
|
|
tf_file.write_text(content, encoding="utf-8")
|
|
return tf_file
|
|
|
|
|
|
def _write_state_file(output_dir: Path, resources: list[dict]) -> Path:
|
|
"""Write a terraform.tfstate file to the output directory."""
|
|
state = {
|
|
"version": 4,
|
|
"terraform_version": "1.7.0",
|
|
"serial": 1,
|
|
"lineage": "test-lineage-uuid",
|
|
"outputs": {},
|
|
"resources": resources,
|
|
}
|
|
state_file = output_dir / "terraform.tfstate"
|
|
state_file.write_text(json.dumps(state, indent=2), encoding="utf-8")
|
|
return state_file
|
|
|
|
|
|
class TestAddedResource:
|
|
"""Tests for adding new resource blocks."""
|
|
|
|
def test_added_resource_creates_block_in_correct_file(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""An ADDED resource creates a new block in the resource type .tf file."""
|
|
change = _make_change(
|
|
resource_id="apps/v1/deployments/default/nginx",
|
|
resource_type="kubernetes_deployment",
|
|
resource_name="nginx",
|
|
change_type=ChangeType.ADDED,
|
|
)
|
|
summary = _make_summary([change])
|
|
attributes = {
|
|
"apps/v1/deployments/default/nginx": {
|
|
"namespace": "default",
|
|
"replicas": 3,
|
|
}
|
|
}
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path), attributes)
|
|
updater.apply()
|
|
|
|
tf_file = tmp_path / "kubernetes_deployment.tf"
|
|
assert tf_file.exists()
|
|
content = tf_file.read_text(encoding="utf-8")
|
|
assert 'resource "kubernetes_deployment" "nginx"' in content
|
|
assert "namespace" in content
|
|
assert "replicas" in content
|
|
assert "# Source: apps/v1/deployments/default/nginx" in content
|
|
|
|
def test_added_resource_appends_to_existing_file(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""An ADDED resource appends to an existing .tf file without overwriting."""
|
|
existing_content = (
|
|
'# Source: existing-id\n'
|
|
'resource "kubernetes_deployment" "existing" {\n'
|
|
' replicas = 1\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "kubernetes_deployment", existing_content)
|
|
|
|
change = _make_change(
|
|
resource_id="new-id",
|
|
resource_type="kubernetes_deployment",
|
|
resource_name="new-service",
|
|
change_type=ChangeType.ADDED,
|
|
)
|
|
summary = _make_summary([change])
|
|
attributes = {"new-id": {"replicas": 2}}
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path), attributes)
|
|
updater.apply()
|
|
|
|
tf_file = tmp_path / "kubernetes_deployment.tf"
|
|
content = tf_file.read_text(encoding="utf-8")
|
|
# Both resources should be present
|
|
assert 'resource "kubernetes_deployment" "existing"' in content
|
|
assert 'resource "kubernetes_deployment" "new_service"' in content
|
|
|
|
def test_added_resource_creates_file_if_not_exists(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""An ADDED resource creates the .tf file if it doesn't exist."""
|
|
change = _make_change(
|
|
resource_id="svc-1",
|
|
resource_type="docker_service",
|
|
resource_name="my-app",
|
|
change_type=ChangeType.ADDED,
|
|
)
|
|
summary = _make_summary([change])
|
|
attributes = {"svc-1": {"image": "nginx:latest"}}
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path), attributes)
|
|
updater.apply()
|
|
|
|
tf_file = tmp_path / "docker_service.tf"
|
|
assert tf_file.exists()
|
|
content = tf_file.read_text(encoding="utf-8")
|
|
assert 'resource "docker_service" "my_app"' in content
|
|
|
|
|
|
class TestRemovedResource:
|
|
"""Tests for removing resource blocks."""
|
|
|
|
def test_removed_resource_removes_block_from_file(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""A REMOVED resource removes its block from the .tf file."""
|
|
content = (
|
|
'# Source: res-1\n'
|
|
'resource "kubernetes_deployment" "nginx" {\n'
|
|
' replicas = 3\n'
|
|
'}\n'
|
|
'\n'
|
|
'# Source: res-2\n'
|
|
'resource "kubernetes_deployment" "redis" {\n'
|
|
' replicas = 1\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "kubernetes_deployment", content)
|
|
|
|
change = _make_change(
|
|
resource_id="res-1",
|
|
resource_type="kubernetes_deployment",
|
|
resource_name="nginx",
|
|
change_type=ChangeType.REMOVED,
|
|
)
|
|
summary = _make_summary([change])
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path))
|
|
updater.apply()
|
|
|
|
tf_file = tmp_path / "kubernetes_deployment.tf"
|
|
result = tf_file.read_text(encoding="utf-8")
|
|
assert "nginx" not in result
|
|
assert 'resource "kubernetes_deployment" "redis"' in result
|
|
|
|
def test_removed_resource_updates_state_file(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""A REMOVED resource removes its entry from the state file."""
|
|
state_resources = [
|
|
{
|
|
"mode": "managed",
|
|
"type": "kubernetes_deployment",
|
|
"name": "nginx",
|
|
"provider": 'provider["registry.terraform.io/hashicorp/kubernetes"]',
|
|
"instances": [
|
|
{
|
|
"schema_version": 1,
|
|
"attributes": {"id": "res-1", "replicas": 3},
|
|
"sensitive_attributes": [],
|
|
"dependencies": [],
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"mode": "managed",
|
|
"type": "kubernetes_deployment",
|
|
"name": "redis",
|
|
"provider": 'provider["registry.terraform.io/hashicorp/kubernetes"]',
|
|
"instances": [
|
|
{
|
|
"schema_version": 1,
|
|
"attributes": {"id": "res-2", "replicas": 1},
|
|
"sensitive_attributes": [],
|
|
"dependencies": [],
|
|
}
|
|
],
|
|
},
|
|
]
|
|
_write_state_file(tmp_path, state_resources)
|
|
|
|
# Also write the .tf file so the removal can proceed
|
|
tf_content = (
|
|
'# Source: res-1\n'
|
|
'resource "kubernetes_deployment" "nginx" {\n'
|
|
' replicas = 3\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "kubernetes_deployment", tf_content)
|
|
|
|
change = _make_change(
|
|
resource_id="res-1",
|
|
resource_type="kubernetes_deployment",
|
|
resource_name="nginx",
|
|
change_type=ChangeType.REMOVED,
|
|
)
|
|
summary = _make_summary([change])
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path))
|
|
updater.apply()
|
|
|
|
state_file = tmp_path / "terraform.tfstate"
|
|
state = json.loads(state_file.read_text(encoding="utf-8"))
|
|
# Only redis should remain
|
|
assert len(state["resources"]) == 1
|
|
assert state["resources"][0]["name"] == "redis"
|
|
# Serial should be incremented
|
|
assert state["serial"] == 2
|
|
|
|
|
|
class TestModifiedResource:
|
|
"""Tests for updating resource blocks."""
|
|
|
|
def test_modified_resource_updates_block_in_file(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""A MODIFIED resource updates the attribute values in the .tf file."""
|
|
content = (
|
|
'# Source: res-1\n'
|
|
'resource "kubernetes_deployment" "nginx" {\n'
|
|
' replicas = 3\n'
|
|
' image = "nginx:1.24"\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "kubernetes_deployment", content)
|
|
|
|
change = _make_change(
|
|
resource_id="res-1",
|
|
resource_type="kubernetes_deployment",
|
|
resource_name="nginx",
|
|
change_type=ChangeType.MODIFIED,
|
|
changed_attributes={
|
|
"replicas": {"old": 3, "new": 5},
|
|
},
|
|
)
|
|
summary = _make_summary([change])
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path))
|
|
updater.apply()
|
|
|
|
tf_file = tmp_path / "kubernetes_deployment.tf"
|
|
result = tf_file.read_text(encoding="utf-8")
|
|
assert "replicas = 5" in result
|
|
assert "replicas = 3" not in result
|
|
# Unchanged attribute should remain
|
|
assert 'image = "nginx:1.24"' in result
|
|
|
|
def test_modified_resource_adds_new_attribute(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""A MODIFIED resource with a new attribute adds it to the block."""
|
|
content = (
|
|
'# Source: res-1\n'
|
|
'resource "kubernetes_deployment" "nginx" {\n'
|
|
' replicas = 3\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "kubernetes_deployment", content)
|
|
|
|
change = _make_change(
|
|
resource_id="res-1",
|
|
resource_type="kubernetes_deployment",
|
|
resource_name="nginx",
|
|
change_type=ChangeType.MODIFIED,
|
|
changed_attributes={
|
|
"image": {"old": None, "new": "nginx:1.25"},
|
|
},
|
|
)
|
|
summary = _make_summary([change])
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path))
|
|
updater.apply()
|
|
|
|
tf_file = tmp_path / "kubernetes_deployment.tf"
|
|
result = tf_file.read_text(encoding="utf-8")
|
|
assert '"nginx:1.25"' in result
|
|
assert "replicas = 3" in result
|
|
|
|
def test_modified_resource_removes_attribute(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""A MODIFIED resource with a removed attribute removes the line."""
|
|
content = (
|
|
'# Source: res-1\n'
|
|
'resource "kubernetes_deployment" "nginx" {\n'
|
|
' replicas = 3\n'
|
|
' image = "nginx:1.24"\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "kubernetes_deployment", content)
|
|
|
|
change = _make_change(
|
|
resource_id="res-1",
|
|
resource_type="kubernetes_deployment",
|
|
resource_name="nginx",
|
|
change_type=ChangeType.MODIFIED,
|
|
changed_attributes={
|
|
"image": {"old": "nginx:1.24", "new": None},
|
|
},
|
|
)
|
|
summary = _make_summary([change])
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path))
|
|
updater.apply()
|
|
|
|
tf_file = tmp_path / "kubernetes_deployment.tf"
|
|
result = tf_file.read_text(encoding="utf-8")
|
|
assert "image" not in result
|
|
assert "replicas = 3" in result
|
|
|
|
|
|
class TestOnlyAffectedFilesModified:
|
|
"""Tests that only files with changed resources are modified."""
|
|
|
|
def test_unrelated_files_are_not_modified(self, tmp_path: Path) -> None:
|
|
"""Files for resource types without changes are not touched."""
|
|
# Write two .tf files
|
|
k8s_content = (
|
|
'# Source: res-1\n'
|
|
'resource "kubernetes_deployment" "nginx" {\n'
|
|
' replicas = 3\n'
|
|
'}\n'
|
|
)
|
|
docker_content = (
|
|
'# Source: svc-1\n'
|
|
'resource "docker_service" "app" {\n'
|
|
' image = "app:latest"\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "kubernetes_deployment", k8s_content)
|
|
docker_file = _write_tf_file(tmp_path, "docker_service", docker_content)
|
|
|
|
# Record the modification time of the docker file
|
|
docker_mtime_before = docker_file.stat().st_mtime
|
|
|
|
# Only change the kubernetes resource
|
|
change = _make_change(
|
|
resource_id="res-1",
|
|
resource_type="kubernetes_deployment",
|
|
resource_name="nginx",
|
|
change_type=ChangeType.MODIFIED,
|
|
changed_attributes={"replicas": {"old": 3, "new": 5}},
|
|
)
|
|
summary = _make_summary([change])
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path))
|
|
updater.apply()
|
|
|
|
# Docker file should not be in modified_files
|
|
assert str(docker_file) not in updater.modified_files
|
|
# Kubernetes file should be in modified_files
|
|
k8s_file = tmp_path / "kubernetes_deployment.tf"
|
|
assert str(k8s_file) in updater.modified_files
|
|
|
|
def test_modified_files_tracks_only_changed(self, tmp_path: Path) -> None:
|
|
"""The modified_files property only contains files that were changed."""
|
|
content = (
|
|
'# Source: res-1\n'
|
|
'resource "kubernetes_deployment" "nginx" {\n'
|
|
' replicas = 3\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "kubernetes_deployment", content)
|
|
|
|
change = _make_change(
|
|
resource_id="new-id",
|
|
resource_type="docker_service",
|
|
resource_name="new-svc",
|
|
change_type=ChangeType.ADDED,
|
|
)
|
|
summary = _make_summary([change])
|
|
attributes = {"new-id": {"image": "app:1.0"}}
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path), attributes)
|
|
updater.apply()
|
|
|
|
# Only docker_service.tf should be modified
|
|
modified = updater.modified_files
|
|
assert any("docker_service.tf" in f for f in modified)
|
|
assert not any("kubernetes_deployment.tf" in f for f in modified)
|
|
|
|
|
|
class TestStateFileUpdatedForRemovedResources:
|
|
"""Tests that state file is properly updated when resources are removed."""
|
|
|
|
def test_state_entry_removed_for_removed_resource(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""Removing a resource also removes its state entry."""
|
|
state_resources = [
|
|
{
|
|
"mode": "managed",
|
|
"type": "docker_service",
|
|
"name": "my_app",
|
|
"provider": 'provider["registry.terraform.io/hashicorp/docker"]',
|
|
"instances": [
|
|
{
|
|
"schema_version": 0,
|
|
"attributes": {"id": "svc-1", "image": "app:1.0"},
|
|
"sensitive_attributes": [],
|
|
"dependencies": [],
|
|
}
|
|
],
|
|
}
|
|
]
|
|
_write_state_file(tmp_path, state_resources)
|
|
|
|
tf_content = (
|
|
'# Source: svc-1\n'
|
|
'resource "docker_service" "my_app" {\n'
|
|
' image = "app:1.0"\n'
|
|
'}\n'
|
|
)
|
|
_write_tf_file(tmp_path, "docker_service", tf_content)
|
|
|
|
change = _make_change(
|
|
resource_id="svc-1",
|
|
resource_type="docker_service",
|
|
resource_name="my-app",
|
|
change_type=ChangeType.REMOVED,
|
|
)
|
|
summary = _make_summary([change])
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path))
|
|
updater.apply()
|
|
|
|
state_file = tmp_path / "terraform.tfstate"
|
|
state = json.loads(state_file.read_text(encoding="utf-8"))
|
|
assert len(state["resources"]) == 0
|
|
assert state["serial"] == 2
|
|
|
|
def test_state_file_not_modified_when_no_removals(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""State file is not modified when there are no REMOVED changes."""
|
|
state_resources = [
|
|
{
|
|
"mode": "managed",
|
|
"type": "kubernetes_deployment",
|
|
"name": "nginx",
|
|
"provider": 'provider["registry.terraform.io/hashicorp/kubernetes"]',
|
|
"instances": [
|
|
{
|
|
"schema_version": 1,
|
|
"attributes": {"id": "res-1"},
|
|
"sensitive_attributes": [],
|
|
"dependencies": [],
|
|
}
|
|
],
|
|
}
|
|
]
|
|
_write_state_file(tmp_path, state_resources)
|
|
|
|
# Only an ADDED change (no removal)
|
|
change = _make_change(
|
|
resource_id="new-id",
|
|
resource_type="docker_service",
|
|
resource_name="new-svc",
|
|
change_type=ChangeType.ADDED,
|
|
)
|
|
summary = _make_summary([change])
|
|
attributes = {"new-id": {"image": "app:1.0"}}
|
|
|
|
updater = IncrementalUpdater(summary, str(tmp_path), attributes)
|
|
updater.apply()
|
|
|
|
# State file should not be in modified_files
|
|
state_path = str(tmp_path / "terraform.tfstate")
|
|
assert state_path not in updater.modified_files
|
|
|
|
# State should still have the original entry
|
|
state_file = tmp_path / "terraform.tfstate"
|
|
state = json.loads(state_file.read_text(encoding="utf-8"))
|
|
assert len(state["resources"]) == 1
|
|
assert state["serial"] == 1
|