fix: re-resolve conflicts for PR #506 after master advance
This commit is contained in:
@@ -18,14 +18,25 @@ def _reset_mutation_authority(monkeypatch):
|
||||
explicitly.
|
||||
"""
|
||||
monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
|
||||
# Isolate durable session-state files so tests never share host cache (#559).
|
||||
import tempfile
|
||||
|
||||
_state_tmp = tempfile.TemporaryDirectory(prefix="gitea-session-state-")
|
||||
monkeypatch.setenv("GITEA_MCP_SESSION_STATE_DIR", _state_tmp.name)
|
||||
try:
|
||||
import mcp_server
|
||||
except Exception:
|
||||
_state_tmp.cleanup()
|
||||
yield
|
||||
return
|
||||
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
|
||||
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
|
||||
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
|
||||
try:
|
||||
import review_workflow_load
|
||||
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
import capability_stop_terminal
|
||||
capability_stop_terminal.clear()
|
||||
@@ -37,3 +48,13 @@ def _reset_mutation_authority(monkeypatch):
|
||||
capability_stop_terminal.clear()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
import review_workflow_load
|
||||
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
mcp_server._REVIEW_DECISION_LOCK = None
|
||||
except Exception:
|
||||
pass
|
||||
_state_tmp.cleanup()
|
||||
|
||||
@@ -498,6 +498,81 @@ class TestCanonicalReconcileSchema(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
class TestReconcilerCloseProof(unittest.TestCase):
|
||||
"""Issue #306: a reconciler PR close must carry its proof fields."""
|
||||
|
||||
def _closed_report(self, **kwargs):
|
||||
# A report that claims PR #99 was closed via the reconciler path.
|
||||
report = (
|
||||
_reconcile_handoff(**kwargs)
|
||||
.replace(
|
||||
"- Capabilities proven: gitea.read, gitea.pr.comment",
|
||||
"- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
|
||||
)
|
||||
.replace("- Missing capabilities: gitea.pr.close", "- Missing capabilities: none")
|
||||
.replace("- PRs closed: none", "- PRs closed: #99")
|
||||
.replace(
|
||||
"- Blocker: missing gitea.pr.close capability", "- Blocker: none"
|
||||
)
|
||||
)
|
||||
return report
|
||||
|
||||
def test_full_proof_close_passes(self):
|
||||
result = assess_final_report_validator(
|
||||
self._closed_report(), "reconcile_already_landed"
|
||||
)
|
||||
self.assertEqual(result["grade"], "A")
|
||||
self.assertFalse(
|
||||
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_close_without_capability_proof_blocks(self):
|
||||
# Claims PRs closed: #99 but never proves gitea.pr.close capability.
|
||||
report = _reconcile_handoff().replace("- PRs closed: none", "- PRs closed: #99")
|
||||
result = assess_final_report_validator(report, "reconcile_already_landed")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_close_without_ancestor_proof_blocks(self):
|
||||
report = self._closed_report(drop=("Ancestor proof",))
|
||||
result = assess_final_report_validator(report, "reconcile_already_landed")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_close_without_linked_issue_result_blocks(self):
|
||||
report = self._closed_report(drop=("Linked issue live status", "Issues closed"))
|
||||
result = assess_final_report_validator(report, "reconcile_already_landed")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_close_lock_requires_proof_even_if_text_says_none(self):
|
||||
# Session lock proves a PR was closed; the report omits close proof.
|
||||
result = assess_final_report_validator(
|
||||
_reconcile_handoff(),
|
||||
"reconcile_already_landed",
|
||||
reconciler_close_lock={"pr_closed": True},
|
||||
)
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_comment_only_reconcile_needs_no_close_proof(self):
|
||||
result = assess_final_report_validator(
|
||||
_reconcile_handoff(), "reconcile_already_landed"
|
||||
)
|
||||
self.assertEqual(result["grade"], "A")
|
||||
self.assertFalse(
|
||||
any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
|
||||
)
|
||||
|
||||
|
||||
class TestEntryPoint(unittest.TestCase):
|
||||
def test_unknown_task_kind_blocks(self):
|
||||
result = assess_final_report_validator("report", "unknown_mode")
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
"""Tests for controller issue-acceptance gate (#500)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import issue_acceptance_gate # noqa: E402
|
||||
from final_report_validator import assess_final_report_validator # noqa: E402
|
||||
|
||||
|
||||
def _accepted_comment(**overrides):
|
||||
lines = [
|
||||
"## Controller Issue Acceptance",
|
||||
"",
|
||||
"STATE:",
|
||||
"accepted",
|
||||
"",
|
||||
"WHO_IS_NEXT:",
|
||||
"controller",
|
||||
"",
|
||||
"NEXT_ACTION:",
|
||||
"Close the tracker follow-up after verification.",
|
||||
"",
|
||||
"NEXT_PROMPT:",
|
||||
"Verify deployment checklist for issue #500 and post acceptance.",
|
||||
"",
|
||||
"ISSUE:",
|
||||
"#500",
|
||||
"",
|
||||
"MERGED_PR:",
|
||||
"#503",
|
||||
"",
|
||||
"MERGE_COMMIT:",
|
||||
"0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
"",
|
||||
"ACCEPTANCE_CRITERIA_CHECKED:",
|
||||
"- [x] documentation added",
|
||||
"- [x] validator tests added",
|
||||
"",
|
||||
"VALIDATION_REVIEWED:",
|
||||
"pytest tests/test_issue_acceptance_gate.py -q",
|
||||
"",
|
||||
"CONTROLLER_DECISION:",
|
||||
"accepted",
|
||||
"",
|
||||
"WHY:",
|
||||
"All acceptance criteria satisfied with proof.",
|
||||
"",
|
||||
"MISSING_WORK:",
|
||||
"none",
|
||||
"",
|
||||
"FOLLOW_UP_ISSUES:",
|
||||
"none",
|
||||
"",
|
||||
"BLOCKERS:",
|
||||
"none",
|
||||
"",
|
||||
"LAST_UPDATED_BY:",
|
||||
"jcwalker3 / prgs-controller / 2026-07-08",
|
||||
]
|
||||
text = "\n".join(lines)
|
||||
for key, value in overrides.items():
|
||||
text = text.replace(f"{key}:\n", f"{key}:\n{value}\n", 1)
|
||||
return text
|
||||
|
||||
|
||||
def _rejection_comment(state="needs-tests"):
|
||||
text = _accepted_comment()
|
||||
text = text.replace("STATE:\naccepted", f"STATE:\n{state}")
|
||||
text = text.replace(
|
||||
"CONTROLLER_DECISION:\naccepted",
|
||||
"CONTROLLER_DECISION:\nrejected",
|
||||
)
|
||||
text = text.replace(
|
||||
"ACCEPTANCE_CRITERIA_CHECKED:\n- [x] documentation added\n- [x] validator tests added",
|
||||
"ACCEPTANCE_CRITERIA_CHECKED:\n- [ ] regression tests for rejection paths",
|
||||
)
|
||||
text = text.replace(
|
||||
"MISSING_WORK:\nnone",
|
||||
"MISSING_WORK:\nAdd regression tests for controller rejection paths.",
|
||||
)
|
||||
text = text.replace(
|
||||
"NEXT_PROMPT:\nVerify deployment checklist for issue #500 and post acceptance.",
|
||||
"NEXT_PROMPT:\nImplement the missing tests for issue #500 and reopen the PR.",
|
||||
)
|
||||
return text
|
||||
|
||||
|
||||
class TestControllerAcceptanceComment(unittest.TestCase):
|
||||
def test_accepted_comment_valid(self):
|
||||
result = issue_acceptance_gate.validate_controller_acceptance_comment(
|
||||
_accepted_comment()
|
||||
)
|
||||
self.assertTrue(result["valid"], result["reasons"])
|
||||
|
||||
def test_missing_who_is_next_rejected(self):
|
||||
text = _accepted_comment().replace("WHO_IS_NEXT:\ncontroller\n", "WHO_IS_NEXT:\n\n")
|
||||
result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
|
||||
self.assertFalse(result["valid"])
|
||||
self.assertTrue(any("WHO_IS_NEXT" in r for r in result["reasons"]))
|
||||
|
||||
def test_missing_next_prompt_on_rejection(self):
|
||||
text = _rejection_comment()
|
||||
text = text.replace(
|
||||
"NEXT_PROMPT:\nImplement the missing tests for issue #500 and reopen the PR.\n",
|
||||
"NEXT_PROMPT:\n\n",
|
||||
)
|
||||
result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
|
||||
self.assertFalse(result["valid"])
|
||||
self.assertTrue(any("NEXT_PROMPT" in r for r in result["reasons"]))
|
||||
|
||||
def test_vague_merge_only_completion_detected(self):
|
||||
text = "PR merged. Issue complete."
|
||||
self.assertTrue(issue_acceptance_gate.claims_merge_only_issue_complete(text))
|
||||
|
||||
def test_rejection_paths_require_missing_work(self):
|
||||
for state in (
|
||||
"more-work-required",
|
||||
"needs-tests",
|
||||
"needs-docs",
|
||||
"needs-feature-enhancement",
|
||||
"needs-follow-up-issue",
|
||||
):
|
||||
text = _rejection_comment(state=state).replace(
|
||||
"MISSING_WORK:\nAdd regression tests for controller rejection paths.\n",
|
||||
"MISSING_WORK:\n\n",
|
||||
)
|
||||
result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
|
||||
self.assertFalse(result["valid"], state)
|
||||
self.assertTrue(any("MISSING_WORK" in r for r in result["reasons"]))
|
||||
|
||||
def test_accepted_requires_checked_criteria(self):
|
||||
text = _accepted_comment().replace(
|
||||
"ACCEPTANCE_CRITERIA_CHECKED:\n- [x] documentation added\n- [x] validator tests added\n",
|
||||
"ACCEPTANCE_CRITERIA_CHECKED:\n- [ ] documentation added\n",
|
||||
)
|
||||
result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
|
||||
self.assertFalse(result["valid"])
|
||||
self.assertTrue(any("checked acceptance criterion" in r for r in result["reasons"]))
|
||||
|
||||
|
||||
class TestFinalReportIntegration(unittest.TestCase):
|
||||
def test_merge_only_complete_blocks_work_issue_report(self):
|
||||
report = (
|
||||
"## Controller Handoff\n\n"
|
||||
"- Task: work issue #500\n"
|
||||
"- Merge result: merged\n"
|
||||
"- Current status: PR merged; issue complete\n"
|
||||
)
|
||||
result = assess_final_report_validator(report, "work_issue")
|
||||
self.assertTrue(
|
||||
any(
|
||||
f["rule_id"] == "shared.issue_acceptance_gate"
|
||||
for f in result["findings"]
|
||||
),
|
||||
result["findings"],
|
||||
)
|
||||
|
||||
def test_pending_acceptance_allowed(self):
|
||||
report = (
|
||||
"## Controller Handoff\n\n"
|
||||
"- Task: work issue #500\n"
|
||||
"- Merge result: merged\n"
|
||||
"- Current status: controller acceptance pending\n"
|
||||
)
|
||||
result = assess_final_report_validator(report, "work_issue")
|
||||
self.assertFalse(
|
||||
any(
|
||||
f["rule_id"] == "shared.issue_acceptance_gate"
|
||||
for f in result["findings"]
|
||||
),
|
||||
result["findings"],
|
||||
)
|
||||
|
||||
def test_accepted_block_allows_complete_claim(self):
|
||||
report = _accepted_comment() + "\n\nIssue is complete."
|
||||
gate = issue_acceptance_gate.validate_final_report_issue_acceptance(report)
|
||||
self.assertTrue(gate["valid"], gate["reasons"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,207 @@
|
||||
"""Tests for durable MCP session state shared across daemon processes (#559)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys_path_root = str(Path(__file__).resolve().parent.parent)
|
||||
import sys
|
||||
|
||||
if sys_path_root not in sys.path:
|
||||
sys.path.insert(0, sys_path_root)
|
||||
|
||||
import mcp_session_state
|
||||
import review_workflow_load
|
||||
import mcp_server
|
||||
|
||||
|
||||
class TestMcpSessionStateStore(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmpdir = tempfile.TemporaryDirectory()
|
||||
self.state_dir = self._tmpdir.name
|
||||
self._env = patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
mcp_session_state.STATE_DIR_ENV: self.state_dir,
|
||||
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
|
||||
"GITEA_MCP_PROFILE": "prgs-reviewer",
|
||||
},
|
||||
clear=False,
|
||||
)
|
||||
self._env.start()
|
||||
|
||||
def tearDown(self):
|
||||
self._env.stop()
|
||||
self._tmpdir.cleanup()
|
||||
|
||||
def test_round_trip_same_identity(self):
|
||||
saved = mcp_session_state.save_state(
|
||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
||||
payload={"loaded": True, "workflow_hash": "abc123def456"},
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertIsNotNone(saved)
|
||||
self.assertEqual(saved["workflow_hash"], "abc123def456")
|
||||
self.assertIn("recorded_at", saved)
|
||||
|
||||
loaded = mcp_session_state.load_state(
|
||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertIsNotNone(loaded)
|
||||
self.assertEqual(loaded["workflow_hash"], "abc123def456")
|
||||
self.assertEqual(loaded["profile_identity"], "prgs-reviewer")
|
||||
|
||||
def test_profile_mismatch_returns_none(self):
|
||||
mcp_session_state.save_state(
|
||||
kind=mcp_session_state.KIND_DECISION_LOCK,
|
||||
payload={"final_review_decision_ready": True, "remote": "prgs"},
|
||||
remote="prgs",
|
||||
)
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
|
||||
clear=False,
|
||||
):
|
||||
loaded = mcp_session_state.load_state(
|
||||
kind=mcp_session_state.KIND_DECISION_LOCK,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIsNone(loaded)
|
||||
|
||||
def test_clear_removes_file(self):
|
||||
mcp_session_state.save_state(
|
||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
||||
payload={"loaded": True},
|
||||
remote="prgs",
|
||||
)
|
||||
path = mcp_session_state.state_file_path(
|
||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
||||
remote="prgs",
|
||||
profile_identity="prgs-reviewer",
|
||||
state_dir=self.state_dir,
|
||||
)
|
||||
self.assertTrue(os.path.exists(path))
|
||||
mcp_session_state.clear_state(
|
||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
||||
remote="prgs",
|
||||
profile_identity="prgs-reviewer",
|
||||
)
|
||||
self.assertFalse(os.path.exists(path))
|
||||
|
||||
def test_files_are_private_mode(self):
|
||||
mcp_session_state.save_state(
|
||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
||||
payload={"loaded": True},
|
||||
remote="prgs",
|
||||
)
|
||||
path = mcp_session_state.state_file_path(
|
||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
||||
remote="prgs",
|
||||
profile_identity="prgs-reviewer",
|
||||
state_dir=self.state_dir,
|
||||
)
|
||||
mode = os.stat(path).st_mode & 0o777
|
||||
self.assertEqual(mode, 0o600)
|
||||
|
||||
|
||||
class TestWorkflowLoadCrossProcess(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmpdir = tempfile.TemporaryDirectory()
|
||||
self._env = patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
|
||||
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
|
||||
"GITEA_MCP_PROFILE": "prgs-reviewer",
|
||||
"GITEA_MCP_REMOTE": "prgs",
|
||||
},
|
||||
clear=False,
|
||||
)
|
||||
self._env.start()
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
|
||||
def tearDown(self):
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
self._env.stop()
|
||||
self._tmpdir.cleanup()
|
||||
|
||||
def test_different_pid_same_profile_accepted(self):
|
||||
root = str(Path(__file__).resolve().parent.parent)
|
||||
recorded = review_workflow_load.record_review_workflow_load(root)
|
||||
self.assertTrue(recorded["loaded"])
|
||||
|
||||
# Simulate another daemon process: clear memory, keep durable file,
|
||||
# and change apparent PID identity only (profile remains the same).
|
||||
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
|
||||
status = review_workflow_load.workflow_load_status(root)
|
||||
self.assertTrue(status["workflow_load_proof_present"])
|
||||
self.assertTrue(
|
||||
status["workflow_load_valid"],
|
||||
msg=status.get("reasons"),
|
||||
)
|
||||
|
||||
def test_profile_mismatch_blocks(self):
|
||||
root = str(Path(__file__).resolve().parent.parent)
|
||||
review_workflow_load.record_review_workflow_load(root)
|
||||
review_workflow_load._REVIEW_WORKFLOW_LOAD = None
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
|
||||
clear=False,
|
||||
):
|
||||
# Durable load uses active identity; mismatched profile key misses.
|
||||
status = review_workflow_load.workflow_load_status(root)
|
||||
self.assertFalse(status["workflow_load_proof_present"])
|
||||
|
||||
|
||||
class TestDecisionLockCrossProcess(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmpdir = tempfile.TemporaryDirectory()
|
||||
self._env = patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
|
||||
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
|
||||
},
|
||||
clear=False,
|
||||
)
|
||||
self._env.start()
|
||||
mcp_server._save_review_decision_lock(None)
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
|
||||
def tearDown(self):
|
||||
mcp_server._save_review_decision_lock(None)
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
self._env.stop()
|
||||
self._tmpdir.cleanup()
|
||||
|
||||
def test_decision_lock_survives_memory_clear(self):
|
||||
with patch.object(mcp_server, "get_profile", return_value={
|
||||
"profile_name": "prgs-reviewer",
|
||||
}):
|
||||
mcp_server.init_review_decision_lock("prgs", "review_pr", force=True)
|
||||
lock = mcp_server._load_review_decision_lock()
|
||||
self.assertIsNotNone(lock)
|
||||
self.assertEqual(lock["remote"], "prgs")
|
||||
self.assertFalse(lock["final_review_decision_ready"])
|
||||
|
||||
# New process: memory empty, durable state remains.
|
||||
mcp_server._REVIEW_DECISION_LOCK = None
|
||||
restored = mcp_server._load_review_decision_lock()
|
||||
self.assertIsNotNone(restored)
|
||||
self.assertEqual(restored["remote"], "prgs")
|
||||
reasons = mcp_server._review_decision_session_reasons(restored)
|
||||
self.assertEqual(reasons, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,71 @@
|
||||
import os
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from datetime import datetime
|
||||
|
||||
import gitea_mcp_server
|
||||
|
||||
class TestMcpStaleRuntime(unittest.TestCase):
|
||||
@patch("subprocess.run")
|
||||
@patch("os.path.getmtime")
|
||||
@patch("os.path.exists")
|
||||
@patch("os.getpid")
|
||||
@patch.dict("os.environ", {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"})
|
||||
def test_stale_and_missing_runtimes(self, mock_getpid, mock_exists, mock_getmtime, mock_run):
|
||||
# Setup mocks
|
||||
mock_getpid.return_value = 12345
|
||||
mock_exists.return_value = True
|
||||
|
||||
# Code modification time: Jul 8 2026, 14:00:00
|
||||
code_time = datetime(2026, 7, 8, 14, 0, 0)
|
||||
mock_getmtime.return_value = code_time.timestamp()
|
||||
|
||||
# Mock ps -ax output
|
||||
# PID 12345 is self (started at 13:00:00 - stale)
|
||||
# PID 54321 is prgs-author (started at 15:00:00 - fresh)
|
||||
# prgs-reviewer is missing
|
||||
ps_output = (
|
||||
" PID LSTART COMMAND\n"
|
||||
"12345 Wed Jul 8 13:00:00 2026 /path/to/python mcp_server.py\n"
|
||||
"54321 Wed Jul 8 15:00:00 2026 /path/to/python mcp_server.py\n"
|
||||
)
|
||||
|
||||
mock_run_ps = MagicMock()
|
||||
mock_run_ps.stdout = ps_output
|
||||
|
||||
# Mock env output for ps eww
|
||||
mock_run_env12345 = MagicMock()
|
||||
mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
|
||||
|
||||
mock_run_env54321 = MagicMock()
|
||||
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
|
||||
|
||||
def side_effect(args, **kwargs):
|
||||
if args[0] == "ps" and "eww" in args:
|
||||
pid = args[2]
|
||||
if pid == "12345":
|
||||
return mock_run_env12345
|
||||
elif pid == "54321":
|
||||
return mock_run_env54321
|
||||
elif args[0] == "ps":
|
||||
return mock_run_ps
|
||||
raise ValueError(f"Unexpected subprocess run args: {args}")
|
||||
|
||||
mock_run.side_effect = side_effect
|
||||
|
||||
# Test 1: required reviewer role matching prgs-reviewer (which is missing)
|
||||
reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("review_pr", ["prgs-reviewer"])
|
||||
|
||||
self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
|
||||
self.assertTrue(any("stale-runtime: None of the matching profiles for task" in r for r in reasons))
|
||||
|
||||
# Test 2: required author role matching prgs-author (which is fresh)
|
||||
reasons_author = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
|
||||
|
||||
# Still contains self stale error
|
||||
self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons_author))
|
||||
# But does NOT contain missing author profile error
|
||||
self.assertFalse(any("None of the matching profiles for task" in r for r in reasons_author))
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -33,12 +33,38 @@ class TestReviewWorkflowLoadModule(unittest.TestCase):
|
||||
self.assertTrue(status["workflow_load_proof_present"])
|
||||
self.assertTrue(status["workflow_load_valid"])
|
||||
|
||||
def test_stale_session_pid_blocks(self):
|
||||
def test_profile_identity_mismatch_blocks(self):
|
||||
"""#559: different daemon PIDs are OK; profile identity mismatch is not."""
|
||||
import tempfile
|
||||
from unittest.mock import patch
|
||||
|
||||
import mcp_session_state
|
||||
|
||||
root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
|
||||
review_workflow_load.record_review_workflow_load(root)
|
||||
review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0
|
||||
blockers = review_workflow_load.review_workflow_load_blockers(root)
|
||||
self.assertTrue(any("different process" in b for b in blockers))
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
mcp_session_state.STATE_DIR_ENV: tmp,
|
||||
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
|
||||
},
|
||||
clear=False,
|
||||
):
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
review_workflow_load.record_review_workflow_load(root)
|
||||
# Corrupt the in-memory profile identity while keeping PID.
|
||||
review_workflow_load._REVIEW_WORKFLOW_LOAD[
|
||||
"session_profile_lock"
|
||||
] = "other-profile"
|
||||
review_workflow_load._REVIEW_WORKFLOW_LOAD[
|
||||
"profile_identity"
|
||||
] = "other-profile"
|
||||
blockers = review_workflow_load.review_workflow_load_blockers(root)
|
||||
self.assertTrue(
|
||||
any("profile identity mismatch" in b for b in blockers),
|
||||
msg=blockers,
|
||||
)
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
|
||||
def test_prompt_conflict_detected(self):
|
||||
conflict, reasons = review_workflow_load.assess_prompt_conflict(
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Tests for web UI lease visibility (#433)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui.app import create_app
|
||||
from webui.lease_loader import (
|
||||
_duplicate_branch_warnings,
|
||||
_duplicate_pr_warnings,
|
||||
load_lease_snapshot,
|
||||
parse_reviewer_lease_comment,
|
||||
snapshot_to_dict,
|
||||
)
|
||||
|
||||
|
||||
class TestLeaseLoader(unittest.TestCase):
|
||||
def test_parse_reviewer_lease_comment(self):
|
||||
body = (
|
||||
"<!-- mcp-review-lease:v1 -->\n"
|
||||
"pr: #42\n"
|
||||
"reviewer_identity: sysadmin\n"
|
||||
"profile: prgs-reviewer\n"
|
||||
"phase: validating\n"
|
||||
"expires_at: 2026-07-07T20:00:00Z\n"
|
||||
)
|
||||
parsed = parse_reviewer_lease_comment(body)
|
||||
self.assertIsNotNone(parsed)
|
||||
assert parsed is not None
|
||||
self.assertEqual(parsed["pr_number"], 42)
|
||||
self.assertEqual(parsed["phase"], "validating")
|
||||
|
||||
def test_duplicate_pr_warnings(self):
|
||||
raw_prs = [
|
||||
{"number": 1, "title": "Closes #99", "body": ""},
|
||||
{"number": 2, "title": "fixes #99", "body": ""},
|
||||
]
|
||||
warnings = _duplicate_pr_warnings(raw_prs)
|
||||
self.assertEqual(len(warnings), 1)
|
||||
self.assertEqual(warnings[0].issue_number, 99)
|
||||
self.assertEqual(warnings[0].pr_numbers, (1, 2))
|
||||
|
||||
def test_duplicate_branch_warnings(self):
|
||||
warnings = _duplicate_branch_warnings([
|
||||
"feat/issue-12-a",
|
||||
"feat/issue-12-b",
|
||||
"master",
|
||||
])
|
||||
self.assertEqual(len(warnings), 1)
|
||||
self.assertEqual(warnings[0].issue_number, 12)
|
||||
|
||||
def test_load_snapshot_with_injected_fetch(self):
|
||||
def fetch_prs(_h, _o, _r, _a):
|
||||
return ([], None)
|
||||
|
||||
def fetch_issues(_h, _o, _r, _a):
|
||||
return ([], None)
|
||||
|
||||
snapshot = load_lease_snapshot(
|
||||
fetch_prs=fetch_prs,
|
||||
fetch_issues=fetch_issues,
|
||||
fetch_comments=lambda *_a, **_k: [],
|
||||
issue_lock_path=str(Path("/nonexistent/lock.json")),
|
||||
)
|
||||
data = snapshot_to_dict(snapshot)
|
||||
self.assertIn("claim_inventory", data)
|
||||
self.assertIn("collision_history", data)
|
||||
|
||||
|
||||
class TestLeaseRoutes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_leases_page_renders(self):
|
||||
response = self.client.get("/leases")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Leases", response.text)
|
||||
self.assertIn("Collision warnings", response.text)
|
||||
self.assertNotIn("child issue", response.text.lower())
|
||||
|
||||
def test_api_leases_json(self):
|
||||
response = self.client.get("/api/leases")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertIn("claim_inventory", data)
|
||||
self.assertIn("duplicate_prs", data)
|
||||
self.assertIn("reviewer_leases", data)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -46,10 +46,14 @@ class TestWebuiSkeleton(unittest.TestCase):
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Gitea-Tools", response.text)
|
||||
|
||||
def test_leases_is_implemented(self):
|
||||
response = self.client.get("/leases")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Leases", response.text)
|
||||
self.assertIn("Collision warnings", response.text)
|
||||
|
||||
def test_extra_stub_routes(self):
|
||||
for path in ("/worktrees", "/leases"):
|
||||
with self.subTest(path=path):
|
||||
self.assertEqual(self.client.get(path).status_code, 200)
|
||||
self.assertEqual(self.client.get("/worktrees").status_code, 200)
|
||||
|
||||
def test_post_is_rejected(self):
|
||||
response = self.client.post("/health")
|
||||
@@ -62,10 +66,10 @@ class TestWebuiSkeleton(unittest.TestCase):
|
||||
self.assertIn("Live queue", response.text)
|
||||
|
||||
def test_nav_links_on_all_pages(self):
|
||||
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
|
||||
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
|
||||
with self.subTest(path=path):
|
||||
text = self.client.get(path).text
|
||||
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
|
||||
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user