shoc-backend/scripts/test-terraform-import-plan-check.py

#!/usr/bin/env python3
"""Deterministic tests for check-terraform-import-plan.py."""

from __future__ import annotations

import json
import subprocess
import sys
import tempfile
from pathlib import Path

from terraform_import_plan_resources import REQUIRED_RESOURCES

SCRIPT = Path(__file__).with_name("check-terraform-import-plan.py")


def run_case(
    environment: str,
    *,
    actions_by_address: dict[str, list[str]] | None = None,
    omit_address: str | None = None,
    extra_resource: tuple[str, str, list[str]] | None = None,
    allowed_updates: tuple[str, ...] = (),
    empty: bool = False,
) -> subprocess.CompletedProcess[str]:
    changes = []
    if not empty:
        for address, resource_type in REQUIRED_RESOURCES[environment].items():
            if address == omit_address:
                continue
            actions = (actions_by_address or {}).get(address, ["no-op"])
            changes.append(
                {
                    "address": address,
                    "mode": "managed",
                    "type": resource_type,
                    "change": {"actions": actions},
                }
            )
    if extra_resource:
        address, resource_type, actions = extra_resource
        changes.append(
            {
                "address": address,
                "mode": "managed",
                "type": resource_type,
                "change": {"actions": actions},
            }
        )

    with tempfile.TemporaryDirectory() as directory:
        plan_path = Path(directory) / "plan.json"
        plan_path.write_text(
            json.dumps({"resource_changes": changes}), encoding="utf-8"
        )
        command = [
            sys.executable,
            str(SCRIPT),
            str(plan_path),
            "--environment",
            environment,
        ]
        for allowed_address in allowed_updates:
            command.extend(["--allow-update-address", allowed_address])
        return subprocess.run(command, check=False, capture_output=True, text=True)


def main() -> int:
    controlled_address = "module.environment.aws_iam_role.github_deploy"
    cases = [
        *[
            (f"{environment} no-op", run_case(environment), 0)
            for environment in REQUIRED_RESOURCES
        ],
        ("empty", run_case("dev", empty=True), 1),
        (
            "missing required",
            run_case("dev", omit_address=controlled_address),
            1,
        ),
        (
            "initial update",
            run_case("dev", actions_by_address={controlled_address: ["update"]}),
            1,
        ),
        (
            "controlled update",
            run_case(
                "dev",
                actions_by_address={controlled_address: ["update"]},
                allowed_updates=(controlled_address,),
            ),
            0,
        ),
        (
            "tf-poc controlled update",
            run_case(
                "tf-poc",
                actions_by_address={controlled_address: ["update"]},
                allowed_updates=(controlled_address,),
            ),
            0,
        ),
        (
            "wrong controlled address",
            run_case(
                "dev",
                actions_by_address={controlled_address: ["update"]},
                allowed_updates=("module.environment.aws_iam_role.runtime",),
            ),
            1,
        ),
        (
            "create",
            run_case("dev", actions_by_address={controlled_address: ["create"]}),
            1,
        ),
        (
            "replacement",
            run_case(
                "dev",
                actions_by_address={controlled_address: ["delete", "create"]},
            ),
            1,
        ),
        (
            "destroy",
            run_case("dev", actions_by_address={controlled_address: ["delete"]}),
            1,
        ),
        (
            "outside address",
            run_case(
                "dev",
                extra_resource=(
                    "module.environment.aws_iam_role.other",
                    "aws_iam_role",
                    ["no-op"],
                ),
            ),
            1,
        ),
        (
            "wrong type",
            run_case(
                "dev",
                extra_resource=(controlled_address, "aws_iam_role_policy", ["no-op"]),
            ),
            1,
        ),
        (
            "dev resource in staging",
            run_case(
                "staging",
                extra_resource=(
                    "module.environment.aws_iam_role_policy.runtime_dynamo[0]",
                    "aws_iam_role_policy",
                    ["no-op"],
                ),
            ),
            1,
        ),
        (
            "live webhook policy in tf-poc",
            run_case(
                "tf-poc",
                extra_resource=(
                    "module.environment.aws_iam_role_policy.runtime_webhook[0]",
                    "aws_iam_role_policy",
                    ["no-op"],
                ),
            ),
            1,
        ),
    ]
    failures = [
        (name, result, expected)
        for name, result, expected in cases
        if result.returncode != expected
    ]
    if failures:
        print(
            "FAIL: plan-check cases failed: "
            + ", ".join(name for name, _, _ in failures),
            file=sys.stderr,
        )
        for name, result, expected in failures:
            print(
                f"{name}: expected {expected}, got {result.returncode}\n"
                f"{result.stdout}{result.stderr}",
                file=sys.stderr,
            )
        return 1
    print("PASS: Terraform import plan safety checks")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())