Merge branch 'master' into feat/issue-285-live-infra-stop-recompute
This commit is contained in:
@@ -0,0 +1,102 @@
|
||||
"""Identity-disclosure checks for workflow reports (#305).
|
||||
|
||||
Workflow reports sometimes included the authenticated user's personal
|
||||
email even though username/profile identity is sufficient (observed:
|
||||
``Identity: jcwalker3 / [email protected]`` in a reconciliation
|
||||
handoff). These tests pin the no-email identity summary helper and the
|
||||
final-report validator that flags unnecessary email disclosure.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import review_proofs
|
||||
|
||||
|
||||
class TestIdentitySummary(unittest.TestCase):
|
||||
def test_author_summary_uses_username_and_profile_only(self):
|
||||
summary = review_proofs.format_identity_summary(
|
||||
"jcwalker3", "prgs-author")
|
||||
self.assertEqual(summary, "jcwalker3 / prgs-author")
|
||||
self.assertNotIn("@", summary)
|
||||
|
||||
def test_reviewer_summary_uses_username_and_profile_only(self):
|
||||
summary = review_proofs.format_identity_summary(
|
||||
"sysadmin", "prgs-reviewer")
|
||||
self.assertEqual(summary, "sysadmin / prgs-reviewer")
|
||||
|
||||
def test_summary_appends_role_and_remote_without_email(self):
|
||||
summary = review_proofs.format_identity_summary(
|
||||
"jcwalker3", "prgs-author", role="author", remote="prgs")
|
||||
self.assertIn("jcwalker3 / prgs-author", summary)
|
||||
self.assertIn("author", summary)
|
||||
self.assertIn("prgs", summary)
|
||||
self.assertNotIn("@", summary)
|
||||
|
||||
def test_summary_never_leaks_email_passed_as_username(self):
|
||||
"""Defense in depth: an email in the username slot is reduced."""
|
||||
summary = review_proofs.format_identity_summary(
|
||||
"[email protected]", "prgs-author")
|
||||
self.assertNotIn("@", summary)
|
||||
self.assertIn("jcwalker3", summary)
|
||||
|
||||
|
||||
class TestEmailDisclosureAssessment(unittest.TestCase):
|
||||
def test_report_without_email_passes(self):
|
||||
report = "\n".join([
|
||||
"Controller Handoff",
|
||||
"Identity: jcwalker3 / prgs-author",
|
||||
"Role: author",
|
||||
])
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertTrue(res["proven"])
|
||||
self.assertFalse(res["flagged"])
|
||||
self.assertEqual(res["emails"], [])
|
||||
self.assertEqual(res["reasons"], [])
|
||||
|
||||
def test_unnecessary_email_is_flagged(self):
|
||||
report = "\n".join([
|
||||
"Controller Handoff",
|
||||
"Identity: jcwalker3 / [email protected]",
|
||||
])
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertTrue(res["flagged"])
|
||||
self.assertIn("[email protected]", res["emails"])
|
||||
self.assertTrue(res["reasons"])
|
||||
self.assertFalse(res["justified"])
|
||||
|
||||
def test_multiple_emails_all_reported(self):
|
||||
report = (
|
||||
"Identity: [email protected] author\n"
|
||||
"Reviewer: [email protected]\n"
|
||||
)
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertTrue(res["flagged"])
|
||||
self.assertEqual(
|
||||
sorted(res["emails"]),
|
||||
["[email protected]", "[email protected]"],
|
||||
)
|
||||
|
||||
def test_justified_email_with_explanation_is_not_flagged(self):
|
||||
report = "\n".join([
|
||||
"Identity: jcwalker3 / prgs-author",
|
||||
"Contact email: [email protected]",
|
||||
"Email required because two accounts share the username and the",
|
||||
"address is necessary to disambiguate identity.",
|
||||
])
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertFalse(res["flagged"])
|
||||
self.assertTrue(res["justified"])
|
||||
self.assertIn("[email protected]", res["emails"])
|
||||
|
||||
def test_email_without_justification_language_is_flagged(self):
|
||||
report = "Contact email: [email protected] just in case.\n"
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertTrue(res["flagged"])
|
||||
self.assertFalse(res["justified"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -36,7 +36,32 @@ class TestIssueLockWorktreeAssessment(unittest.TestCase):
|
||||
porcelain_status="",
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("issue lock must be taken from base branch", result["reasons"][0])
|
||||
self.assertIn("base-equivalence could not be proven", result["reasons"][0])
|
||||
|
||||
def test_base_equivalent_feature_branch_passes(self):
|
||||
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path="/repo/branches/issue-275",
|
||||
current_branch="feat/issue-275-claim-lock-branches-worktree",
|
||||
porcelain_status="",
|
||||
base_equivalent=True,
|
||||
inspected_git_root="/repo/branches/issue-275",
|
||||
base_branch="origin/master",
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["base_branch"], "origin/master")
|
||||
|
||||
def test_non_base_equivalent_branch_fails(self):
|
||||
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path="/repo/branches/issue-275",
|
||||
current_branch="feat/issue-275-claim-lock-branches-worktree",
|
||||
porcelain_status="",
|
||||
base_equivalent=False,
|
||||
inspected_git_root="/repo/branches/issue-275",
|
||||
base_branch=None,
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("must be base-equivalent", result["reasons"][0])
|
||||
|
||||
def test_untracked_files_do_not_block(self):
|
||||
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
@@ -96,4 +121,4 @@ class TestPrWorktreeMatch(unittest.TestCase):
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Doc-contract checks for task-specific LLM workflow split (#333)."""
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
SKILL_DIR = REPO_ROOT / "skills" / "llm-project-workflow"
|
||||
SKILL = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_skill_md_exists():
|
||||
assert SKILL_DIR.joinpath("SKILL.md").is_file()
|
||||
|
||||
|
||||
def test_all_workflow_files_exist():
|
||||
for name in (
|
||||
"review-merge-pr.md",
|
||||
"reconcile-landed-pr.md",
|
||||
"create-issue.md",
|
||||
"work-issue.md",
|
||||
):
|
||||
assert (SKILL_DIR / "workflows" / name).is_file(), name
|
||||
|
||||
|
||||
def test_skill_references_all_workflow_files():
|
||||
for name in (
|
||||
"workflows/review-merge-pr.md",
|
||||
"workflows/reconcile-landed-pr.md",
|
||||
"workflows/create-issue.md",
|
||||
"workflows/work-issue.md",
|
||||
):
|
||||
assert name in SKILL
|
||||
|
||||
|
||||
def test_skill_contains_mode_isolation_language():
|
||||
assert "## Mode isolation" in SKILL
|
||||
assert "review-merge-pr" in SKILL
|
||||
assert "reconcile-landed-pr" in SKILL
|
||||
assert "create-issue" in SKILL
|
||||
assert "work-issue" in SKILL
|
||||
assert "Do not mix modes" in SKILL or "do not mix modes" in SKILL.lower()
|
||||
|
||||
|
||||
def test_skill_is_router_not_monolithic_review_body():
|
||||
assert "This skill is a **router**" in SKILL or "router" in SKILL.lower()
|
||||
assert "## F. Review workflow" not in SKILL
|
||||
assert "gitea_mark_final_review_decision" not in SKILL
|
||||
assert "gitea_submit_pr_review" not in SKILL
|
||||
|
||||
|
||||
def test_review_merge_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "review-merge-pr.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
assert "## 26A. Terminal review mutation hard-stop" in text
|
||||
assert "## 11A. Skipped PRs are read-only" in text
|
||||
assert "## 35. Duplicate request-changes prevention" in text
|
||||
assert "ALREADY_LANDED_RECONCILE_REQUIRED" in text
|
||||
|
||||
|
||||
def test_reconcile_landed_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "reconcile-landed-pr.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
assert "Do not review or merge normal PRs" in text
|
||||
assert "Already-landed proof" in text
|
||||
|
||||
|
||||
def test_create_issue_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "create-issue.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
assert "## 9. Duplicate search before mutation" in text
|
||||
|
||||
|
||||
def test_work_issue_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
assert "Do not merge your own PR" in text
|
||||
assert "## 21. PR creation rules" in text
|
||||
|
||||
|
||||
def test_final_report_schemas_exist():
|
||||
for name in (
|
||||
"review-merge-final-report.md",
|
||||
"reconcile-landed-final-report.md",
|
||||
"create-issue-final-report.md",
|
||||
"work-issue-final-report.md",
|
||||
):
|
||||
assert (SKILL_DIR / "schemas" / name).is_file(), name
|
||||
|
||||
|
||||
def test_review_pr_template_references_workflow():
|
||||
text = (SKILL_DIR / "templates" / "review-pr.md").read_text(encoding="utf-8")
|
||||
assert "workflows/review-merge-pr.md" in text
|
||||
|
||||
|
||||
def test_merge_pr_template_references_workflow():
|
||||
text = (SKILL_DIR / "templates" / "merge-pr.md").read_text(encoding="utf-8")
|
||||
assert "workflows/review-merge-pr.md" in text
|
||||
|
||||
|
||||
def test_start_issue_template_references_work_issue_workflow():
|
||||
text = (SKILL_DIR / "templates" / "start-issue.md").read_text(encoding="utf-8")
|
||||
assert "workflows/work-issue.md" in text
|
||||
@@ -2937,20 +2937,31 @@ class TestVerifyMutationAuthority(unittest.TestCase):
|
||||
mcp_server.verify_mutation_authority("prgs")
|
||||
|
||||
|
||||
def _clean_master_git_state_for_lock():
|
||||
return {
|
||||
"current_branch": "master",
|
||||
"porcelain_status": "",
|
||||
"base_equivalent": True,
|
||||
"inspected_git_root": "/scratch/wt",
|
||||
"base_branch": "origin/master",
|
||||
}
|
||||
|
||||
|
||||
class TestIssueLocking(unittest.TestCase):
|
||||
"""Test issue locking and PR gating constraints."""
|
||||
|
||||
@staticmethod
|
||||
def _clean_master_git_state():
|
||||
return {"current_branch": "master", "porcelain_status": ""}
|
||||
def setUp(self):
|
||||
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
|
||||
self._env_patcher.start()
|
||||
|
||||
def tearDown(self):
|
||||
self._env_patcher.stop()
|
||||
if os.path.exists(ISSUE_LOCK_FILE):
|
||||
os.remove(ISSUE_LOCK_FILE)
|
||||
|
||||
@patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={"current_branch": "master", "porcelain_status": ""},
|
||||
return_value=_clean_master_git_state_for_lock(),
|
||||
)
|
||||
@patch("mcp_server.api_get_all")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@@ -2970,7 +2981,7 @@ class TestIssueLocking(unittest.TestCase):
|
||||
|
||||
@patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={"current_branch": "master", "porcelain_status": ""},
|
||||
return_value=_clean_master_git_state_for_lock(),
|
||||
)
|
||||
@patch("mcp_server.api_get_all")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@@ -2987,7 +2998,7 @@ class TestIssueLocking(unittest.TestCase):
|
||||
|
||||
@patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={"current_branch": "master", "porcelain_status": ""},
|
||||
return_value=_clean_master_git_state_for_lock(),
|
||||
)
|
||||
@patch("mcp_server.api_get_all")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@@ -3008,7 +3019,7 @@ class TestIssueLocking(unittest.TestCase):
|
||||
scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean"
|
||||
with patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={"current_branch": "master", "porcelain_status": ""},
|
||||
return_value=_clean_master_git_state_for_lock(),
|
||||
) as mock_git:
|
||||
res = gitea_lock_issue(
|
||||
issue_number=249,
|
||||
@@ -3028,6 +3039,7 @@ class TestIssueLocking(unittest.TestCase):
|
||||
return_value={
|
||||
"current_branch": "master",
|
||||
"porcelain_status": " M gitea_mcp_server.py\n",
|
||||
"base_equivalent": True,
|
||||
},
|
||||
):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
@@ -3047,6 +3059,7 @@ class TestIssueLocking(unittest.TestCase):
|
||||
return_value={
|
||||
"current_branch": "feat/issue-243-forbidden-git-gaps",
|
||||
"porcelain_status": "",
|
||||
"base_equivalent": False,
|
||||
},
|
||||
):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
@@ -3056,7 +3069,7 @@ class TestIssueLocking(unittest.TestCase):
|
||||
remote="prgs",
|
||||
worktree_path="/tmp/scratch/wt",
|
||||
)
|
||||
self.assertIn("issue lock must be taken from base branch", str(ctx.exception))
|
||||
self.assertIn("base-equivalent", str(ctx.exception))
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@@ -3199,7 +3212,12 @@ class TestPreflightVerification(unittest.TestCase):
|
||||
mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
|
||||
mcp_server._preflight_capability_violation_files = self.orig_capability_files
|
||||
mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
|
||||
for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
|
||||
for key in (
|
||||
"GITEA_TEST_FORCE_DIRTY",
|
||||
"GITEA_TEST_PORCELAIN",
|
||||
"GITEA_ACTIVE_WORKTREE",
|
||||
"GITEA_AUTHOR_WORKTREE",
|
||||
):
|
||||
if key in os.environ:
|
||||
del os.environ[key]
|
||||
|
||||
@@ -3289,3 +3307,35 @@ class TestPreflightVerification(unittest.TestCase):
|
||||
status = mcp_server.assess_preflight_status()
|
||||
self.assertFalse(status["preflight_ready"])
|
||||
self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])
|
||||
|
||||
def test_declared_clean_task_worktree_ignores_control_checkout_violation(self):
|
||||
"""#275: active branches/ worktree is the inspected mutation workspace."""
|
||||
import mcp_server
|
||||
mcp_server._preflight_whoami_called = True
|
||||
mcp_server._preflight_capability_called = True
|
||||
mcp_server._preflight_whoami_violation = True
|
||||
mcp_server._preflight_whoami_violation_files = ["mcp_server.py"]
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = ""
|
||||
|
||||
worktree = "/repo/branches/issue-275-clean"
|
||||
status = mcp_server.assess_preflight_status(worktree_path=worktree)
|
||||
self.assertTrue(status["preflight_ready"])
|
||||
self.assertEqual(status["preflight_block_reasons"], [])
|
||||
self.assertIn("preflight_workspace", status)
|
||||
mcp_server.verify_preflight_purity(worktree_path=worktree)
|
||||
|
||||
def test_declared_dirty_task_worktree_reports_workspace_diagnostics(self):
|
||||
"""#275: dirty failures name the inspected task workspace, not just files."""
|
||||
import mcp_server
|
||||
mcp_server._preflight_whoami_called = True
|
||||
mcp_server._preflight_capability_called = True
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = " M task_file.py\n"
|
||||
|
||||
worktree = "/repo/branches/issue-275-dirty"
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(worktree_path=worktree)
|
||||
msg = str(ctx.exception)
|
||||
self.assertIn("active task workspace root", msg)
|
||||
self.assertIn("inspected git root", msg)
|
||||
self.assertIn("dirty files: task_file.py", msg)
|
||||
self.assertIn("dirty scope:", msg)
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
"""Tests for final-report mutation ledger verifier (#331)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import assess_mutation_ledger_report # noqa: E402
|
||||
|
||||
|
||||
class TestMutationLedgerReport(unittest.TestCase):
|
||||
def test_none_claim_with_performed_edit_blocks(self):
|
||||
report = (
|
||||
"Review decision: request_changes.\n"
|
||||
"File edits by reviewer: none\n"
|
||||
)
|
||||
action_log = [
|
||||
{"action": "Edited", "path": "walkthrough.md", "tracked": False},
|
||||
]
|
||||
result = assess_mutation_ledger_report(report, action_log=action_log)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["file_edits_claimed_none"])
|
||||
|
||||
def test_reported_edit_passes(self):
|
||||
report = (
|
||||
"File edits by reviewer: Edited walkthrough.md (untracked)\n"
|
||||
"Mutation ledger: Edited walkthrough.md untracked in review worktree\n"
|
||||
)
|
||||
action_log = [
|
||||
{
|
||||
"action": "Edited",
|
||||
"path": "walkthrough.md",
|
||||
"tracked": False,
|
||||
"in_repo": True,
|
||||
},
|
||||
]
|
||||
result = assess_mutation_ledger_report(
|
||||
report,
|
||||
action_log=action_log,
|
||||
walkthrough_explicitly_requested=True,
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_unreported_mutation_blocks(self):
|
||||
report = "File edits by reviewer: none\n"
|
||||
action_log = [{"action": "Wrote", "path": "/tmp/review-notes.txt"}]
|
||||
result = assess_mutation_ledger_report(report, action_log=action_log)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("/tmp/review-notes.txt", result["unreported_paths"])
|
||||
|
||||
def test_outside_repo_requires_label(self):
|
||||
report = (
|
||||
"File edits by reviewer: Wrote /tmp/review-notes.txt\n"
|
||||
)
|
||||
action_log = [
|
||||
{
|
||||
"action": "Wrote",
|
||||
"path": "/tmp/review-notes.txt",
|
||||
"outside_repo": True,
|
||||
},
|
||||
]
|
||||
result = assess_mutation_ledger_report(report, action_log=action_log)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("outside repo", " ".join(result["reasons"]).lower())
|
||||
|
||||
def test_gated_rejection_excluded_from_performed(self):
|
||||
report = "File edits by reviewer: none\nRejected gated calls: submit_pr_review blocked\n"
|
||||
action_log = [
|
||||
{
|
||||
"action": "Edited",
|
||||
"path": "walkthrough.md",
|
||||
"performed": False,
|
||||
"gated_rejected": True,
|
||||
},
|
||||
]
|
||||
result = assess_mutation_ledger_report(report, action_log=action_log)
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertEqual(result["performed_mutations"], [])
|
||||
|
||||
def test_post_status_artifact_requires_final_git_status(self):
|
||||
report = (
|
||||
"File edits by reviewer: Created scratch/notes.md (untracked)\n"
|
||||
)
|
||||
action_log = [
|
||||
{
|
||||
"action": "Created",
|
||||
"path": "scratch/notes.md",
|
||||
"tracked": False,
|
||||
"after_git_status": True,
|
||||
},
|
||||
]
|
||||
result = assess_mutation_ledger_report(
|
||||
report,
|
||||
action_log=action_log,
|
||||
final_git_status_reported=False,
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("final git status", " ".join(result["reasons"]).lower())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -133,7 +133,8 @@ class TestControlPlaneGuide(GuideTestBase):
|
||||
for key in ("hard_stops", "fail_closed", "head_sha_pinning",
|
||||
"merge_confirmation", "redaction", "separation",
|
||||
"profile_switching", "identity_verification",
|
||||
"work_selection", "global_worktree"):
|
||||
"work_selection", "global_worktree",
|
||||
"shell_spawn_hard_stop"):
|
||||
self.assertIn(key, rules)
|
||||
self.assertIn("MERGE PR", json.dumps(rules["merge_confirmation"]))
|
||||
self.assertTrue(rules["hard_stops"])
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Tests for REQUEST_CHANGES override proof before approval (#326)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import assess_request_changes_approval_proof # noqa: E402
|
||||
|
||||
HEAD_A = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
HEAD_B = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||
|
||||
|
||||
def _feedback(**overrides):
|
||||
base = {
|
||||
"success": True,
|
||||
"current_head_sha": HEAD_A,
|
||||
"has_blocking_change_requests": False,
|
||||
"author_pushed_after_request_changes": False,
|
||||
"reviews": [],
|
||||
}
|
||||
base.update(overrides)
|
||||
return base
|
||||
|
||||
|
||||
def _blocking_review(**overrides):
|
||||
entry = {
|
||||
"reviewer": "reviewer-a",
|
||||
"verdict": "REQUEST_CHANGES",
|
||||
"body": "Tests fail on pinned head; fix test_commit_files_gate.",
|
||||
"submitted_at": "2026-07-07T01:00:00-05:00",
|
||||
"reviewed_head_sha": HEAD_A,
|
||||
"dismissed": False,
|
||||
"stale": False,
|
||||
}
|
||||
entry.update(overrides)
|
||||
return entry
|
||||
|
||||
|
||||
class TestRequestChangesApprovalProof(unittest.TestCase):
|
||||
def test_missing_feedback_blocks_approval(self):
|
||||
result = assess_request_changes_approval_proof(None)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_no_prior_request_changes_allows_approval(self):
|
||||
feedback = _feedback(
|
||||
reviews=[{
|
||||
"reviewer": "sysadmin",
|
||||
"verdict": "APPROVED",
|
||||
"body": "LGTM",
|
||||
"submitted_at": "2026-07-07T02:00:00-05:00",
|
||||
"reviewed_head_sha": HEAD_A,
|
||||
"dismissed": False,
|
||||
}]
|
||||
)
|
||||
result = assess_request_changes_approval_proof(feedback)
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertIsNone(result["blocking_review"])
|
||||
|
||||
def test_changed_head_since_blocker_allows_approval(self):
|
||||
feedback = _feedback(
|
||||
current_head_sha=HEAD_B,
|
||||
author_pushed_after_request_changes=True,
|
||||
has_blocking_change_requests=True,
|
||||
reviews=[_blocking_review(reviewed_head_sha=HEAD_A)],
|
||||
)
|
||||
result = assess_request_changes_approval_proof(feedback)
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertTrue(result["head_changed_since_blocker"])
|
||||
|
||||
def test_unchanged_head_without_override_blocks_approval(self):
|
||||
feedback = _feedback(
|
||||
has_blocking_change_requests=True,
|
||||
reviews=[_blocking_review()],
|
||||
)
|
||||
result = assess_request_changes_approval_proof(feedback)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertIn("override_reason", " ".join(result["reasons"]))
|
||||
|
||||
def test_unchanged_head_with_valid_override_allows_approval(self):
|
||||
blocker = _blocking_review()
|
||||
feedback = _feedback(
|
||||
has_blocking_change_requests=True,
|
||||
reviews=[blocker],
|
||||
)
|
||||
report = (
|
||||
"Blocker text: Tests fail on pinned head; fix test_commit_files_gate.\n"
|
||||
"Override: wrong_validation_environment — CI used stale worktree."
|
||||
)
|
||||
result = assess_request_changes_approval_proof(
|
||||
feedback,
|
||||
override_reason="wrong_validation_environment",
|
||||
override_explanation="CI used stale worktree; local rerun passed.",
|
||||
report_text=report,
|
||||
)
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertEqual(
|
||||
result["blocking_review"]["blocking_reviewer"], "reviewer-a"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -141,6 +141,17 @@ class TestResolveTaskCapability(unittest.TestCase):
|
||||
self.assertTrue(res["allowed_in_current_session"])
|
||||
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api):
|
||||
# #275: lock_issue must be an exact resolver task for claim/lock gates.
|
||||
with patch.dict(os.environ, self._env("author-profile")):
|
||||
res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs")
|
||||
self.assertEqual(res["requested_task"], "lock_issue")
|
||||
self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
|
||||
self.assertTrue(res["allowed_in_current_session"])
|
||||
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
||||
|
||||
def test_resolve_unknown_task_fails_closed(self):
|
||||
with patch.dict(os.environ, self._env("author-profile")):
|
||||
with self.assertRaises(ValueError):
|
||||
|
||||
@@ -206,28 +206,35 @@ class TestRoleSessionRouter(unittest.TestCase):
|
||||
|
||||
class TestCheckMidMerge(unittest.TestCase):
|
||||
def test_skip_scan_walk_root_skips_sibling_worktrees_only(self):
|
||||
worktree_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
main_root = os.path.dirname(os.path.dirname(worktree_root))
|
||||
self.assertFalse(
|
||||
role_session_router.skip_python_scan_walk_root(
|
||||
main_root, main_root
|
||||
# Hermetic <main>/branches/<worktree> layout (#283): deriving these
|
||||
# paths from __file__ made the test pass only when the suite ran from
|
||||
# a branches/ worktree, because skip_python_scan_walk_root skips a
|
||||
# branches/ walk root only when <project_root>/branches exists.
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
main_root = os.path.join(tmp, "main")
|
||||
worktree_root = os.path.join(main_root, "branches", "fix-issue-1")
|
||||
os.makedirs(os.path.join(worktree_root, "tests"))
|
||||
|
||||
self.assertFalse(
|
||||
role_session_router.skip_python_scan_walk_root(
|
||||
main_root, main_root
|
||||
)
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
role_session_router.skip_python_scan_walk_root(
|
||||
main_root, os.path.join(main_root, "branches", "fix-issue-1")
|
||||
self.assertTrue(
|
||||
role_session_router.skip_python_scan_walk_root(
|
||||
main_root, os.path.join(main_root, "branches", "fix-issue-1")
|
||||
)
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
role_session_router.skip_python_scan_walk_root(
|
||||
worktree_root, worktree_root
|
||||
self.assertFalse(
|
||||
role_session_router.skip_python_scan_walk_root(
|
||||
worktree_root, worktree_root
|
||||
)
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
role_session_router.skip_python_scan_walk_root(
|
||||
worktree_root, os.path.join(worktree_root, "tests")
|
||||
self.assertFalse(
|
||||
role_session_router.skip_python_scan_walk_root(
|
||||
worktree_root, os.path.join(worktree_root, "tests")
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
def test_decorative_equals_banner_is_not_mid_merge(self):
|
||||
self.assertFalse(role_session_router.check_mid_merge())
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
"""Doc-contract checks for the shell-spawn hard-stop rule (#258).
|
||||
|
||||
When the shell executor fails to spawn (exit_code: -1 with empty
|
||||
stdout/stderr), agents must probe once, mark shell unavailable, hard-stop
|
||||
after two consecutive spawn failures, and emit a recovery report instead of
|
||||
retry-spiralling. These tests pin that guidance in the runbooks, the portable
|
||||
skill doc, and the operator guide so it cannot silently regress.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
|
||||
SKILL = REPO_ROOT / "skills" / "llm-project-workflow" / "SKILL.md"
|
||||
|
||||
|
||||
def _normalize(text: str) -> str:
|
||||
"""Collapse whitespace so phrase checks survive markdown line wrapping."""
|
||||
return " ".join(text.split())
|
||||
|
||||
|
||||
def _runbook_text() -> str:
|
||||
return _normalize(RUNBOOK.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _skill_text() -> str:
|
||||
return _normalize(SKILL.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def test_runbook_has_shell_spawn_hard_stop_section():
|
||||
text = _runbook_text()
|
||||
assert "## Shell Spawn Hard-Stop Rule" in text
|
||||
for phrase in (
|
||||
"exit_code: -1",
|
||||
"empty stdout/stderr",
|
||||
"trivial probe",
|
||||
"two consecutive spawn failures",
|
||||
"mark shell unavailable",
|
||||
"recovery report",
|
||||
):
|
||||
assert phrase in text, f"runbook missing phrase: {phrase!r}"
|
||||
|
||||
|
||||
def test_runbook_recovery_report_names_required_steps():
|
||||
text = _runbook_text()
|
||||
for phrase in (
|
||||
"restart the session",
|
||||
"kill hung background terminals",
|
||||
"MCP-native",
|
||||
):
|
||||
assert phrase in text, f"runbook recovery guidance missing: {phrase!r}"
|
||||
|
||||
|
||||
def test_runbook_forbids_retry_spirals():
|
||||
text = _runbook_text()
|
||||
assert "never retry the same failing spawn" in text
|
||||
|
||||
|
||||
def test_skill_doc_declares_shell_spawn_hard_stop_rule():
|
||||
text = _skill_text()
|
||||
assert "## Shell Spawn Hard-Stop Rule" in text
|
||||
for phrase in (
|
||||
"exit_code: -1",
|
||||
"two consecutive spawn failures",
|
||||
"recovery report",
|
||||
):
|
||||
assert phrase in text, f"SKILL.md missing phrase: {phrase!r}"
|
||||
|
||||
|
||||
def test_operator_guide_rules_include_shell_spawn_hard_stop():
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
import gitea_mcp_server
|
||||
|
||||
rule = gitea_mcp_server._GUIDE_RULES["shell_spawn_hard_stop"]
|
||||
for phrase in (
|
||||
"exit_code: -1",
|
||||
"two consecutive",
|
||||
"recovery report",
|
||||
):
|
||||
assert phrase in rule, f"operator guide rule missing: {phrase!r}"
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Tests for fail-closed subagent delegation gates (#266).
|
||||
|
||||
Covers the acceptance criteria: blocked write delegation, allowed read-only
|
||||
delegation, missing inherited context, and invalid subagent final reports.
|
||||
"""
|
||||
import unittest
|
||||
|
||||
import subagent_gate
|
||||
|
||||
|
||||
def _full_context():
|
||||
return {
|
||||
"issue_lock": "issue #266 locked to feat/issue-266-subagent-gate-inheritance",
|
||||
"branch": "feat/issue-266-subagent-gate-inheritance",
|
||||
"worktree_path": "branches/feat-issue-266-subagent-gate-inheritance",
|
||||
"identity_profile": "jcwalker3/prgs-author",
|
||||
"allowed_tool_class": "read_write_files",
|
||||
"command_deny_list": "git push --force; rm -rf",
|
||||
"validation_ledger_requirement": "record command/exit/output for every validation claim",
|
||||
"final_report_schema": "controller-handoff-v1",
|
||||
}
|
||||
|
||||
|
||||
def _full_report():
|
||||
return {
|
||||
"identity_profile": "jcwalker3/prgs-author",
|
||||
"worktree_path": "branches/feat-issue-266-subagent-gate-inheritance",
|
||||
"branch": "feat/issue-266-subagent-gate-inheritance",
|
||||
"changed_files": "subagent_gate.py, tests/test_subagent_gate.py",
|
||||
"validation_results": "pytest tests/test_subagent_gate.py -q: exit 0, 14 passed",
|
||||
"workspace_mutations": "edited subagent_gate.py",
|
||||
}
|
||||
|
||||
|
||||
class TestAssessSubagentDelegation(unittest.TestCase):
|
||||
# AC1: deterministic write tasks are blocked by default
|
||||
def test_write_delegation_blocked_by_default(self):
|
||||
for task in ("claim_issue", "create_branch", "edit_code", "commit",
|
||||
"push_branch", "create_pr", "review_pr", "merge_pr",
|
||||
"cleanup_branch"):
|
||||
res = subagent_gate.assess_subagent_delegation(task)
|
||||
self.assertTrue(res["block"], task)
|
||||
self.assertFalse(res["allowed"], task)
|
||||
self.assertEqual(res["task_class"], "deterministic_write", task)
|
||||
self.assertTrue(
|
||||
any("explicitly allowed" in r for r in res["reasons"]), task)
|
||||
|
||||
# AC2: explicit allowance without a recorded justification still blocks
|
||||
def test_write_delegation_requires_recorded_justification(self):
|
||||
res = subagent_gate.assess_subagent_delegation(
|
||||
"create_pr", explicitly_allowed=True,
|
||||
inherited_context=_full_context())
|
||||
self.assertTrue(res["block"])
|
||||
self.assertTrue(
|
||||
any("justification" in r for r in res["reasons"]))
|
||||
|
||||
# AC3: explicit allowance + justification but missing inherited context
|
||||
def test_write_delegation_blocks_on_missing_inherited_context(self):
|
||||
context = _full_context()
|
||||
del context["issue_lock"]
|
||||
del context["command_deny_list"]
|
||||
res = subagent_gate.assess_subagent_delegation(
|
||||
"commit", explicitly_allowed=True,
|
||||
justification="parent session proved batch commit needs isolation",
|
||||
inherited_context=context)
|
||||
self.assertTrue(res["block"])
|
||||
self.assertIn("issue_lock", res["missing_context"])
|
||||
self.assertIn("command_deny_list", res["missing_context"])
|
||||
|
||||
def test_write_delegation_blocks_on_empty_context_values(self):
|
||||
context = _full_context()
|
||||
context["worktree_path"] = " "
|
||||
res = subagent_gate.assess_subagent_delegation(
|
||||
"commit", explicitly_allowed=True,
|
||||
justification="isolation required",
|
||||
inherited_context=context)
|
||||
self.assertTrue(res["block"])
|
||||
self.assertIn("worktree_path", res["missing_context"])
|
||||
|
||||
def test_write_delegation_allowed_with_authorization_and_full_context(self):
|
||||
res = subagent_gate.assess_subagent_delegation(
|
||||
"edit_code", explicitly_allowed=True,
|
||||
justification="parallel mechanical rename across many files",
|
||||
inherited_context=_full_context())
|
||||
self.assertFalse(res["block"])
|
||||
self.assertTrue(res["allowed"])
|
||||
self.assertEqual(res["missing_context"], [])
|
||||
|
||||
# Allowed read-only delegation needs no explicit authorization
|
||||
def test_read_only_delegation_allowed(self):
|
||||
for task in ("read_files", "code_search", "inventory_prs",
|
||||
"summarize_issue"):
|
||||
res = subagent_gate.assess_subagent_delegation(task)
|
||||
self.assertFalse(res["block"], task)
|
||||
self.assertTrue(res["allowed"], task)
|
||||
self.assertEqual(res["task_class"], "read_only", task)
|
||||
|
||||
# Unknown task types fail closed
|
||||
def test_unknown_task_fails_closed(self):
|
||||
res = subagent_gate.assess_subagent_delegation("launch_missiles")
|
||||
self.assertTrue(res["block"])
|
||||
self.assertEqual(res["task_class"], "unknown")
|
||||
|
||||
def test_blank_task_fails_closed(self):
|
||||
res = subagent_gate.assess_subagent_delegation(" ")
|
||||
self.assertTrue(res["block"])
|
||||
self.assertEqual(res["task_class"], "unknown")
|
||||
|
||||
|
||||
class TestValidateSubagentReport(unittest.TestCase):
|
||||
# AC4: subagent output must carry the same proof fields as the parent
|
||||
def test_full_report_valid(self):
|
||||
res = subagent_gate.validate_subagent_report(_full_report())
|
||||
self.assertTrue(res["valid"])
|
||||
self.assertFalse(res["block"])
|
||||
self.assertEqual(res["missing_fields"], [])
|
||||
|
||||
def test_missing_proof_fields_invalid(self):
|
||||
report = _full_report()
|
||||
del report["validation_results"]
|
||||
del report["worktree_path"]
|
||||
res = subagent_gate.validate_subagent_report(report)
|
||||
self.assertFalse(res["valid"])
|
||||
self.assertTrue(res["block"])
|
||||
self.assertIn("validation_results", res["missing_fields"])
|
||||
self.assertIn("worktree_path", res["missing_fields"])
|
||||
|
||||
def test_empty_field_values_invalid(self):
|
||||
report = _full_report()
|
||||
report["changed_files"] = ""
|
||||
res = subagent_gate.validate_subagent_report(report)
|
||||
self.assertFalse(res["valid"])
|
||||
self.assertIn("changed_files", res["missing_fields"])
|
||||
|
||||
def test_none_report_invalid(self):
|
||||
res = subagent_gate.validate_subagent_report(None)
|
||||
self.assertFalse(res["valid"])
|
||||
self.assertTrue(res["block"])
|
||||
|
||||
|
||||
class TestOperatorGuideRule(unittest.TestCase):
|
||||
def test_operator_guide_declares_subagent_rule(self):
|
||||
import gitea_mcp_server
|
||||
|
||||
rule = gitea_mcp_server._GUIDE_RULES["subagent_delegation"].lower()
|
||||
for phrase in ("deterministic write", "inherit", "read-only",
|
||||
"fail closed"):
|
||||
self.assertIn(phrase, rule)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,155 @@
|
||||
"""Tests for validation ledger and final-report verifier (#271)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from validation_ledger import ( # noqa: E402
|
||||
assess_full_suite_claim,
|
||||
assess_ledger_entry,
|
||||
new_ledger_entry,
|
||||
verify_final_report,
|
||||
)
|
||||
|
||||
|
||||
class TestLedgerEntry(unittest.TestCase):
|
||||
def test_valid_entry_is_claimable(self):
|
||||
entry = new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/feat-issue-271",
|
||||
exit_code=0,
|
||||
output_summary="994 passed, 6 skipped",
|
||||
)
|
||||
result = assess_ledger_entry(entry)
|
||||
self.assertTrue(result["claimable"])
|
||||
self.assertEqual(result["verdict"], "strong")
|
||||
|
||||
def test_missing_command_output_is_invalid(self):
|
||||
entry = new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/task",
|
||||
exit_code=0,
|
||||
output_summary="",
|
||||
)
|
||||
result = assess_ledger_entry(entry)
|
||||
self.assertFalse(result["claimable"])
|
||||
self.assertEqual(result["verdict"], "invalid")
|
||||
|
||||
|
||||
class TestFullSuiteClaim(unittest.TestCase):
|
||||
def _entry(self, **kwargs):
|
||||
base = new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/task",
|
||||
exit_code=0,
|
||||
output_summary="10 passed",
|
||||
)
|
||||
base.update(kwargs)
|
||||
return base
|
||||
|
||||
def test_full_suite_overclaim_with_ignored_paths_fails(self):
|
||||
entry = self._entry(
|
||||
ignored_paths=[{"path": "tests/integration/", "justification": ""}],
|
||||
)
|
||||
result = assess_full_suite_claim(
|
||||
[entry],
|
||||
report_text="Validation: full suite passed.",
|
||||
)
|
||||
self.assertFalse(result["claimable"])
|
||||
self.assertTrue(
|
||||
any("full suite" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_full_suite_except_note_allows_ignored_paths(self):
|
||||
entry = self._entry(
|
||||
ignored_paths=[{
|
||||
"path": "tests/integration/",
|
||||
"justification": "network tests require GITEA_INTEGRATION=1",
|
||||
}],
|
||||
)
|
||||
result = assess_full_suite_claim(
|
||||
[entry],
|
||||
report_text="Validation: full suite except integration tests.",
|
||||
)
|
||||
self.assertTrue(result["claimable"])
|
||||
|
||||
|
||||
class TestFinalReportVerifier(unittest.TestCase):
|
||||
def _base_report(self, **overrides):
|
||||
report = {
|
||||
"identity": "jcwalker3",
|
||||
"profile": "prgs-author",
|
||||
"capability_resolved": True,
|
||||
"worktree_path": "/repo/branches/feat-issue-271",
|
||||
"changed_files": ["validation_ledger.py"],
|
||||
"validation_ledger": [
|
||||
new_ledger_entry(
|
||||
command="pytest tests/test_validation_ledger.py -q",
|
||||
cwd="/repo/branches/feat-issue-271",
|
||||
exit_code=0,
|
||||
output_summary="12 passed",
|
||||
)
|
||||
],
|
||||
"validation_claim": "passed",
|
||||
"review_submitted": False,
|
||||
"review_visible_approved": False,
|
||||
"review_pending": False,
|
||||
"merge_performed": False,
|
||||
"merge_blocker": "not a reviewer task",
|
||||
"workspace_clean": True,
|
||||
}
|
||||
report.update(overrides)
|
||||
return report
|
||||
|
||||
def test_complete_report_is_proven(self):
|
||||
result = verify_final_report(self._base_report())
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_validation_overclaim_fails(self):
|
||||
report = self._base_report(
|
||||
validation_claim="passed",
|
||||
validation_ledger=[
|
||||
new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/task",
|
||||
exit_code=1,
|
||||
output_summary="3 failed",
|
||||
)
|
||||
],
|
||||
)
|
||||
result = verify_final_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["contradictions"])
|
||||
|
||||
def test_pending_approval_misreported_as_approved_fails(self):
|
||||
report = self._base_report(
|
||||
claims_approved_review=True,
|
||||
review_visible_approved=False,
|
||||
review_pending=True,
|
||||
)
|
||||
result = verify_final_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("approved" in c.lower() for c in result["contradictions"])
|
||||
)
|
||||
|
||||
def test_missing_command_output_in_ledger_fails(self):
|
||||
report = self._base_report(
|
||||
validation_ledger=[
|
||||
new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/task",
|
||||
exit_code=0,
|
||||
output_summary="",
|
||||
)
|
||||
],
|
||||
)
|
||||
result = verify_final_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["reasons"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Documentation checks for MCP-native commit path rules (#260)."""
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
|
||||
SAFETY = REPO_ROOT / "docs" / "safety-model.md"
|
||||
|
||||
|
||||
def test_runbook_requires_mcp_native_commit_path():
|
||||
text = RUNBOOK.read_text(encoding="utf-8")
|
||||
assert "MCP-native commit path" in text
|
||||
assert "gitea_commit_files" in text
|
||||
assert "gitea.repo.commit" in text
|
||||
assert "only" in text.lower() and "approved" in text.lower()
|
||||
lower = text.lower()
|
||||
for forbidden in ("webfetch", "playwright", "manual llm-generated base64"):
|
||||
assert forbidden in lower
|
||||
|
||||
|
||||
def test_runbook_forbids_fallback_loops():
|
||||
lower = RUNBOOK.read_text(encoding="utf-8").lower()
|
||||
assert "retry shell encoding" in lower
|
||||
assert "loop" in lower
|
||||
assert "recovery report" in lower
|
||||
|
||||
|
||||
def test_safety_model_documents_commit_fallback_ban():
|
||||
text = SAFETY.read_text(encoding="utf-8")
|
||||
assert "Agent Commit Path" in text
|
||||
assert "gitea_commit_files" in text
|
||||
assert "WebFetch" in text
|
||||
assert "Playwright" in text
|
||||
assert "llm-workflow-runbooks.md" in text
|
||||
Reference in New Issue
Block a user