fce343cea1
Template moved to _archived/ (c7da48d7); glob discovery no longer sees it. Claude-Session: https://claude.ai/code/session_01N218bERyNeLhbLmDRDvbWF
117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
"""state-tx script nodes must declare CP_URL, STATE_TX_NEW_STATE and STATE_TX_FROM_STATES.
|
|
|
|
scripts/workflow/state-tx fails with exit 2 when any is missing
|
|
(from_states_missing for STATE_TX_FROM_STATES).
|
|
CP_SERVICE_TOKEN_FILE is supplied by the workflow-script/v1 composite, not
|
|
runtime_env, and must not be declared here. STATE_TX_NEW_STATE must equal the
|
|
template's own output.state. STATE_TX_FROM_STATES is a comma list of acceptable
|
|
current flow_states and must be non-empty.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
_DIR = Path(__file__).resolve().parents[2] / "task-templates" / "workflow"
|
|
|
|
# Every template with a state-tx node, discovered from disk so new @N variants
|
|
# are covered automatically.
|
|
_NAMES = sorted(
|
|
p.name
|
|
for p in _DIR.glob("*.yaml")
|
|
if any(
|
|
n.get("cmd") == "scripts/workflow/state-tx"
|
|
for n in (yaml.safe_load(p.read_text()).get("nodes") or [])
|
|
)
|
|
)
|
|
|
|
|
|
def test_state_tx_templates_discovered():
|
|
assert {
|
|
"test-validate@1.yaml",
|
|
"frozen-fixture-emit@1.yaml",
|
|
} <= set(_NAMES)
|
|
assert "bug-ancestor-auto-advance@1.yaml" not in _NAMES # archived (WT-ANC-15)
|
|
|
|
|
|
@pytest.mark.parametrize("name", _NAMES)
|
|
def test_state_tx_node_declares_required_runtime_env(name):
|
|
doc = yaml.safe_load((_DIR / name).read_text())
|
|
nodes = [n for n in doc["nodes"] if n.get("cmd") == "scripts/workflow/state-tx"]
|
|
assert len(nodes) == 1
|
|
env = nodes[0].get("runtime_env") or {}
|
|
assert env.get("CP_URL")
|
|
assert env.get("STATE_TX_NEW_STATE") == doc["output"]["state"]
|
|
assert "CP_SERVICE_TOKEN_FILE" not in env
|
|
assert not any(k.startswith("OIDC_") for k in env)
|
|
|
|
|
|
@pytest.mark.parametrize("name", _NAMES)
|
|
def test_state_tx_node_declares_from_states(name):
|
|
doc = yaml.safe_load((_DIR / name).read_text())
|
|
nodes = [n for n in doc["nodes"] if n.get("cmd") == "scripts/workflow/state-tx"]
|
|
env = nodes[0].get("runtime_env") or {}
|
|
raw = env.get("STATE_TX_FROM_STATES")
|
|
assert isinstance(raw, str) and raw.strip()
|
|
states = [s.strip() for s in raw.split(",")]
|
|
assert all(states), f"empty entry in STATE_TX_FROM_STATES {raw!r}"
|
|
assert doc["output"]["state"] not in states
|
|
# The template's own input state must be an acceptable from-state.
|
|
assert doc["input"]["state"] in states
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CP_URL must be the ingress form, never the in-cluster Service on :8100.
|
|
#
|
|
# The deploy repo's networkpolicy-cp-internal-traefik-only.yaml admits :8100 only
|
|
# from dashboard / git-planner-backend / cert-expiry-checker and the traefik and
|
|
# cms-proxy namespaces. Agent/script pods are dropped, so a template pointing
|
|
# CP_URL at svc.cluster.local:8100 times out (30s) and, with on_fail: continue,
|
|
# silently loses the transition. The workflow scripts join CP_URL + "/v1/...",
|
|
# so the /api ingress path prefix is preserved.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CP_URL_INGRESS = "https://agents.oreillyit.nz/api"
|
|
_FORBIDDEN_CP_URL_MARKERS = ("svc.cluster.local:8100", "controlplane.agent-runtimes.svc")
|
|
|
|
_ALL_TEMPLATES = sorted(_DIR.glob("*.yaml"))
|
|
|
|
|
|
def _cp_url_values(path: Path) -> list[str]:
|
|
doc = yaml.safe_load(path.read_text())
|
|
return [
|
|
(n.get("runtime_env") or {})["CP_URL"]
|
|
for n in (doc.get("nodes") or [])
|
|
if "CP_URL" in (n.get("runtime_env") or {})
|
|
]
|
|
|
|
|
|
def test_cp_url_templates_discovered():
|
|
names = {p.name for p in _ALL_TEMPLATES if _cp_url_values(p)}
|
|
assert {
|
|
"test-validate@1.yaml",
|
|
"frozen-fixture-emit@1.yaml",
|
|
"scope-decompose-sonnet@1.yaml",
|
|
"scope-decompose-sonnet@2.yaml",
|
|
"bug-verify-sonnet@1.yaml",
|
|
"bug-verify-sonnet@2.yaml",
|
|
} <= names
|
|
|
|
|
|
@pytest.mark.parametrize("path", _ALL_TEMPLATES, ids=lambda p: p.name)
|
|
def test_cp_url_is_ingress_not_in_cluster_service(path):
|
|
for value in _cp_url_values(path):
|
|
assert not any(m in value for m in _FORBIDDEN_CP_URL_MARKERS), (
|
|
f"{path.name}: CP_URL={value!r} targets the in-cluster CP service; agent pods "
|
|
"are blocked on :8100 by NetworkPolicy"
|
|
)
|
|
assert value == CP_URL_INGRESS, f"{path.name}: CP_URL={value!r}"
|
|
|
|
|
|
@pytest.mark.parametrize("path", _ALL_TEMPLATES, ids=lambda p: p.name)
|
|
def test_no_template_mentions_in_cluster_cp_service(path):
|
|
text = path.read_text()
|
|
assert not any(m in text for m in _FORBIDDEN_CP_URL_MARKERS)
|