Christian Nennemann
bbf557e54b
Move Go reference implementation to refimpl/go-lang/ and add new Python reference implementation in refimpl/python/. Update build.sh with renamed draft and simplified tool paths. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
124 lines
3.0 KiB
Python
124 lines
3.0 KiB
Python
"""Tests for DAG validation."""
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from ect import Payload, MemoryLedger, validate_dag, default_dag_config, POL_DECISION_APPROVED
|
|
|
|
|
|
def test_validate_dag_root():
|
|
store = MemoryLedger()
|
|
payload = Payload(
|
|
iss="",
|
|
aud=[],
|
|
iat=0,
|
|
exp=0,
|
|
jti="jti-001",
|
|
exec_act="",
|
|
par=[],
|
|
pol="",
|
|
pol_decision=POL_DECISION_APPROVED,
|
|
wid="wf-1",
|
|
)
|
|
validate_dag(payload, store, default_dag_config())
|
|
|
|
|
|
def test_validate_dag_duplicate_jti():
|
|
store = MemoryLedger()
|
|
p = Payload(
|
|
iss="x",
|
|
aud=["y"],
|
|
iat=0,
|
|
exp=0,
|
|
jti="jti-001",
|
|
exec_act="a",
|
|
par=[],
|
|
pol="p",
|
|
pol_decision=POL_DECISION_APPROVED,
|
|
wid="wf-1",
|
|
)
|
|
store.append("dummy-jws", p)
|
|
payload = Payload(
|
|
iss="",
|
|
aud=[],
|
|
iat=0,
|
|
exp=0,
|
|
jti="jti-001",
|
|
exec_act="",
|
|
par=[],
|
|
pol="",
|
|
pol_decision=POL_DECISION_APPROVED,
|
|
wid="wf-1",
|
|
)
|
|
with pytest.raises(ValueError, match="task ID.*already exists"):
|
|
validate_dag(payload, store, default_dag_config())
|
|
|
|
|
|
def test_validate_dag_parent_exists():
|
|
store = MemoryLedger()
|
|
now = int(time.time())
|
|
p = Payload(
|
|
iss="x",
|
|
aud=["y"],
|
|
iat=now - 60,
|
|
exp=now + 600,
|
|
jti="jti-001",
|
|
exec_act="a",
|
|
par=[],
|
|
pol="p",
|
|
pol_decision=POL_DECISION_APPROVED,
|
|
wid="wf-1",
|
|
)
|
|
store.append("jws1", p)
|
|
payload = Payload(
|
|
iss="",
|
|
aud=[],
|
|
iat=now,
|
|
exp=now + 600,
|
|
jti="jti-002",
|
|
exec_act="b",
|
|
par=["jti-001"],
|
|
pol="p",
|
|
pol_decision=POL_DECISION_APPROVED,
|
|
wid="wf-1",
|
|
)
|
|
validate_dag(payload, store, default_dag_config())
|
|
|
|
|
|
def test_validate_dag_parent_not_found():
|
|
store = MemoryLedger()
|
|
now = int(time.time())
|
|
payload = Payload(
|
|
iss="",
|
|
aud=[],
|
|
iat=now,
|
|
exp=now + 600,
|
|
jti="jti-002",
|
|
exec_act="",
|
|
par=["jti-missing"],
|
|
pol="",
|
|
pol_decision=POL_DECISION_APPROVED,
|
|
)
|
|
with pytest.raises(ValueError, match="parent task not found"):
|
|
validate_dag(payload, store, default_dag_config())
|
|
|
|
|
|
def test_validate_dag_parent_policy_rejected_requires_compensation():
|
|
from ect import POL_DECISION_REJECTED
|
|
store = MemoryLedger()
|
|
now = int(time.time())
|
|
p = Payload(
|
|
iss="x", aud=["y"], iat=now - 60, exp=now + 600,
|
|
jti="jti-rej", exec_act="a", par=[], pol="p", pol_decision=POL_DECISION_REJECTED, wid="wf-1",
|
|
)
|
|
store.append("jws1", p)
|
|
payload = Payload(
|
|
iss="", aud=[], iat=now, exp=now + 600,
|
|
jti="jti-child", exec_act="b", par=["jti-rej"], pol="p", pol_decision=POL_DECISION_APPROVED, wid="wf-1",
|
|
)
|
|
with pytest.raises(ValueError, match="compensation"):
|
|
validate_dag(payload, store, default_dag_config())
|
|
payload.ext = {"compensation_required": True}
|
|
validate_dag(payload, store, default_dag_config())
|