diff --git a/deploy/nestquest-controller.service b/deploy/nestquest-controller.service index 5f220d5..3084257 100644 --- a/deploy/nestquest-controller.service +++ b/deploy/nestquest-controller.service @@ -8,7 +8,7 @@ ExecStartPre=/home/overseer/.local/share/nestquest-controller/scripts/nq-control ExecStart=/home/overseer/.local/share/nestquest-controller/scripts/nq-controller Restart=on-failure RestartSec=60s -RestartPreventExitStatus=75 77 +RestartPreventExitStatus=77 TimeoutStartSec=0 KillMode=control-group Environment=PYTHONUNBUFFERED=1 diff --git a/orchestration_tests/test_nq_approve_merge.py b/orchestration_tests/test_nq_approve_merge.py index c070316..daef122 100644 --- a/orchestration_tests/test_nq_approve_merge.py +++ b/orchestration_tests/test_nq_approve_merge.py @@ -1,5 +1,8 @@ from pathlib import Path +import contextlib +import io import runpy +import sys import tempfile import unittest from unittest.mock import patch @@ -10,6 +13,46 @@ HEAD = "a" * 40 class MergeGateTests(unittest.TestCase): + def test_main_rejects_moved_head_unmergeable_and_pending_checks(self): + root = Path("/tmp/opencode") + root.mkdir(exist_ok=True) + with tempfile.TemporaryDirectory(dir=root) as directory: + review_file = Path(directory) / "review.txt" + review_file.write_text(f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n") + for problem in ("moved", "unmergeable", "pending", "none"): + with self.subTest(problem=problem): + merged = False + calls = [] + def fake_request(path, token, payload=None): + nonlocal merged + calls.append((path, payload)) + if path.endswith("/reviews"): + return [{"id": 1, "state": "APPROVED", "commit_id": HEAD, + "user": {"login": "review_agent"}}] + if path.endswith("/status"): + return {"total_count": 1, "state": "pending" if problem == "pending" else "success"} + if path.endswith("/merge"): + merged = True + return {} + if path.endswith("/pulls/149"): + return {"number": 149, "base": {"ref": "dev"}, + "head": {"sha": "b" * 40 if problem == "moved" else HEAD}, + "state": "open", "merged": merged, + "mergeable": problem != "unmergeable"} + raise AssertionError(path) + globals_ = APPROVE["main"].__globals__ + with patch.dict(globals_, {"request": fake_request, + "token_from_file": lambda: "test"}), \ + patch.object(sys, "argv", ["nq-approve-merge", "merge", "149", HEAD, + str(review_file)]), \ + contextlib.redirect_stdout(io.StringIO()): + if problem == "none": + self.assertEqual(APPROVE["main"](), 0) + else: + with self.assertRaises(RuntimeError): + APPROVE["main"]() + self.assertEqual(merged, problem == "none") + def test_review_evidence_must_match_exact_head_and_approval(self): root = Path("/tmp/opencode") root.mkdir(exist_ok=True) diff --git a/orchestration_tests/test_nq_controller.py b/orchestration_tests/test_nq_controller.py index c5e50d6..76e7c46 100644 --- a/orchestration_tests/test_nq_controller.py +++ b/orchestration_tests/test_nq_controller.py @@ -1,16 +1,38 @@ import json +import io from pathlib import Path import runpy import sqlite3 import sys import tempfile import unittest +from unittest.mock import patch CONTROL = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-controller")) class ControllerProtocolTests(unittest.TestCase): + def test_ctxd_preflight_reads_configured_endpoint_and_checks_identity(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + config = root / ".config/opencode/opencode.json" + config.parent.mkdir(parents=True) + config.write_text(json.dumps({"mcp": {"ctxd": { + "url": "http://localhost:9091/mcp", + "headers": {"Authorization": "Bearer test"}, + }}})) + record = {"project_id": CONTROL["CTXD_PROJECT"], + "file_path": CONTROL["CTXD_FILE"], + "content": "# CONTROLLER_TICK", "version": 6} + with patch.object(Path, "home", return_value=root), \ + patch.dict(CONTROL["check_ctxd"].__globals__["urllib"].request.__dict__, + {"urlopen": lambda request, timeout: io.BytesIO(json.dumps(record).encode())}): + self.assertEqual(CONTROL["check_ctxd"](), 6) + record["project_id"] = "wrong" + with self.assertRaisesRegex(RuntimeError, "wrong NestQuest prompt"): + CONTROL["check_ctxd"]() + def test_installed_opencode_text_event(self): event = {"type": "text", "sessionID": "ses_test", "part": {"type": "text", "text": "Next task queued.\nORCHESTRATOR_STATE: CONTINUE\n"}} self.assertEqual(CONTROL["parse_event"](json.dumps(event)), (event["part"]["text"], None))