107 lines
4.7 KiB
Python
107 lines
4.7 KiB
Python
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[1] / "scripts/security_flow_report.py"
|
|
|
|
|
|
def workplan(root, repo, task):
|
|
directory = root / repo / "workplans"
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
(directory / "test.md").write_text(
|
|
"---\nstatus: active\nflavor: implementation\n---\n\n"
|
|
"## Work\n\n```task\n" + task + "\n```\n"
|
|
)
|
|
|
|
|
|
def report(root):
|
|
return json.loads(subprocess.check_output(
|
|
[sys.executable, str(SCRIPT), "--root", str(root), "--json"], text=True
|
|
))
|
|
|
|
|
|
def test_default_estate_observes_activity_owner_dependency(tmp_path):
|
|
workplan(tmp_path, "railiance-platform",
|
|
"id: RPF-WP-0048-T02\nstatus: wait\npriority: high\n"
|
|
"depends_on: [ACTIVITY-WP-0041-T03]")
|
|
workplan(tmp_path, "activity-core",
|
|
"id: ACTIVITY-WP-0041-T03\nstatus: todo\npriority: high")
|
|
waiting = report(tmp_path)
|
|
assert [t["id"] for t in waiting["READY"]] == ["ACTIVITY-WP-0041-T03"]
|
|
assert waiting["UNBLOCKED"] == []
|
|
workplan(tmp_path, "activity-core",
|
|
"id: ACTIVITY-WP-0041-T03\nstatus: done\npriority: high")
|
|
assert [t["id"] for t in report(tmp_path)["UNBLOCKED"]] == ["RPF-WP-0048-T02"]
|
|
|
|
|
|
def test_default_estate_includes_custodian_and_explains_human_gate(tmp_path):
|
|
workplan(tmp_path, "the-custodian",
|
|
"id: CUST-WP-0079-T05\nstatus: progress\npriority: medium")
|
|
workplan(tmp_path, "policy-nexus",
|
|
"id: PNEX-WP-0006-T01\nstatus: wait\npriority: high\n"
|
|
'needs_human: true\nintervention_note: "Public publication in daily batch"')
|
|
result = report(tmp_path)
|
|
assert [t["id"] for t in result["READY"]] == ["CUST-WP-0079-T05"]
|
|
assert result["ATTENDED"][0]["reason"] == "Public publication in daily batch"
|
|
assert result["UNBLOCKED"] == []
|
|
|
|
|
|
def test_pending_owner_task_is_not_ready_until_predecessor_finishes(tmp_path):
|
|
workplan(tmp_path, "railiance-platform",
|
|
"id: RPF-WP-0044-T06\nstatus: todo\npriority: medium\n"
|
|
"depends_on: [ACTIVITY-WP-0041-T03]")
|
|
workplan(tmp_path, "activity-core",
|
|
"id: ACTIVITY-WP-0041-T03\nstatus: progress\npriority: high")
|
|
waiting = report(tmp_path)
|
|
assert [t["id"] for t in waiting["READY"]] == ["ACTIVITY-WP-0041-T03"]
|
|
assert waiting["DEPENDENT"][0]["pending_dependencies"] == ["ACTIVITY-WP-0041-T03"]
|
|
workplan(tmp_path, "activity-core",
|
|
"id: ACTIVITY-WP-0041-T03\nstatus: done\npriority: high")
|
|
assert [t["id"] for t in report(tmp_path)["READY"]] == ["RPF-WP-0044-T06"]
|
|
|
|
|
|
def test_unknown_dependency_and_legacy_owner_gate_are_visible_but_not_ready(tmp_path):
|
|
workplan(tmp_path, "access-engine",
|
|
"id: FLEX-WP-0036-T06\nstatus: todo\npriority: medium\n"
|
|
"depends_on: [UNKNOWN-WP-0001-T01]")
|
|
workplan(tmp_path, "ops-mason",
|
|
"id: MASON-WP-0010-T03\nstatus: progress\npriority: high\n"
|
|
"blocked_on: message-from:railiance-platform")
|
|
result = report(tmp_path)
|
|
assert not result["READY"]
|
|
assert len(result["DEPENDENT"]) == 2
|
|
assert result["DEPENDENT"][0]["blocked_on"] == "message-from:railiance-platform"
|
|
|
|
|
|
def test_human_flag_prevents_ready_even_if_status_has_not_been_requalified(tmp_path):
|
|
workplan(tmp_path, "user-engine",
|
|
"id: USER-WP-0027-T04\nstatus: progress\npriority: high\n"
|
|
"needs_human: true\nintervention_note: Real MFA journey")
|
|
result = report(tmp_path)
|
|
assert not result["READY"]
|
|
assert result["ATTENDED"][0]["id"] == "USER-WP-0027-T04"
|
|
|
|
|
|
def test_workplan_and_multiline_task_dependencies_both_gate_successor(tmp_path):
|
|
workplan(tmp_path, "railiance-platform",
|
|
"id: RPF-WP-0048-T02\nstatus: todo\npriority: high\n"
|
|
"depends_on:\n - ACTIVITY-WP-0041-T03\n - ACTIVITY-WP-0041-T04")
|
|
source = tmp_path / "railiance-platform/workplans/test.md"
|
|
source.write_text(source.read_text().replace(
|
|
"flavor: implementation", "flavor: implementation\ndepends_on: [ACTIVITY-WP-0042]"))
|
|
result = report(tmp_path)
|
|
assert not result["READY"]
|
|
assert result["DEPENDENT"][0]["pending_dependencies"] == [
|
|
"ACTIVITY-WP-0042", "ACTIVITY-WP-0041-T03", "ACTIVITY-WP-0041-T04"]
|
|
|
|
|
|
def test_attended_item_keeps_unfinished_ordered_prerequisites_visible(tmp_path):
|
|
workplan(tmp_path, "key-cape",
|
|
"id: KEY-WP-0039-T02\nstatus: wait\npriority: medium\n"
|
|
"needs_human: true\nintervention_note: Attended owner token session\n"
|
|
"depends_on: [KEY-WP-0038-T03]")
|
|
result = report(tmp_path)
|
|
assert not result["READY"] and not result["UNBLOCKED"]
|
|
assert result["ATTENDED"][0]["pending_dependencies"] == ["KEY-WP-0038-T03"]
|