194 lines
6.7 KiB
Python
194 lines
6.7 KiB
Python
|
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
|