80 lines
3.6 KiB
Python
80 lines
3.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import pytest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
import security_flow_daily as daily
|
|
import security_flow_report as flow
|
|
|
|
|
|
def test_notifications_compare_entire_actionable_lists():
|
|
report = {"READY": [{"id": "T1", "status": "todo"}], "UNBLOCKED": [], "ATTENDED": []}
|
|
first = daily.notifications(report, {}, "coding", "human")
|
|
assert first == [("agent", "coding", {"READY": report["READY"], "UNBLOCKED": []})]
|
|
assert daily.notifications(report, {"agent": first[0][2]}, "coding", "human") == []
|
|
report["READY"] = []
|
|
assert daily.notifications(report, {"agent": first[0][2]}, "coding", "human")
|
|
|
|
|
|
def test_attended_goes_only_to_founder_and_is_daily():
|
|
report = {"READY": [], "UNBLOCKED": [], "ATTENDED": [{"id": "T2"}]}
|
|
state = {"agent": {"READY": [], "UNBLOCKED": []}}
|
|
assert daily.notifications(report, state, "coding", "human") == [
|
|
("founder", "human", {"ATTENDED": report["ATTENDED"]})]
|
|
|
|
|
|
def test_atomic_checkpoint_retains_prior_recipient(tmp_path):
|
|
checkpoint = tmp_path / "delivery.json"
|
|
daily.save(checkpoint, {"agent_message_id": "receipt-1"})
|
|
assert checkpoint.read_text().find("receipt-1") >= 0
|
|
assert not checkpoint.with_suffix(".json.tmp").exists()
|
|
|
|
|
|
def test_retry_does_not_replay_successful_agent_delivery(tmp_path, monkeypatch):
|
|
report = {"READY": [{"id": "T1"}], "UNBLOCKED": [], "ATTENDED": [{"id": "T2"}]}
|
|
monkeypatch.setattr(sys, "argv", ["security_flow_daily", "--apply", "--state-dir", str(tmp_path)])
|
|
commands = []
|
|
monkeypatch.setattr(daily.subprocess, "run", lambda command, **kw: commands.append(command))
|
|
monkeypatch.setattr(daily.subprocess, "check_output", lambda *a, **kw: json.dumps(report))
|
|
sent = []
|
|
def fail_founder(hub, recipient, payload):
|
|
sent.append(recipient)
|
|
if recipient == "human":
|
|
raise OSError("hub interrupted")
|
|
return "agent-receipt"
|
|
monkeypatch.setattr(daily, "post", fail_founder)
|
|
with pytest.raises(OSError):
|
|
daily.main()
|
|
assert "--apply" in commands[0]
|
|
assert json.loads((tmp_path / "delivery.json").read_text())["agent_message_id"] == "agent-receipt"
|
|
monkeypatch.setattr(daily, "post", lambda hub, recipient, payload: sent.append(recipient) or "receipt")
|
|
daily.main()
|
|
assert sent == ["the-custodian", "human", "human"]
|
|
|
|
|
|
def test_sync_failure_prevents_notifications(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(sys, "argv", ["security_flow_daily", "--apply", "--state-dir", str(tmp_path)])
|
|
def fail_sync(*args, **kwargs):
|
|
raise daily.subprocess.CalledProcessError(1, args[0])
|
|
monkeypatch.setattr(daily.subprocess, "run", fail_sync)
|
|
with pytest.raises(daily.subprocess.CalledProcessError):
|
|
daily.main()
|
|
assert not (tmp_path / "delivery.json").exists()
|
|
|
|
|
|
def test_report_classifies_terminal_unknown_and_human_dependencies():
|
|
def task(tid, status, deps=(), human=False, flavor="implementation"):
|
|
return {"id": tid, "status": status, "deps": deps, "human": human,
|
|
"blocked_on": "", "priority": "high",
|
|
"wp": {"status": "active", "flavor": flavor}}
|
|
tasks = {"X-T01": task("X-T01", "done"),
|
|
"X-T02": task("X-T02", "wait", ["X-T01"]),
|
|
"X-T03": task("X-T03", "wait", ["missing"]),
|
|
"X-T04": task("X-T04", "wait", ["X-T01"], human=True),
|
|
"Y-T01": task("Y-T01", "todo", flavor="residual")}
|
|
report = flow.classify(tasks)
|
|
assert [t["id"] for t in report["UNBLOCKED"]] == ["X-T02"]
|
|
assert [t["id"] for t in report["ATTENDED"]] == ["X-T04"]
|
|
assert not report["READY"]
|