63 lines
2.3 KiB
Python
63 lines
2.3 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",
|
|
"bug-ancestor-auto-advance@1.yaml",
|
|
} <= set(_NAMES)
|
|
|
|
|
|
@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
|