ops-warden/tests/test_scaffold.py

from copy import deepcopy
import socket
import subprocess

import pytest
import yaml
from typer.testing import CliRunner

from warden.cli import app
from warden.routing.catalog import Catalog, load_catalog
from warden.routing.scaffold import ScaffoldError, render, scaffold


@pytest.fixture
def draft_request():
    return {
        "schema_version": 1,
        "lane_id": "example-workload-kv",
        "title": "Example workload read",
        "need_keywords": ["example workload"],
        "consumer_repo": "example-consumer",
        "owner_repo": "railiance-platform",
        "workload_ref": {
            "applicability": "applicable",
            "unknown_reason": "Synthetic fixture has no published deployment identity",
        },
        "risk": "standard",
        "wiki_ref": "railiance-platform:docs/catalog-lane-scaffold.md",
        "canon_ref": "railiance-platform:credential-change-requests/CCR-2026-9999.yaml",
        "platform_request_ref": "railiance-platform:examples/catalog-lane-scaffold/platform-request.yaml",
        "reviewed": "2026-10-04",
    }


def empty_catalog(tmp_path):
    return Catalog(path=tmp_path / "unused.yaml", entries=[])


def test_deterministic_draft_roundtrip_and_input_unchanged(draft_request, tmp_path):
    before = deepcopy(draft_request)
    contents = render(draft_request, empty_catalog(tmp_path))
    assert contents == render(draft_request, empty_catalog(tmp_path))
    assert draft_request == before
    for name in ["one", "two"]:
        source = tmp_path / f"{name}.yaml"
        source.write_text(yaml.safe_dump(draft_request))
        paths = scaffold(source, tmp_path / name, catalog=empty_catalog(tmp_path))
        assert len(paths) == 3
    assert (tmp_path / "one/catalog.yaml").read_bytes() == (
        tmp_path / "two/catalog.yaml"
    ).read_bytes()
    entry = load_catalog(tmp_path / "one/catalog.yaml").entries[0]
    assert entry.status == "draft"
    assert not entry.exec_capable and not entry.warden_executes
    assert entry.fetch_command is None
    assert entry.workload_ref.resolution == "unknown"
    assert entry.workload_ref.name is None
    assert entry.risk == "standard"
    assert yaml.safe_load((tmp_path / "one/request.yaml").read_text()) == draft_request


@pytest.mark.parametrize("risk", ["high", "ungraded"])
def test_explicit_risk_is_preserved(draft_request, tmp_path, risk):
    draft_request["risk"] = risk
    raw = yaml.safe_load(render(draft_request, empty_catalog(tmp_path))["catalog.yaml"])
    assert raw["entries"][0]["risk"] == risk


@pytest.mark.parametrize(
    "mutation",
    [
        lambda r: r.update(exec_capable=True),
        lambda r: r.pop("risk"),
        lambda r: r.update(risk=""),
        lambda r: r.update(schema_version=True),
        lambda r: r.update(lane_id="../escape"),
        lambda r: r.update(reviewed="yesterday"),
        lambda r: r.update(need_keywords=[]),
        lambda r: r.update(workload_ref={"applicability": "applicable"}),
        lambda r: r.update(
            workload_ref={
                "applicability": "applicable",
                "unknown_reason": "unknown",
                "name": "inferred",
            }
        ),
        lambda r: r.update(
            workload_ref={
                "applicability": "applicable",
                "unknown_reason": "unknown",
                "role": "admin",
            }
        ),
    ],
)
def test_invalid_request_publishes_nothing(draft_request, tmp_path, mutation):
    mutation(draft_request)
    source = tmp_path / "input.yaml"
    source.write_text(yaml.safe_dump(draft_request))
    with pytest.raises(ScaffoldError):
        scaffold(source, tmp_path / "stage", catalog=empty_catalog(tmp_path))
    assert not (tmp_path / "stage").exists()


@pytest.mark.parametrize(
    "field", ["title", "platform_request_ref", "need_keywords", "workload_ref"]
)
def test_secret_rejection_does_not_echo_value(draft_request, tmp_path, field):
    token = "ghp_" + "Ab12" * 12
    draft_request[field] = (
        [token]
        if field == "need_keywords"
        else {"applicability": "applicable", "unknown_reason": token}
        if field == "workload_ref"
        else token
    )
    with pytest.raises(ScaffoldError) as exc:
        render(draft_request, empty_catalog(tmp_path))
    assert token not in str(exc.value)
    assert "Ab12" not in str(exc.value)


def test_existing_lane_collision_even_when_draft(draft_request, tmp_path):
    raw = render(draft_request, empty_catalog(tmp_path))["catalog.yaml"]
    path = tmp_path / "catalog.yaml"
    path.write_text(raw)
    with pytest.raises(ScaffoldError, match="already exists"):
        render(draft_request, load_catalog(path))


@pytest.mark.parametrize("symlink", [False, True])
def test_preexisting_destination_is_preserved(draft_request, tmp_path, symlink):
    source = tmp_path / "input.yaml"
    source.write_text(yaml.safe_dump(draft_request))
    stage = tmp_path / "stage"
    if symlink:
        stage.symlink_to(tmp_path / "missing")
    else:
        stage.mkdir()
        (stage / "keep").write_text("original")
    with pytest.raises(ScaffoldError, match="already exists"):
        scaffold(source, stage, catalog=empty_catalog(tmp_path))
    if not symlink:
        assert (stage / "keep").read_text() == "original"
    else:
        assert stage.is_symlink()


def test_cli_offline_and_bad_yaml_diagnostic(draft_request, tmp_path, monkeypatch):
    def forbidden(*args, **kwargs):
        raise AssertionError("credential/network execution forbidden")

    monkeypatch.setattr(socket, "create_connection", forbidden)
    monkeypatch.setattr(subprocess, "run", forbidden)
    monkeypatch.setenv("WARDEN_MEMORY", "0")
    source = tmp_path / "input.yaml"
    source.write_text(yaml.safe_dump(draft_request))
    # Override live catalog with a different inert draft lane.
    other = deepcopy(draft_request)
    other["lane_id"] = "other-example"
    catalog_path = tmp_path / "catalog.yaml"
    catalog_path.write_text(render(other, empty_catalog(tmp_path))["catalog.yaml"])
    monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(catalog_path))
    result = CliRunner().invoke(
        app,
        [
            "route",
            "scaffold",
            "--input",
            str(source),
            "--output-dir",
            str(tmp_path / "stage"),
        ],
    )
    assert result.exit_code == 0, result.output
    assert (
        load_catalog(tmp_path / "stage/catalog.yaml").entries[0].id
        == draft_request["lane_id"]
    )
    source.write_text("title: [ghp_private\n")
    result = CliRunner().invoke(
        app,
        [
            "route",
            "scaffold",
            "--input",
            str(source),
            "--output-dir",
            str(tmp_path / "bad"),
        ],
    )
    assert result.exit_code == 1
    assert "ghp_private" not in result.output