98 lines
4.1 KiB
Python
98 lines
4.1 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("monitor", ROOT / "scripts/monitor_dependency_triage.py")
|
|
monitor = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(monitor)
|
|
|
|
|
|
class DependencyTriageTests(unittest.TestCase):
|
|
def test_yaml_task_lists_and_quoted_hash_are_preserved(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
scratch = Path(directory)
|
|
repo = scratch / "repo"
|
|
(repo / "workplans").mkdir(parents=True)
|
|
(scratch / "repos.json").write_text(json.dumps([{"slug": "example", "local_path": str(repo)}]))
|
|
(repo / "workplans/plan.md").write_text('''---
|
|
id: EX-WP-0001
|
|
type: workplan
|
|
status: blocked
|
|
flavor: implementation
|
|
---
|
|
```task
|
|
id: EX-WP-0001-T01
|
|
status: wait
|
|
depends_on:
|
|
- EX-WP-0002-T01
|
|
needs_human: true
|
|
blocking_reason: "Review packet #1"
|
|
wait_until: 2026-10-04
|
|
```
|
|
''')
|
|
subprocess.run([sys.executable, str(monitor.SCANNER)], check=True, capture_output=True,
|
|
env=dict(os.environ, TRIAGE_SCRATCH=directory))
|
|
task = json.loads((scratch / "files.json").read_text())["T"]["EX-WP-0001-T01"]
|
|
self.assertEqual(task["depends_on"], ["EX-WP-0002-T01"])
|
|
self.assertEqual(task["reason"], "Review packet #1")
|
|
self.assertTrue(task["needs_human"])
|
|
self.assertEqual(task["wait_until"], "2026-10-04")
|
|
|
|
def records(self):
|
|
task = {"wp": "plan", "status": "wait", "hub_id": "uuid", "needs_human": True,
|
|
"reason": "Review", "decision_id": "decision", "blocked_on": None,
|
|
"wait_until": None, "depends_on": []}
|
|
return {"WP": {"plan": {"status": "blocked", "flavor": "implementation"}}, "T": {"task": task}}
|
|
|
|
def test_missing_projection_is_visible(self):
|
|
result = monitor.audit(self.records(), [])
|
|
self.assertEqual(result["qualifier_drift"][0]["field"], "projection")
|
|
|
|
def test_decision_sync_gap_is_detected(self):
|
|
result = monitor.audit(self.records(), [{"id": "uuid", "status": "wait",
|
|
"needs_human": True, "blocking_reason": "Review"}])
|
|
self.assertEqual(result["qualifier_drift"], [{"task": "task", "field": "decision_id",
|
|
"file": "decision", "hub": None}])
|
|
self.assertEqual(result["unrouted_human_gates"], [])
|
|
|
|
def test_residuals_are_excluded_from_actionable_findings(self):
|
|
records = self.records()
|
|
records["WP"]["plan"]["flavor"] = "residual"
|
|
result = monitor.audit(records, [])
|
|
self.assertEqual(result["wait_tasks"], 0)
|
|
self.assertEqual(result["qualifier_drift"], [])
|
|
|
|
def test_terminal_predecessor_keeps_separate_human_gate(self):
|
|
records = self.records()
|
|
records["T"]["task"]["depends_on"] = ["predecessor"]
|
|
records["T"]["predecessor"] = {"wp": "plan", "status": "done"}
|
|
result = monitor.audit(records, [])
|
|
self.assertTrue(result["terminal_dependency_waits"][0]["needs_human"])
|
|
self.assertEqual(result["human_gates"], 1)
|
|
|
|
def test_tasks_follow_offset_pagination(self):
|
|
response = MagicMock()
|
|
with patch.object(monitor, "urlopen", return_value=response) as fetch, patch.object(
|
|
monitor.json, "load", side_effect=[[{"id": str(i)} for i in range(200)], [{"id": "last"}]]):
|
|
rows = monitor.get_rows("http://hub", "tasks")
|
|
self.assertEqual(len(rows), 201)
|
|
self.assertIn("offset=200", fetch.call_args_list[1].args[0])
|
|
|
|
def test_messages_do_not_use_unsupported_offset(self):
|
|
with patch.object(monitor, "urlopen", return_value=MagicMock()) as fetch, patch.object(
|
|
monitor.json, "load", return_value=[{"id": str(i)} for i in range(200)]):
|
|
rows = monitor.get_rows("http://hub", "messages", from_agent="owner")
|
|
self.assertEqual(len(rows), 200)
|
|
self.assertEqual(fetch.call_count, 1)
|
|
self.assertNotIn("offset", fetch.call_args.args[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|