690 lines
25 KiB
Python
690 lines
25 KiB
Python
"""Unit tests for the Terraform Validator.
|
|
|
|
Tests cover:
|
|
- Successful validation (init + validate + plan all pass)
|
|
- Missing terraform binary
|
|
- terraform init failure
|
|
- terraform validate failure with errors
|
|
- terraform plan showing drift (planned changes)
|
|
- Error parsing from terraform JSON output
|
|
"""
|
|
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.models import PlannedChange, 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},
|
|
}
|
|
),
|
|
]
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: missing terraform binary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMissingTerraformBinary:
|
|
def test_returns_failure_result_when_terraform_not_found(self, tmp_path):
|
|
"""When terraform binary is absent, all success flags are False and
|
|
a descriptive error is included."""
|
|
with patch("shutil.which", return_value=None):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert isinstance(result, ValidationResult)
|
|
assert result.init_success is False
|
|
assert result.validate_success is False
|
|
assert result.plan_success is False
|
|
assert len(result.errors) == 1
|
|
error = result.errors[0]
|
|
assert "Terraform" in error.message
|
|
assert "required" in error.message.lower() or "PATH" in error.message
|
|
|
|
def test_no_planned_changes_when_terraform_not_found(self, tmp_path):
|
|
with patch("shutil.which", return_value=None):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert result.planned_changes == []
|
|
|
|
def test_correction_attempts_zero_when_terraform_not_found(self, tmp_path):
|
|
with patch("shutil.which", return_value=None):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert result.correction_attempts == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: terraform init failure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTerraformInitFailure:
|
|
def test_init_failure_returns_correct_flags(self, tmp_path):
|
|
"""When terraform init fails, init_success is False and subsequent
|
|
stages are not run."""
|
|
init_result = _make_completed_process(
|
|
returncode=1, stderr="Error: Failed to install provider"
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", return_value=init_result
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert result.init_success is False
|
|
assert result.validate_success is False
|
|
assert result.plan_success is False
|
|
|
|
def test_init_failure_includes_error_message(self, tmp_path):
|
|
init_result = _make_completed_process(
|
|
returncode=1, stderr="Error: Failed to install provider"
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", return_value=init_result
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert len(result.errors) >= 1
|
|
assert "terraform init failed" in result.errors[0].message.lower()
|
|
assert "Failed to install provider" in result.errors[0].message
|
|
|
|
def test_init_failure_stops_pipeline(self, tmp_path):
|
|
"""After init failure, validate and plan should not be called."""
|
|
init_result = _make_completed_process(returncode=1, stderr="init error")
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", return_value=init_result
|
|
) as mock_run:
|
|
validator = Validator()
|
|
validator.validate(str(tmp_path))
|
|
|
|
# Only one subprocess.run call (for init)
|
|
assert mock_run.call_count == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: successful validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSuccessfulValidation:
|
|
def test_all_flags_true_on_success(self, tmp_path):
|
|
"""When init, validate, and plan all succeed with zero changes,
|
|
all success flags are True."""
|
|
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.init_success is True
|
|
assert result.validate_success is True
|
|
assert result.plan_success is True
|
|
|
|
def test_no_errors_on_success(self, tmp_path):
|
|
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.errors == []
|
|
|
|
def test_no_planned_changes_on_success(self, tmp_path):
|
|
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.planned_changes == []
|
|
|
|
def test_correction_attempts_zero_on_success(self, tmp_path):
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: terraform validate failure with errors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTerraformValidateFailure:
|
|
def _make_validate_error_json(
|
|
self, filename="main.tf", line=10, summary="Invalid attribute", detail="No such attribute"
|
|
):
|
|
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 test_validate_failure_sets_correct_flags(self, tmp_path):
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1, stdout=self._make_validate_error_json()
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", side_effect=[init_result, validate_result]
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert result.init_success is True
|
|
assert result.validate_success is False
|
|
assert result.plan_success is False
|
|
|
|
def test_validate_failure_parses_file_name(self, tmp_path):
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1,
|
|
stdout=self._make_validate_error_json(filename="kubernetes_deployment.tf"),
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", side_effect=[init_result, validate_result]
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert len(result.errors) == 1
|
|
assert result.errors[0].file == "kubernetes_deployment.tf"
|
|
|
|
def test_validate_failure_parses_line_number(self, tmp_path):
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1,
|
|
stdout=self._make_validate_error_json(line=42),
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", side_effect=[init_result, validate_result]
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert result.errors[0].line == 42
|
|
|
|
def test_validate_failure_parses_error_message(self, tmp_path):
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1,
|
|
stdout=self._make_validate_error_json(
|
|
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_result]
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert "Unsupported argument" in result.errors[0].message
|
|
assert "foo" in result.errors[0].message
|
|
|
|
def test_validate_failure_multiple_errors(self, tmp_path):
|
|
validate_json = json.dumps(
|
|
{
|
|
"valid": False,
|
|
"error_count": 2,
|
|
"diagnostics": [
|
|
{
|
|
"severity": "error",
|
|
"summary": "Error one",
|
|
"detail": "",
|
|
"range": {
|
|
"filename": "main.tf",
|
|
"start": {"line": 5, "column": 1},
|
|
},
|
|
},
|
|
{
|
|
"severity": "error",
|
|
"summary": "Error two",
|
|
"detail": "",
|
|
"range": {
|
|
"filename": "variables.tf",
|
|
"start": {"line": 12, "column": 3},
|
|
},
|
|
},
|
|
],
|
|
}
|
|
)
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1, stdout=validate_json
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", side_effect=[init_result, validate_result]
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert len(result.errors) == 2
|
|
assert result.errors[0].file == "main.tf"
|
|
assert result.errors[1].file == "variables.tf"
|
|
|
|
def test_validate_ignores_warning_diagnostics(self, tmp_path):
|
|
"""Only error-severity diagnostics should be included in errors."""
|
|
validate_json = json.dumps(
|
|
{
|
|
"valid": False,
|
|
"error_count": 1,
|
|
"diagnostics": [
|
|
{
|
|
"severity": "warning",
|
|
"summary": "Deprecated attribute",
|
|
"detail": "Use new_attr instead.",
|
|
"range": {
|
|
"filename": "main.tf",
|
|
"start": {"line": 3, "column": 1},
|
|
},
|
|
},
|
|
{
|
|
"severity": "error",
|
|
"summary": "Real error",
|
|
"detail": "",
|
|
"range": {
|
|
"filename": "main.tf",
|
|
"start": {"line": 7, "column": 1},
|
|
},
|
|
},
|
|
],
|
|
}
|
|
)
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1, stdout=validate_json
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", side_effect=[init_result, validate_result]
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert len(result.errors) == 1
|
|
assert result.errors[0].message == "Real error"
|
|
|
|
def test_validate_failure_stops_plan(self, tmp_path):
|
|
"""When validate fails, plan should not be run."""
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1,
|
|
stdout=self._make_validate_error_json(),
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", side_effect=[init_result, validate_result]
|
|
) as mock_run:
|
|
validator = Validator()
|
|
validator.validate(str(tmp_path))
|
|
|
|
# Only init and validate calls, no plan
|
|
assert mock_run.call_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: terraform plan showing drift
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTerraformPlanDrift:
|
|
def _make_plan_with_changes(self, changes):
|
|
"""Build a terraform plan JSON stream with the given changes.
|
|
|
|
changes: 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,
|
|
},
|
|
}
|
|
)
|
|
)
|
|
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 test_plan_with_add_sets_plan_success_false(self, tmp_path):
|
|
plan_output = self._make_plan_with_changes(
|
|
[("kubernetes_deployment.nginx", "create")]
|
|
)
|
|
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 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.plan_success is False
|
|
|
|
def test_plan_with_add_reports_change_type_add(self, tmp_path):
|
|
plan_output = self._make_plan_with_changes(
|
|
[("kubernetes_deployment.nginx", "create")]
|
|
)
|
|
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 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 len(result.planned_changes) == 1
|
|
change = result.planned_changes[0]
|
|
assert change.resource_address == "kubernetes_deployment.nginx"
|
|
assert change.change_type == "add"
|
|
|
|
def test_plan_with_update_reports_change_type_modify(self, tmp_path):
|
|
plan_output = self._make_plan_with_changes(
|
|
[("docker_service.web", "update")]
|
|
)
|
|
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 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.planned_changes[0].change_type == "modify"
|
|
|
|
def test_plan_with_delete_reports_change_type_destroy(self, tmp_path):
|
|
plan_output = self._make_plan_with_changes(
|
|
[("harvester_virtualmachine.dev_vm", "delete")]
|
|
)
|
|
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 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.planned_changes[0].change_type == "destroy"
|
|
|
|
def test_plan_with_multiple_changes(self, tmp_path):
|
|
plan_output = self._make_plan_with_changes(
|
|
[
|
|
("kubernetes_deployment.nginx", "create"),
|
|
("docker_service.web", "update"),
|
|
("harvester_virtualmachine.old_vm", "delete"),
|
|
]
|
|
)
|
|
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 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 len(result.planned_changes) == 3
|
|
addresses = {c.resource_address for c in result.planned_changes}
|
|
assert "kubernetes_deployment.nginx" in addresses
|
|
assert "docker_service.web" in addresses
|
|
assert "harvester_virtualmachine.old_vm" in addresses
|
|
|
|
def test_plan_with_changes_sets_validate_success_true(self, tmp_path):
|
|
"""Drift does not affect validate_success — only plan_success."""
|
|
plan_output = self._make_plan_with_changes(
|
|
[("kubernetes_deployment.nginx", "create")]
|
|
)
|
|
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 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.init_success is True
|
|
assert result.validate_success is True
|
|
assert result.plan_success is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: JSON parsing edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestJsonParsing:
|
|
def test_validate_with_invalid_json_output(self, tmp_path):
|
|
"""When terraform validate returns non-JSON, a parse error is reported."""
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1, stdout="not valid json"
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", side_effect=[init_result, validate_result]
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert result.validate_success is False
|
|
assert len(result.errors) >= 1
|
|
assert any("parse" in e.message.lower() or "json" in e.message.lower() for e in result.errors)
|
|
|
|
def test_validate_error_without_range(self, tmp_path):
|
|
"""Errors without range info should still be parsed with empty file and no line."""
|
|
validate_json = json.dumps(
|
|
{
|
|
"valid": False,
|
|
"error_count": 1,
|
|
"diagnostics": [
|
|
{
|
|
"severity": "error",
|
|
"summary": "No range error",
|
|
"detail": "Something went wrong",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
init_result = _make_completed_process(returncode=0)
|
|
validate_result = _make_completed_process(
|
|
returncode=1, stdout=validate_json
|
|
)
|
|
|
|
with patch("shutil.which", return_value="/usr/bin/terraform"), patch(
|
|
"subprocess.run", side_effect=[init_result, validate_result]
|
|
):
|
|
validator = Validator()
|
|
result = validator.validate(str(tmp_path))
|
|
|
|
assert len(result.errors) == 1
|
|
assert result.errors[0].file == ""
|
|
assert result.errors[0].line is None
|
|
assert "No range error" in result.errors[0].message
|
|
|
|
def test_plan_with_malformed_lines_skipped(self, tmp_path):
|
|
"""Malformed JSON lines in plan output should be skipped gracefully."""
|
|
plan_output = "\n".join(
|
|
[
|
|
"not json",
|
|
json.dumps({"type": "version", "terraform": "1.7.0"}),
|
|
"also not json",
|
|
json.dumps(
|
|
{
|
|
"type": "change_summary",
|
|
"changes": {"add": 0, "change": 0, "remove": 0},
|
|
}
|
|
),
|
|
]
|
|
)
|
|
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_output)
|
|
|
|
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))
|
|
|
|
# Should not raise; plan_success True because no changes
|
|
assert result.plan_success is True
|
|
assert result.planned_changes == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Validator export
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValidatorExport:
|
|
def test_validator_importable_from_package(self):
|
|
from iac_reverse.validator import Validator as V
|
|
|
|
assert V is Validator
|
|
|
|
def test_validator_is_instantiable(self):
|
|
v = Validator()
|
|
assert v is not None
|