Files
SnarfCode/tests/unit/test_validator_autocorrect.py
2026-05-22 00:19:30 -04:00

767 lines
27 KiB
Python

"""Unit tests for the Validator auto-correction loop.
Tests cover:
- Successful correction on first attempt
- Correction after multiple attempts
- Failure after max attempts exhausted
- correction_attempts count is accurate
- Original errors are preserved when correction fails
"""
import json
from unittest.mock import MagicMock, patch
import pytest
from iac_reverse.models import ValidationError, ValidationResult
from iac_reverse.validator import Validator
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
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
VALIDATE_SUCCESS_JSON = json.dumps(
{"valid": True, "error_count": 0, "diagnostics": []}
)
PLAN_NO_CHANGES_JSON = "\n".join(
[
json.dumps({"type": "version", "terraform": "1.7.0"}),
json.dumps(
{
"type": "change_summary",
"changes": {"add": 0, "change": 0, "remove": 0},
}
),
]
)
def _make_validate_error_json(
filename="main.tf", line=10, summary="Unsupported argument", detail="An argument named 'bad_attr' is not expected here."
):
return json.dumps(
{
"valid": False,
"error_count": 1,
"diagnostics": [
{
"severity": "error",
"summary": summary,
"detail": detail,
"range": {
"filename": filename,
"start": {"line": line, "column": 1},
"end": {"line": line, "column": 20},
},
}
],
}
)
def _make_missing_provider_error_json(provider_name="kubernetes"):
return json.dumps(
{
"valid": False,
"error_count": 1,
"diagnostics": [
{
"severity": "error",
"summary": "Missing required provider",
"detail": f"provider \"{provider_name}\" configuration not present",
"range": {
"filename": "main.tf",
"start": {"line": 1, "column": 1},
"end": {"line": 1, "column": 20},
},
}
],
}
)
# ---------------------------------------------------------------------------
# Tests: Successful correction on first attempt
# ---------------------------------------------------------------------------
class TestSuccessfulCorrectionFirstAttempt:
"""Validation fails initially but succeeds after one correction attempt."""
def test_removes_unknown_attribute_and_passes(self, tmp_path):
"""When an unknown attribute error occurs, the offending line is removed
and re-validation succeeds on the first attempt."""
# Create a .tf file with a bad attribute on line 3
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "aws_instance" "example" {\n'
' ami = "ami-123"\n'
' bad_attr = "should be removed"\n'
' instance_type = "t2.micro"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
# First validate fails with unknown attribute on line 3
validate_fail = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf",
line=3,
summary="Unsupported argument",
detail="An argument named 'bad_attr' is not expected here.",
),
)
# Second validate succeeds after correction
validate_success = _make_completed_process(
returncode=0, stdout=VALIDATE_SUCCESS_JSON
)
plan_result = _make_completed_process(
returncode=0, stdout=PLAN_NO_CHANGES_JSON
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[init_result, validate_fail, validate_success, plan_result],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.init_success is True
assert result.validate_success is True
assert result.correction_attempts == 1
# Verify the file was corrected
content = tf_file.read_text()
assert "bad_attr" not in content
assert "ami" in content
assert "instance_type" in content
def test_adds_missing_provider_block_and_passes(self, tmp_path):
"""When a missing provider error occurs, an empty provider block is added
and re-validation succeeds."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "kubernetes_deployment" "app" {\n'
' metadata {\n'
' name = "app"\n'
' }\n'
'}\n'
)
init_result = _make_completed_process(returncode=0)
validate_fail = _make_completed_process(
returncode=1,
stdout=_make_missing_provider_error_json("kubernetes"),
)
validate_success = _make_completed_process(
returncode=0, stdout=VALIDATE_SUCCESS_JSON
)
plan_result = _make_completed_process(
returncode=0, stdout=PLAN_NO_CHANGES_JSON
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[init_result, validate_fail, validate_success, plan_result],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is True
assert result.correction_attempts == 1
# Verify provider block was added
providers_file = tmp_path / "providers.tf"
assert providers_file.exists()
content = providers_file.read_text()
assert 'provider "kubernetes"' in content
# ---------------------------------------------------------------------------
# Tests: Correction after multiple attempts
# ---------------------------------------------------------------------------
class TestCorrectionAfterMultipleAttempts:
"""Validation requires multiple correction attempts before succeeding."""
def test_succeeds_after_two_correction_attempts(self, tmp_path):
"""Two different errors require two correction passes."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "aws_instance" "example" {\n'
' ami = "ami-123"\n'
' bad_attr1 = "remove me"\n'
' instance_type = "t2.micro"\n'
' bad_attr2 = "also remove me"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
# First validate: error on line 3 (bad_attr1)
validate_fail_1 = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf",
line=3,
summary="Unsupported argument",
detail="An argument named 'bad_attr1' is not expected here.",
),
)
# Second validate: error on line 4 (bad_attr2 is now on line 4 after removal)
validate_fail_2 = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf",
line=4,
summary="Unsupported argument",
detail="An argument named 'bad_attr2' is not expected here.",
),
)
# Third validate succeeds
validate_success = _make_completed_process(
returncode=0, stdout=VALIDATE_SUCCESS_JSON
)
plan_result = _make_completed_process(
returncode=0, stdout=PLAN_NO_CHANGES_JSON
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[
init_result,
validate_fail_1,
validate_fail_2,
validate_success,
plan_result,
],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is True
assert result.correction_attempts == 2
def test_succeeds_on_third_attempt(self, tmp_path):
"""Validation succeeds on the third (max) correction attempt."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "aws_instance" "example" {\n'
' ami = "ami-123"\n'
' bad1 = "x"\n'
' bad2 = "y"\n'
' bad3 = "z"\n'
' instance_type = "t2.micro"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
validate_fail_1 = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf", line=3,
summary="Unsupported argument",
detail="An argument named 'bad1' is not expected here.",
),
)
validate_fail_2 = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf", line=3,
summary="Unsupported argument",
detail="An argument named 'bad2' is not expected here.",
),
)
validate_fail_3 = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf", line=3,
summary="Unsupported argument",
detail="An argument named 'bad3' is not expected here.",
),
)
validate_success = _make_completed_process(
returncode=0, stdout=VALIDATE_SUCCESS_JSON
)
plan_result = _make_completed_process(
returncode=0, stdout=PLAN_NO_CHANGES_JSON
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[
init_result,
validate_fail_1,
validate_fail_2,
validate_fail_3,
validate_success,
plan_result,
],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is True
assert result.correction_attempts == 3
# ---------------------------------------------------------------------------
# Tests: Failure after max attempts exhausted
# ---------------------------------------------------------------------------
class TestFailureAfterMaxAttempts:
"""Validation still fails after all correction attempts are exhausted."""
def test_fails_after_max_attempts_with_uncorrectable_error(self, tmp_path):
"""When errors cannot be corrected, fails after max attempts."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "aws_instance" "example" {\n'
' ami = "ami-123"\n'
' bad_attr = "remove me"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
# Each validate returns the same error (correction removes line but
# new errors keep appearing)
validate_fail = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf",
line=3,
summary="Unsupported argument",
detail="An argument named 'bad_attr' is not expected here.",
),
)
# After first correction removes line 3, subsequent validates still fail
# with a different uncorrectable error (no file/line info)
validate_fail_no_fix = _make_completed_process(
returncode=1,
stdout=json.dumps({
"valid": False,
"error_count": 1,
"diagnostics": [
{
"severity": "error",
"summary": "Some complex error",
"detail": "Cannot be auto-corrected",
}
],
}),
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[
init_result,
validate_fail,
validate_fail_no_fix, # After first correction, new error with no file
],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is False
# Stopped after 1 attempt because the second error has no file info
# and cannot be corrected
assert result.correction_attempts >= 1
def test_fails_with_max_3_attempts_default(self, tmp_path):
"""With default max_correction_attempts=3, stops after 3 attempts."""
tf_file = tmp_path / "main.tf"
# Write a file where the error line keeps being valid for removal
tf_file.write_text(
'resource "aws_instance" "example" {\n'
' attr1 = "a"\n'
' attr2 = "b"\n'
' attr3 = "c"\n'
' attr4 = "d"\n'
' attr5 = "e"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
def make_fail(line, attr):
return _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf",
line=line,
summary="Unsupported argument",
detail=f"An argument named '{attr}' is not expected here.",
),
)
# 4 failures: first 3 get corrected, 4th is returned as final failure
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[
init_result,
make_fail(2, "attr1"),
make_fail(2, "attr2"),
make_fail(2, "attr3"),
make_fail(2, "attr4"), # This one exceeds max attempts
],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is False
assert result.correction_attempts == 3
def test_custom_max_attempts_respected(self, tmp_path):
"""Custom max_correction_attempts value is respected."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "aws_instance" "example" {\n'
' attr1 = "a"\n'
' attr2 = "b"\n'
' attr3 = "c"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
def make_fail(line, attr):
return _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf",
line=line,
summary="Unsupported argument",
detail=f"An argument named '{attr}' is not expected here.",
),
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[
init_result,
make_fail(2, "attr1"),
make_fail(2, "attr2"), # Exceeds max_correction_attempts=1
],
):
validator = Validator()
result = validator.validate(str(tmp_path), max_correction_attempts=1)
assert result.validate_success is False
assert result.correction_attempts == 1
# ---------------------------------------------------------------------------
# Tests: correction_attempts count is accurate
# ---------------------------------------------------------------------------
class TestCorrectionAttemptsCount:
"""The correction_attempts field accurately reflects the number of attempts."""
def test_zero_attempts_when_validation_passes_immediately(self, tmp_path):
"""No correction attempts when validation passes on first try."""
init_result = _make_completed_process(returncode=0)
validate_result = _make_completed_process(
returncode=0, stdout=VALIDATE_SUCCESS_JSON
)
plan_result = _make_completed_process(
returncode=0, stdout=PLAN_NO_CHANGES_JSON
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[init_result, validate_result, plan_result],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.correction_attempts == 0
def test_one_attempt_when_first_correction_fixes(self, tmp_path):
"""correction_attempts is 1 when first correction resolves the issue."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "null_resource" "test" {\n'
' unknown_field = "value"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
validate_fail = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf",
line=2,
summary="Unsupported argument",
detail="An argument named 'unknown_field' is not expected here.",
),
)
validate_success = _make_completed_process(
returncode=0, stdout=VALIDATE_SUCCESS_JSON
)
plan_result = _make_completed_process(
returncode=0, stdout=PLAN_NO_CHANGES_JSON
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[init_result, validate_fail, validate_success, plan_result],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.correction_attempts == 1
def test_attempts_count_matches_actual_corrections_applied(self, tmp_path):
"""correction_attempts matches the number of correction loops executed."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "null_resource" "test" {\n'
' bad1 = "x"\n'
' bad2 = "y"\n'
' good = "keep"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
validate_fail_1 = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf", line=2,
summary="Unsupported argument",
detail="An argument named 'bad1' is not expected here.",
),
)
validate_fail_2 = _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf", line=2,
summary="Unsupported argument",
detail="An argument named 'bad2' is not expected here.",
),
)
validate_success = _make_completed_process(
returncode=0, stdout=VALIDATE_SUCCESS_JSON
)
plan_result = _make_completed_process(
returncode=0, stdout=PLAN_NO_CHANGES_JSON
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[
init_result,
validate_fail_1,
validate_fail_2,
validate_success,
plan_result,
],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.correction_attempts == 2
# ---------------------------------------------------------------------------
# Tests: Original errors preserved when correction fails
# ---------------------------------------------------------------------------
class TestOriginalErrorsPreserved:
"""When correction fails, the remaining errors are reported accurately."""
def test_errors_from_final_validation_are_returned(self, tmp_path):
"""After exhausting attempts, the errors from the last validation run
are returned in the result."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "aws_instance" "example" {\n'
' attr1 = "a"\n'
' attr2 = "b"\n'
' attr3 = "c"\n'
' attr4 = "d"\n'
"}\n"
)
init_result = _make_completed_process(returncode=0)
def make_fail(attr):
return _make_completed_process(
returncode=1,
stdout=_make_validate_error_json(
filename="main.tf",
line=2,
summary="Unsupported argument",
detail=f"An argument named '{attr}' is not expected here.",
),
)
# 3 corrections + final failure
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[
init_result,
make_fail("attr1"),
make_fail("attr2"),
make_fail("attr3"),
make_fail("attr4"),
],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is False
assert len(result.errors) >= 1
# The final error should be about attr4
assert "attr4" in result.errors[0].message
def test_uncorrectable_errors_preserved_with_file_info(self, tmp_path):
"""Errors that cannot be corrected retain their file and line info."""
tf_file = tmp_path / "main.tf"
tf_file.write_text('invalid terraform content\n')
init_result = _make_completed_process(returncode=0)
# Error that doesn't match any correction pattern
validate_fail = _make_completed_process(
returncode=1,
stdout=json.dumps({
"valid": False,
"error_count": 1,
"diagnostics": [
{
"severity": "error",
"summary": "Completely unknown error type",
"detail": "This cannot be auto-corrected at all",
"range": {
"filename": "main.tf",
"start": {"line": 1, "column": 1},
"end": {"line": 1, "column": 10},
},
}
],
}),
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[init_result, validate_fail],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is False
assert result.correction_attempts == 0 # No correction was possible
assert len(result.errors) == 1
assert result.errors[0].file == "main.tf"
assert result.errors[0].line == 1
assert "Completely unknown error type" in result.errors[0].message
def test_no_correction_when_error_has_no_file(self, tmp_path):
"""Errors without file information cannot be corrected."""
init_result = _make_completed_process(returncode=0)
validate_fail = _make_completed_process(
returncode=1,
stdout=json.dumps({
"valid": False,
"error_count": 1,
"diagnostics": [
{
"severity": "error",
"summary": "Unsupported argument",
"detail": "An argument named 'foo' is not expected here.",
}
],
}),
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[init_result, validate_fail],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is False
# No file info means no correction can be applied
assert result.correction_attempts == 0
# ---------------------------------------------------------------------------
# Tests: Syntax error correction
# ---------------------------------------------------------------------------
class TestSyntaxErrorCorrection:
"""Tests for syntax error auto-correction."""
def test_trailing_comma_fixed(self, tmp_path):
"""Trailing commas before closing braces are removed."""
tf_file = tmp_path / "main.tf"
tf_file.write_text(
'resource "null_resource" "test" {\n'
' triggers = {\n'
' key = "value",\n'
' }\n'
'}\n'
)
init_result = _make_completed_process(returncode=0)
validate_fail = _make_completed_process(
returncode=1,
stdout=json.dumps({
"valid": False,
"error_count": 1,
"diagnostics": [
{
"severity": "error",
"summary": "Invalid character",
"detail": "trailing comma not allowed",
"range": {
"filename": "main.tf",
"start": {"line": 3, "column": 18},
"end": {"line": 3, "column": 19},
},
}
],
}),
)
validate_success = _make_completed_process(
returncode=0, stdout=VALIDATE_SUCCESS_JSON
)
plan_result = _make_completed_process(
returncode=0, stdout=PLAN_NO_CHANGES_JSON
)
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
"subprocess.run",
side_effect=[init_result, validate_fail, validate_success, plan_result],
):
validator = Validator()
result = validator.validate(str(tmp_path))
assert result.validate_success is True
assert result.correction_attempts == 1
content = tf_file.read_text()
assert ",\n }" not in content