309 lines
11 KiB
Python
309 lines
11 KiB
Python
"""Property-based tests for drift report correctness.
|
|
|
|
**Validates: Requirements 7.3**
|
|
|
|
Properties tested:
|
|
- Property 22: Drift report correctness — For any terraform plan output
|
|
containing planned changes, the Validator SHALL report each change with
|
|
the correct resource address and change type (add, modify, destroy).
|
|
"""
|
|
|
|
import json
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from hypothesis import given, settings, assume
|
|
from hypothesis import strategies as st
|
|
|
|
from iac_reverse.models import PlannedChange, ValidationResult
|
|
from iac_reverse.validator import Validator
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hypothesis Strategies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Terraform action types that map to our change types
|
|
TERRAFORM_ACTIONS = ["create", "update", "delete"]
|
|
|
|
# Expected mapping from terraform actions to our change types
|
|
ACTION_TO_CHANGE_TYPE = {
|
|
"create": "add",
|
|
"update": "modify",
|
|
"delete": "destroy",
|
|
}
|
|
|
|
# Strategy for valid terraform resource addresses
|
|
# Format: <resource_type>.<resource_name> or <module>.<resource_type>.<name>
|
|
resource_type_prefix_strategy = st.sampled_from([
|
|
"aws_instance",
|
|
"kubernetes_deployment",
|
|
"docker_service",
|
|
"harvester_virtualmachine",
|
|
"synology_shared_folder",
|
|
"windows_service",
|
|
"bare_metal_hardware",
|
|
"null_resource",
|
|
"local_file",
|
|
"random_id",
|
|
])
|
|
|
|
resource_name_suffix_strategy = st.text(
|
|
min_size=1,
|
|
max_size=20,
|
|
alphabet=st.characters(whitelist_categories=("Ll",), whitelist_characters="_"),
|
|
).filter(lambda s: s[0].isalpha() or s[0] == "_")
|
|
|
|
|
|
@st.composite
|
|
def resource_address_strategy(draw):
|
|
"""Generate a valid terraform resource address like 'aws_instance.my_server'."""
|
|
prefix = draw(resource_type_prefix_strategy)
|
|
suffix = draw(resource_name_suffix_strategy)
|
|
# Optionally add a module prefix
|
|
use_module = draw(st.booleans())
|
|
if use_module:
|
|
module_name = draw(st.text(
|
|
min_size=1,
|
|
max_size=10,
|
|
alphabet=st.characters(whitelist_categories=("Ll",), whitelist_characters="_"),
|
|
).filter(lambda s: s[0].isalpha()))
|
|
return f"module.{module_name}.{prefix}.{suffix}"
|
|
return f"{prefix}.{suffix}"
|
|
|
|
|
|
terraform_action_strategy = st.sampled_from(TERRAFORM_ACTIONS)
|
|
|
|
|
|
@st.composite
|
|
def planned_change_entry_strategy(draw):
|
|
"""Generate a single planned change entry as it appears in terraform plan JSON output."""
|
|
addr = draw(resource_address_strategy())
|
|
action = draw(terraform_action_strategy)
|
|
return (addr, action)
|
|
|
|
|
|
@st.composite
|
|
def planned_changes_list_strategy(draw):
|
|
"""Generate a list of planned changes with unique resource addresses."""
|
|
num_changes = draw(st.integers(min_value=1, max_value=10))
|
|
changes = []
|
|
seen_addrs = set()
|
|
|
|
for _ in range(num_changes):
|
|
entry = draw(planned_change_entry_strategy())
|
|
addr, action = entry
|
|
# Ensure unique addresses
|
|
if addr in seen_addrs:
|
|
continue
|
|
seen_addrs.add(addr)
|
|
changes.append((addr, action))
|
|
|
|
assume(len(changes) >= 1)
|
|
return changes
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
VALIDATE_SUCCESS_JSON = json.dumps(
|
|
{"valid": True, "error_count": 0, "diagnostics": []}
|
|
)
|
|
|
|
|
|
def _make_completed_process(returncode=0, stdout="", stderr=""):
|
|
"""Create a mock CompletedProcess-like object."""
|
|
mock = MagicMock()
|
|
mock.returncode = returncode
|
|
mock.stdout = stdout
|
|
mock.stderr = stderr
|
|
return mock
|
|
|
|
|
|
def build_plan_output(changes: list[tuple[str, str]]) -> str:
|
|
"""Build terraform plan JSON streaming output from a list of (addr, action) tuples."""
|
|
lines = [json.dumps({"type": "version", "terraform": "1.7.0"})]
|
|
|
|
for addr, action in changes:
|
|
lines.append(
|
|
json.dumps(
|
|
{
|
|
"type": "planned_change",
|
|
"change": {
|
|
"resource": {"addr": addr},
|
|
"action": action,
|
|
},
|
|
}
|
|
)
|
|
)
|
|
|
|
# Add change_summary
|
|
total_add = sum(1 for _, a in changes if a == "create")
|
|
total_change = sum(1 for _, a in changes if a == "update")
|
|
total_remove = sum(1 for _, a in changes if a == "delete")
|
|
lines.append(
|
|
json.dumps(
|
|
{
|
|
"type": "change_summary",
|
|
"changes": {
|
|
"add": total_add,
|
|
"change": total_change,
|
|
"remove": total_remove,
|
|
},
|
|
}
|
|
)
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def run_validator_with_plan(plan_output: str) -> ValidationResult:
|
|
"""Run the Validator with mocked subprocess calls, returning the result."""
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=0, stdout=VALIDATE_SUCCESS_JSON
|
|
)
|
|
plan_result = _make_completed_process(returncode=2, stdout=plan_output)
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run",
|
|
side_effect=[init_result, validate_result, plan_result],
|
|
):
|
|
validator = Validator()
|
|
return validator.validate(tmp_dir)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 22: Drift report correctness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDriftReportCorrectness:
|
|
"""Property 22: Drift report correctness.
|
|
|
|
**Validates: Requirements 7.3**
|
|
|
|
For any terraform plan output containing N planned changes, the drift
|
|
report SHALL list exactly N entries, each with the correct resource
|
|
address and change type (add, modify, or destroy).
|
|
"""
|
|
|
|
@given(changes=planned_changes_list_strategy())
|
|
@settings(max_examples=100)
|
|
def test_drift_report_count_matches_planned_changes(
|
|
self, changes: list[tuple[str, str]]
|
|
):
|
|
"""The number of reported planned changes equals the number in the plan output."""
|
|
plan_output = build_plan_output(changes)
|
|
result = run_validator_with_plan(plan_output)
|
|
|
|
assert len(result.planned_changes) == len(changes), (
|
|
f"Expected {len(changes)} planned changes, "
|
|
f"got {len(result.planned_changes)}. "
|
|
f"Input changes: {changes}"
|
|
)
|
|
|
|
@given(changes=planned_changes_list_strategy())
|
|
@settings(max_examples=100)
|
|
def test_drift_report_resource_addresses_match(
|
|
self, changes: list[tuple[str, str]]
|
|
):
|
|
"""Each reported change has the correct resource address from the plan."""
|
|
plan_output = build_plan_output(changes)
|
|
result = run_validator_with_plan(plan_output)
|
|
|
|
expected_addrs = {addr for addr, _ in changes}
|
|
actual_addrs = {c.resource_address for c in result.planned_changes}
|
|
|
|
assert actual_addrs == expected_addrs, (
|
|
f"Resource address mismatch.\n"
|
|
f"Expected: {sorted(expected_addrs)}\n"
|
|
f"Actual: {sorted(actual_addrs)}"
|
|
)
|
|
|
|
@given(changes=planned_changes_list_strategy())
|
|
@settings(max_examples=100)
|
|
def test_drift_report_change_types_correct(
|
|
self, changes: list[tuple[str, str]]
|
|
):
|
|
"""Each reported change has the correct change type mapping."""
|
|
plan_output = build_plan_output(changes)
|
|
result = run_validator_with_plan(plan_output)
|
|
|
|
# Build expected mapping: addr -> change_type
|
|
expected_map = {
|
|
addr: ACTION_TO_CHANGE_TYPE[action] for addr, action in changes
|
|
}
|
|
|
|
for planned_change in result.planned_changes:
|
|
addr = planned_change.resource_address
|
|
assert addr in expected_map, (
|
|
f"Unexpected resource address '{addr}' in planned changes"
|
|
)
|
|
expected_type = expected_map[addr]
|
|
assert planned_change.change_type == expected_type, (
|
|
f"For resource '{addr}': expected change_type='{expected_type}', "
|
|
f"got '{planned_change.change_type}'"
|
|
)
|
|
|
|
@given(changes=planned_changes_list_strategy())
|
|
@settings(max_examples=100)
|
|
def test_drift_report_plan_success_is_false(
|
|
self, changes: list[tuple[str, str]]
|
|
):
|
|
"""When there are planned changes, plan_success is always False."""
|
|
plan_output = build_plan_output(changes)
|
|
result = run_validator_with_plan(plan_output)
|
|
|
|
assert result.plan_success is False, (
|
|
f"plan_success should be False when there are {len(changes)} "
|
|
f"planned changes, but got True"
|
|
)
|
|
|
|
@given(changes=planned_changes_list_strategy())
|
|
@settings(max_examples=100)
|
|
def test_drift_report_each_change_is_planned_change_instance(
|
|
self, changes: list[tuple[str, str]]
|
|
):
|
|
"""Each entry in the drift report is a PlannedChange instance."""
|
|
plan_output = build_plan_output(changes)
|
|
result = run_validator_with_plan(plan_output)
|
|
|
|
for i, change in enumerate(result.planned_changes):
|
|
assert isinstance(change, PlannedChange), (
|
|
f"Entry {i} is {type(change).__name__}, expected PlannedChange"
|
|
)
|
|
|
|
@given(changes=planned_changes_list_strategy())
|
|
@settings(max_examples=100)
|
|
def test_drift_report_change_type_in_valid_set(
|
|
self, changes: list[tuple[str, str]]
|
|
):
|
|
"""Every reported change_type is one of 'add', 'modify', or 'destroy'."""
|
|
plan_output = build_plan_output(changes)
|
|
result = run_validator_with_plan(plan_output)
|
|
|
|
valid_types = {"add", "modify", "destroy"}
|
|
for change in result.planned_changes:
|
|
assert change.change_type in valid_types, (
|
|
f"Invalid change_type '{change.change_type}' for resource "
|
|
f"'{change.resource_address}'. Must be one of {valid_types}"
|
|
)
|
|
|
|
@given(changes=planned_changes_list_strategy())
|
|
@settings(max_examples=100)
|
|
def test_drift_report_no_duplicate_addresses(
|
|
self, changes: list[tuple[str, str]]
|
|
):
|
|
"""No resource address appears more than once in the drift report."""
|
|
plan_output = build_plan_output(changes)
|
|
result = run_validator_with_plan(plan_output)
|
|
|
|
addresses = [c.resource_address for c in result.planned_changes]
|
|
assert len(addresses) == len(set(addresses)), (
|
|
f"Duplicate resource addresses found in drift report: "
|
|
f"{[a for a in addresses if addresses.count(a) > 1]}"
|
|
)
|