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

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

from __future__ import annotations

import importlib.util
import subprocess
import sys
from pathlib import Path
from urllib.request import Request

SCRIPT = Path(__file__).with_name("check-terraform-release-plan.py")
FIXTURES = Path(__file__).with_name("testdata") / "terraform-release-plans"
EXPECTED_LABEL = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-2-1"
PLAN_ID = "plan-8F5JFydVYAmtTjET"


def run_case(
    fixture_name: str,
    *,
    expected_label: str = EXPECTED_LABEL,
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [
            sys.executable,
            str(SCRIPT),
            str(FIXTURES / fixture_name),
            "--expected-version-label",
            expected_label,
        ],
        check=False,
        capture_output=True,
        text=True,
    )


class FakeResponse:
    def __init__(
        self,
        *,
        url: str,
        status: int,
        headers: dict[str, str] | None = None,
        body: bytes = b"",
    ) -> None:
        self.url = url
        self.status = status
        self.headers = headers or {}
        self._body = body

    def read(self) -> bytes:
        return self._body

    def close(self) -> None:
        return None


def load_check_module():
    spec = importlib.util.spec_from_file_location("check_terraform_release_plan", SCRIPT)
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    spec.loader.exec_module(module)
    return module


def test_download_pinning() -> list[str]:
    module = load_check_module()
    fixture = (FIXTURES / "version-only.json").read_bytes()
    archive_url = "https://archivist.terraform.io/v1/object/example"
    calls: list[str] = []

    def fake_urlopen(request: Request, **_kwargs):
        url = request.full_url
        calls.append(url)
        host = request.host if hasattr(request, "host") else ""
        if url.startswith("https://app.terraform.io/api/v2/plans/"):
            if request.get_header("Authorization") != "Bearer test-token":
                raise AssertionError("API request is missing the bearer token")
            if "/runs" in url or "/apply" in url or "/discard" in url:
                raise AssertionError(f"download contacted a run-control path: {url}")
            return FakeResponse(
                url=url,
                status=307,
                headers={"Location": archive_url},
            )
        if url == archive_url:
            if request.get_header("Authorization"):
                raise AssertionError("archivist request must not send TF_API_TOKEN")
            return FakeResponse(url=url, status=200, body=fixture)
        raise AssertionError(f"unexpected URL {url} host={host}")

    plan = module.download_plan_json(PLAN_ID, "test-token", urlopen=fake_urlopen)
    failures: list[str] = []
    if plan["resource_changes"][1]["address"] != (
        "module.environment.aws_elastic_beanstalk_environment.this"
    ):
        failures.append("download did not return the version-only fixture")
    if calls != [
        f"https://app.terraform.io/api/v2/plans/{PLAN_ID}/json-output",
        archive_url,
    ]:
        failures.append(f"download URLs were {calls}")

    try:
        module.download_plan_json("run-not-a-plan", "test-token", urlopen=fake_urlopen)
        failures.append("invalid plan id was accepted")
    except ValueError:
        pass

    def redirect_elsewhere(request: Request, **_kwargs):
        return FakeResponse(
            url=request.full_url,
            status=307,
            headers={"Location": "https://evil.example/plan.json"},
        )

    try:
        module.download_plan_json(PLAN_ID, "test-token", urlopen=redirect_elsewhere)
        failures.append("redirect to a non-archivist host was accepted")
    except ValueError:
        pass

    def double_redirect(request: Request, **_kwargs):
        if request.full_url.startswith("https://app.terraform.io/"):
            return FakeResponse(
                url=request.full_url,
                status=307,
                headers={"Location": archive_url},
            )
        return FakeResponse(
            url=request.full_url,
            status=307,
            headers={"Location": "https://archivist.terraform.io/v1/object/other"},
        )

    try:
        module.download_plan_json(PLAN_ID, "test-token", urlopen=double_redirect)
        failures.append("second archivist redirect was accepted")
    except ValueError:
        pass

    def not_ready(request: Request, **_kwargs):
        return FakeResponse(url=request.full_url, status=204)

    try:
        module.download_plan_json(PLAN_ID, "test-token", urlopen=not_ready)
        failures.append("HTTP 204 was polled or accepted")
    except ValueError as exc:
        if "poll" not in str(exc):
            failures.append(f"HTTP 204 error was {exc}")

    source = SCRIPT.read_text(encoding="utf-8")
    for banned in ("/apply", "/discard", "/runs"):
        if banned in source:
            failures.append(f"download client contains run-control path {banned}")

    return failures


def main() -> int:
    cases = [
        ("version-only", run_case("version-only.json"), 0),
        ("wrong-label", run_case("wrong-label.json"), 1),
        ("eb-setting-change", run_case("eb-setting-change.json"), 1),
        ("iam-update", run_case("iam-update.json"), 1),
        ("dns-update", run_case("dns-update.json"), 1),
        ("create", run_case("create.json"), 1),
        ("delete", run_case("delete.json"), 1),
        ("replace", run_case("replace.json"), 1),
        ("multiple-updates", run_case("multiple-updates.json"), 1),
        ("empty", run_case("empty.json"), 1),
    ]
    failures = [
        (name, result, expected)
        for name, result, expected in cases
        if result.returncode != expected
    ]
    download_failures = test_download_pinning()
    if failures or download_failures:
        if failures:
            print(
                "FAIL: release 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,
                )
        for item in download_failures:
            print(f"FAIL: {item}", file=sys.stderr)
        return 1
    print("PASS: Terraform release plan safety checks")
    return 0


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