Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity
Files
ea8e186549d4a21e7e3e8f41261ff527b6a60e97
Gitea-Tools/tests/test_anti_stomp_preflight.py
T
sysadmin ea8e186549 feat: common anti-stomp preflight before mutation tools (Closes #604) 
Add a shared fail-closed anti-stomp preflight that mutation tools invoke
before create/comment/lease/review/approve/request-changes/merge/cleanup/
label mutations. Composes repo/role/root/worktree/lease/terminal-lock/
head-SHA/stale-runtime/workflow-hash/contamination checks into a typed
blocker with an exact next action. No agent bypass flags.
2026-07-12 05:10:30 -04:00

567 lines
20 KiB
Python
Raw Blame History

"""Regression coverage for the common anti-stomp preflight (#604).
Acceptance criteria:
1. All mutation tools call the common preflight (wired via
``verify_preflight_purity`` / ``_run_anti_stomp_preflight``).
2. Failure returns a typed blocker and exact next action.
3. Tests cover wrong repo defaulting to Timesheet, stale runtime, wrong
worktree, foreign lease, terminal lock, and contaminated approval.
4. No mutation can proceed using stale prompt data if live state disagrees.
5. Existing successful paths continue to work (happy-path assessment).
"""
from __future__ import annotations
import os
import unittest
from unittest import mock
import anti_stomp_preflight as asp
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
STARTUP = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
ADVANCED = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
HEAD_A = "1111111111111111111111111111111111111111"
HEAD_B = "2222222222222222222222222222222222222222"
def _happy_kwargs(**overrides):
base = dict(
task="create_issue",
remote="prgs",
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Gitea-Tools",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=True,
repo_explicit=True,
profile_name="prgs-author",
profile_role="author",
required_role="author",
workspace_path="/repo/branches/issue-604",
project_root="/repo",
current_branch="feat/issue-604-anti-stomp-preflight",
root_head_sha=STARTUP,
root_porcelain="",
remote_master_sha=STARTUP,
startup_head=STARTUP,
current_code_head=STARTUP,
source_contaminated=False,
manual_bypass_attempted=False,
)
base.update(overrides)
return base
class TestIsMutationTask(unittest.TestCase):
def test_core_mutation_tasks(self):
for task in (
"create_issue",
"comment_issue",
"set_issue_labels",
"acquire_reviewer_pr_lease",
"submit_pr_review",
"approve_pr",
"request_changes_pr",
"merge_pr",
"cleanup_stale_claims",
"delete_branch",
):
self.assertTrue(asp.is_mutation_task(task), task)
def test_gitea_prefix_accepted(self):
self.assertTrue(asp.is_mutation_task("gitea_create_issue"))
self.assertTrue(asp.is_mutation_task("gitea_merge_pr"))
def test_read_only_not_mutation(self):
self.assertFalse(asp.is_mutation_task("list_prs"))
self.assertFalse(asp.is_mutation_task("whoami"))
self.assertFalse(asp.is_mutation_task(""))
self.assertFalse(asp.is_mutation_task(None))
class TestHappyPath(unittest.TestCase):
def test_allowed_when_all_checks_pass(self):
result = asp.assess_anti_stomp_preflight(**_happy_kwargs())
self.assertTrue(result["allowed"])
self.assertFalse(result["block"])
self.assertEqual(result["blockers"], [])
self.assertEqual(result["exact_next_action"], "proceed")
self.assertIsNone(result["blocker_kind"])
def test_reviewer_happy_path_skips_author_worktree(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="review_pr",
profile_name="prgs-reviewer",
profile_role="reviewer",
required_role="reviewer",
# Under branches/ the root guard short-circuits for non-merger.
workspace_path="/repo/branches/review-pr-42",
project_root="/repo",
current_branch="review/pr-42",
foreign_lease=False,
terminal_lock_blocks=False,
expected_head_sha=HEAD_A,
live_head_sha=HEAD_A,
workflow_hash_valid=True,
)
)
self.assertTrue(result["allowed"], result.get("reasons"))
self.assertTrue(result["checks"]["worktree"].get("skipped"))
class TestWrongRepoTimesheet(unittest.TestCase):
"""AC3: wrong repo defaulting to Timesheet."""
def test_prgs_default_timesheet_vs_local_gitea_tools(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Timesheet",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=False,
repo_explicit=False,
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_REPO)
self.assertTrue(result["exact_next_action"])
self.assertIn("org=", result["exact_next_action"])
self.assertTrue(
any("Timesheet" in r or "does not match" in r for r in result["reasons"])
)
def test_explicit_org_repo_skips_mismatch(self):
# Explicit intent is authoritative (remote_repo_guard contract).
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
resolved_repo="Timesheet",
org_explicit=True,
repo_explicit=True,
)
)
self.assertTrue(result["allowed"])
class TestStaleRuntime(unittest.TestCase):
"""AC3: stale runtime."""
def test_stale_runtime_blocks(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
startup_head=STARTUP,
current_code_head=ADVANCED,
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_STALE_RUNTIME)
self.assertIn("Restart", result["exact_next_action"])
self.assertTrue(result["checks"]["stale_runtime"]["stale"])
def test_in_parity_allows(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
startup_head=STARTUP,
current_code_head=STARTUP,
)
)
self.assertTrue(result["allowed"])
class TestWrongWorktree(unittest.TestCase):
"""AC3: wrong worktree."""
def test_author_on_control_checkout_blocked(self):
# Clean master control checkout: root guard may pass for a clean master
# HEAD, but the author worktree guard must still refuse mutation from
# outside branches/.
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
workspace_path="/repo",
project_root="/repo",
current_branch="master",
root_porcelain="",
root_head_sha=STARTUP,
remote_master_sha=STARTUP,
profile_role="author",
required_role="author",
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_WORKTREE)
self.assertIn("branches/", result["exact_next_action"])
def test_author_under_branches_allowed(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
workspace_path="/repo/branches/issue-604",
project_root="/repo",
)
)
self.assertTrue(result["allowed"])
class TestForeignLease(unittest.TestCase):
"""AC3: foreign lease."""
def test_foreign_lease_blocks(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="merge_pr",
profile_role="merger",
required_role="merger",
# Merger is not auto-exempted under branches/; pass clean master.
workspace_path="/repo",
project_root="/repo",
current_branch="master",
root_porcelain="",
foreign_lease=True,
lease_reasons=["lease owned by other-session"],
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_FOREIGN_LEASE)
self.assertIn("foreign lease", result["exact_next_action"].lower())
def test_lease_required_without_ownership_fails_closed(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="review_pr",
profile_role="reviewer",
required_role="reviewer",
workspace_path="/repo/branches/review-pr-1",
project_root="/repo",
lease_required=True,
lease_owner_session="sess-A",
active_session_id="sess-B",
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_FOREIGN_LEASE)
def test_owned_lease_allows(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="review_pr",
profile_role="reviewer",
required_role="reviewer",
workspace_path="/repo/branches/review-pr-1",
project_root="/repo",
foreign_lease=False,
)
)
self.assertTrue(result["allowed"])
class TestTerminalLock(unittest.TestCase):
"""AC3: terminal lock."""
def test_terminal_lock_blocks(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="approve_pr",
profile_role="reviewer",
required_role="reviewer",
workspace_path="/repo/branches/review-pr-9",
project_root="/repo",
terminal_lock_blocks=True,
terminal_lock_reasons=["#332 terminal lock active for this head"],
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_TERMINAL_LOCK)
self.assertIn("#332", result["exact_next_action"])
def test_no_terminal_lock_allows(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="approve_pr",
profile_role="reviewer",
required_role="reviewer",
workspace_path="/repo/branches/review-pr-9",
project_root="/repo",
terminal_lock_blocks=False,
)
)
self.assertTrue(result["allowed"])
class TestHeadShaStalePrompt(unittest.TestCase):
"""AC4: no mutation with stale prompt head SHA."""
def _merger_kwargs(self, **overrides):
base = _happy_kwargs(
task="merge_pr",
profile_role="merger",
required_role="merger",
workspace_path="/repo",
project_root="/repo",
current_branch="master",
root_porcelain="",
root_head_sha=STARTUP,
remote_master_sha=STARTUP,
)
base.update(overrides)
return base
def test_head_mismatch_blocks(self):
result = asp.assess_anti_stomp_preflight(
**self._merger_kwargs(
expected_head_sha=HEAD_A,
live_head_sha=HEAD_B,
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_HEAD_SHA)
self.assertIn("stale prompt", " ".join(result["reasons"]).lower())
def test_require_head_sha_missing_blocks(self):
result = asp.assess_anti_stomp_preflight(
**self._merger_kwargs(
require_head_sha=True,
expected_head_sha=None,
live_head_sha=HEAD_A,
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_HEAD_SHA)
def test_matching_head_allows(self):
result = asp.assess_anti_stomp_preflight(
**self._merger_kwargs(
expected_head_sha=HEAD_A,
live_head_sha=HEAD_A,
)
)
self.assertTrue(result["allowed"])
class TestContaminatedApproval(unittest.TestCase):
"""AC3: contaminated approval / source contamination."""
def test_source_contamination_blocks(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="merge_pr",
profile_role="merger",
required_role="merger",
workspace_path="/repo",
project_root="/repo",
current_branch="master",
root_porcelain="",
source_contaminated=True,
contamination_reasons=[
"session contaminated by direct stable-branch push attempt"
],
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_SOURCE_CONTAMINATION)
self.assertIn("reconciler", result["exact_next_action"].lower())
def test_manual_bypass_blocks(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
manual_bypass_attempted=True,
manual_bypass_reasons=[
"attempted manual deletion of session-state lock files"
],
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_MANUAL_BYPASS)
class TestWrongRole(unittest.TestCase):
def test_author_cannot_merge(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="merge_pr",
profile_role="author",
required_role="merger",
workspace_path="/repo/branches/issue-604",
project_root="/repo",
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE)
def test_reconciler_may_run_author_class_tasks(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="comment_issue",
profile_name="prgs-reconciler",
profile_role="reconciler",
required_role="author",
workspace_path="/repo",
project_root="/repo",
current_branch="master",
root_porcelain="",
)
)
self.assertTrue(result["allowed"], result.get("reasons"))
self.assertTrue(asp.roles_compatible("reconciler", "author"))
self.assertFalse(asp.roles_compatible("author", "merger"))
class TestWorkflowHash(unittest.TestCase):
def test_stale_workflow_hash_blocks(self):
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="review_pr",
profile_role="reviewer",
required_role="reviewer",
workspace_path="/repo/branches/review-pr-3",
project_root="/repo",
workflow_hash_valid=False,
workflow_hash_reasons=["stored workflow hash is stale"],
)
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], asp.BLOCKER_WORKFLOW_HASH)
class TestTypedBlockerResponse(unittest.TestCase):
"""AC2: typed blocker + exact next action in response payload."""
def test_block_response_shape(self):
assessment = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
task="review_pr",
profile_role="reviewer",
required_role="reviewer",
workspace_path="/repo/branches/review-pr-7",
project_root="/repo",
foreign_lease=True,
)
)
payload = asp.block_response(assessment, pr_number=42)
self.assertFalse(payload["success"])
self.assertFalse(payload["performed"])
self.assertTrue(payload["blocked"])
self.assertTrue(payload["anti_stomp"])
self.assertEqual(payload["blocker_kind"], asp.BLOCKER_FOREIGN_LEASE)
self.assertTrue(payload["exact_next_action"])
self.assertEqual(payload["pr_number"], 42)
self.assertTrue(payload["blockers"])
self.assertEqual(payload["blockers"][0]["kind"], asp.BLOCKER_FOREIGN_LEASE)
def test_format_error_includes_kind_and_next_action(self):
assessment = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
startup_head=STARTUP,
current_code_head=ADVANCED,
)
)
msg = asp.format_anti_stomp_error(assessment)
self.assertIn("#604", msg)
self.assertIn(asp.BLOCKER_STALE_RUNTIME, msg)
self.assertIn("exact_next_action", msg)
class TestRootCheckoutContamination(unittest.TestCase):
def test_dirty_control_checkout_blocks_author(self):
# Workspace is control checkout with dirty porcelain.
result = asp.assess_anti_stomp_preflight(
**_happy_kwargs(
workspace_path="/repo",
project_root="/repo",
current_branch="master",
root_porcelain=" M gitea_mcp_server.py\n",
root_head_sha=STARTUP,
remote_master_sha=STARTUP,
)
)
self.assertTrue(result["block"])
# Author worktree check or root checkout may fire first.
self.assertIn(
result["blocker_kind"],
{asp.BLOCKER_ROOT_CHECKOUT, asp.BLOCKER_WRONG_WORKTREE},
)
class TestMutationTaskInventory(unittest.TestCase):
"""AC1: mutation task set covers issue-listed paths."""
def test_issue_required_paths_covered(self):
required = {
"create_issue",
"comment_issue",
"set_issue_labels",
"acquire_reviewer_pr_lease",
"submit_pr_review",
"approve_pr",
"request_changes_pr",
"merge_pr",
"cleanup_merged_pr_branch",
"cleanup_stale_claims",
}
missing = required - asp.MUTATION_TASKS
self.assertFalse(missing, f"missing mutation tasks: {missing}")
class TestServerWiring(unittest.TestCase):
"""Smoke: MCP server imports anti_stomp and exposes the runner."""
def test_server_imports_anti_stomp(self):
import gitea_mcp_server as server
self.assertTrue(hasattr(server, "_run_anti_stomp_preflight"))
self.assertTrue(hasattr(server, "anti_stomp_preflight"))
self.assertIs(server.anti_stomp_preflight, asp)
def test_runner_skipped_under_pytest_by_default(self):
import gitea_mcp_server as server
# Default pytest path must not raise (suite isolation).
self.assertIsNone(
server._run_anti_stomp_preflight(
"create_issue",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Timesheet",
)
)
def test_runner_blocks_when_forced(self):
import gitea_mcp_server as server
with mock.patch.dict(
os.environ,
{"GITEA_TEST_FORCE_ANTI_STOMP": "1"},
clear=False,
):
# Force assessor to return a block without full git/workspace setup.
blocked = {
"allowed": False,
"block": True,
"blockers": [{
"kind": asp.BLOCKER_STALE_RUNTIME,
"reasons": ["stale"],
"exact_next_action": "restart",
"detail": {},
}],
"reasons": ["stale"],
"exact_next_action": "restart",
"blocker_kind": asp.BLOCKER_STALE_RUNTIME,
"checks": {},
}
with mock.patch.object(
asp, "assess_anti_stomp_preflight", return_value=blocked
):
with self.assertRaises(RuntimeError) as ctx:
server._run_anti_stomp_preflight(
"create_issue",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertIn("#604", str(ctx.exception))
self.assertIn(asp.BLOCKER_STALE_RUNTIME, str(ctx.exception))
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink