Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eb97060867 |
@@ -32,7 +32,6 @@ Wiki-related issues cannot be closed until the live Wiki is verified — see
|
|||||||
| Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
|
| Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
|
||||||
|---|---|---|---|
|
|---|---|---|---|
|
||||||
| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
|
| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
|
||||||
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home); 10 pages (History, Home, Identity-and-Profiles, MCP-Tools, Open-Decisions, Operator-Guide, Repositories, Runbooks, Safety-and-Gates, Workflow); wiki git log head `ef3dec2` |
|
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home) |
|
||||||
|
|
||||||
|
|
||||||
Update this table whenever a wiki is published, re-synced, or found stale.
|
Update this table whenever a wiki is published, re-synced, or found stale.
|
||||||
+1
-9
@@ -205,7 +205,7 @@ def selected_profile_name():
|
|||||||
|
|
||||||
|
|
||||||
def is_runtime_switching_enabled(path=None):
|
def is_runtime_switching_enabled(path=None):
|
||||||
"""Check if runtime profile switching is enabled in config."""
|
"""Check if runtime profile switching is explicitly enabled in config."""
|
||||||
try:
|
try:
|
||||||
config = load_config(path)
|
config = load_config(path)
|
||||||
except Exception:
|
except Exception:
|
||||||
@@ -213,18 +213,10 @@ def is_runtime_switching_enabled(path=None):
|
|||||||
if not config:
|
if not config:
|
||||||
return False
|
return False
|
||||||
rules = config.get("rules") or {}
|
rules = config.get("rules") or {}
|
||||||
if rules.get("allow_runtime_switching") is False:
|
|
||||||
return False
|
|
||||||
if config.get("allow_runtime_switching") is False:
|
|
||||||
return False
|
|
||||||
if rules.get("allow_runtime_switching") is True:
|
if rules.get("allow_runtime_switching") is True:
|
||||||
return True
|
return True
|
||||||
if config.get("allow_runtime_switching") is True:
|
if config.get("allow_runtime_switching") is True:
|
||||||
return True
|
return True
|
||||||
# Default to True if multiple profiles exist in the config
|
|
||||||
profiles = config.get("profiles") or {}
|
|
||||||
if len(profiles) > 1:
|
|
||||||
return True
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+4
-154
@@ -259,7 +259,6 @@ import issue_duplicate_gate # noqa: E402
|
|||||||
import role_session_router # noqa: E402
|
import role_session_router # noqa: E402
|
||||||
import role_namespace_gate # noqa: E402
|
import role_namespace_gate # noqa: E402
|
||||||
import task_capability_map # noqa: E402
|
import task_capability_map # noqa: E402
|
||||||
import review_proofs # noqa: E402
|
|
||||||
|
|
||||||
|
|
||||||
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
|
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
|
||||||
@@ -2275,23 +2274,6 @@ def gitea_merge_pr(
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
def _local_git_remote_url(remote_name: str) -> str | None:
|
|
||||||
"""Best-effort local ``git remote get-url`` for trust-gate corroboration."""
|
|
||||||
try:
|
|
||||||
proc = subprocess.run(
|
|
||||||
["git", "remote", "get-url", remote_name],
|
|
||||||
capture_output=True,
|
|
||||||
text=True,
|
|
||||||
cwd=PROJECT_ROOT,
|
|
||||||
)
|
|
||||||
if proc.returncode != 0:
|
|
||||||
return None
|
|
||||||
url = (proc.stdout or "").strip()
|
|
||||||
return url or None
|
|
||||||
except Exception:
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def gitea_review_pr(
|
def gitea_review_pr(
|
||||||
pr_number: int,
|
pr_number: int,
|
||||||
@@ -2359,8 +2341,6 @@ def gitea_review_pr(
|
|||||||
prs_found_count = 0
|
prs_found_count = 0
|
||||||
pr_details_list = []
|
pr_details_list = []
|
||||||
inventory_msg = ""
|
inventory_msg = ""
|
||||||
inventory_trust_gate = None
|
|
||||||
prs: list = []
|
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = None
|
auth = None
|
||||||
@@ -2445,28 +2425,7 @@ def gitea_review_pr(
|
|||||||
if inventory_msg:
|
if inventory_msg:
|
||||||
report_lines.append(inventory_msg)
|
report_lines.append(inventory_msg)
|
||||||
else:
|
else:
|
||||||
inventory_trust_gate = review_proofs.pr_inventory_trust_gate(
|
report_lines.append("Open PRs found: 0")
|
||||||
prs if inventory_attempted else None,
|
|
||||||
remote=remote,
|
|
||||||
org=o,
|
|
||||||
repo=r,
|
|
||||||
state="open",
|
|
||||||
authenticated_profile=profile,
|
|
||||||
local_remote_url=_local_git_remote_url(remote),
|
|
||||||
has_finality_metadata=True,
|
|
||||||
)
|
|
||||||
report_lines.extend(
|
|
||||||
review_proofs.format_pr_inventory_trust_gate_report(
|
|
||||||
inventory_trust_gate
|
|
||||||
)
|
|
||||||
)
|
|
||||||
if inventory_trust_gate.get("status") == "trusted_empty":
|
|
||||||
report_lines.append("Open PRs found: 0 (trusted_empty)")
|
|
||||||
else:
|
|
||||||
report_lines.append(
|
|
||||||
"Empty-queue claim blocked: "
|
|
||||||
"pr_inventory_trust_gate did not return trusted_empty"
|
|
||||||
)
|
|
||||||
else:
|
else:
|
||||||
report_lines.append(inventory_msg)
|
report_lines.append(inventory_msg)
|
||||||
|
|
||||||
@@ -2758,76 +2717,6 @@ def _permission_block_report(required_operation: str,
|
|||||||
return report
|
return report
|
||||||
|
|
||||||
|
|
||||||
def _role_for_operation(op: str) -> str | None:
|
|
||||||
# Normalize op first
|
|
||||||
try:
|
|
||||||
op_n = gitea_config.normalize_operation(op)
|
|
||||||
except Exception:
|
|
||||||
op_n = op
|
|
||||||
if op_n in ("gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", "gitea.pr.request_changes"):
|
|
||||||
return "reviewer"
|
|
||||||
if op_n in (
|
|
||||||
"gitea.issue.create",
|
|
||||||
"gitea.issue.comment",
|
|
||||||
"gitea.issue.close",
|
|
||||||
"gitea.branch.create",
|
|
||||||
"gitea.branch.push",
|
|
||||||
"gitea.pr.create",
|
|
||||||
"gitea.pr.comment",
|
|
||||||
"gitea.pr.close",
|
|
||||||
"gitea.branch.delete",
|
|
||||||
"gitea.repo.commit"
|
|
||||||
):
|
|
||||||
return "author"
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
|
|
||||||
"""Try to find a profile in config that allows op and has valid credentials.
|
|
||||||
|
|
||||||
If found, switch to it, clear identity cache, and return True.
|
|
||||||
Otherwise return False.
|
|
||||||
"""
|
|
||||||
role = _role_for_operation(op)
|
|
||||||
if not role:
|
|
||||||
return False
|
|
||||||
if not gitea_config.is_runtime_switching_enabled():
|
|
||||||
return False
|
|
||||||
config = gitea_config.load_config()
|
|
||||||
if not config or "profiles" not in config:
|
|
||||||
return False
|
|
||||||
for p_name, p_data in config["profiles"].items():
|
|
||||||
# Role classification matching
|
|
||||||
p_role = p_data.get("role") or _role_kind(p_data.get("allowed_operations", []), p_data.get("forbidden_operations", []))
|
|
||||||
if p_role != role:
|
|
||||||
continue
|
|
||||||
p_allowed = p_data.get("allowed_operations") or []
|
|
||||||
p_forbidden = p_data.get("forbidden_operations") or []
|
|
||||||
p_allowed_n = []
|
|
||||||
for op_val in p_allowed:
|
|
||||||
try:
|
|
||||||
p_allowed_n.append(gitea_config.normalize_operation(op_val))
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
p_forbidden_n = []
|
|
||||||
for op_val in p_forbidden:
|
|
||||||
try:
|
|
||||||
p_forbidden_n.append(gitea_config.normalize_operation(op_val))
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
ok, _ = gitea_config.check_operation(op, p_allowed_n, p_forbidden_n)
|
|
||||||
if ok:
|
|
||||||
try:
|
|
||||||
tok = gitea_config.resolve_token(p_data)
|
|
||||||
if tok:
|
|
||||||
gitea_config._active_profile_override = p_name
|
|
||||||
_IDENTITY_CACHE.clear()
|
|
||||||
return True
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def _profile_operation_gate(op: str) -> list[str]:
|
def _profile_operation_gate(op: str) -> list[str]:
|
||||||
"""Profile permission check for a single gated operation (#126, #216).
|
"""Profile permission check for a single gated operation (#126, #216).
|
||||||
|
|
||||||
@@ -2845,17 +2734,6 @@ def _profile_operation_gate(op: str) -> list[str]:
|
|||||||
op, profile["allowed_operations"], profile["forbidden_operations"])
|
op, profile["allowed_operations"], profile["forbidden_operations"])
|
||||||
if op_ok:
|
if op_ok:
|
||||||
return []
|
return []
|
||||||
|
|
||||||
if _try_auto_switch_for_operation(op):
|
|
||||||
try:
|
|
||||||
profile = get_profile()
|
|
||||||
op_ok, op_reason = gitea_config.check_operation(
|
|
||||||
op, profile["allowed_operations"], profile["forbidden_operations"])
|
|
||||||
if op_ok:
|
|
||||||
return []
|
|
||||||
except Exception as exc:
|
|
||||||
return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
|
|
||||||
|
|
||||||
if op_reason == "no-allowed-operations":
|
if op_reason == "no-allowed-operations":
|
||||||
return ["profile has no configured allowed operations (fail closed)"]
|
return ["profile has no configured allowed operations (fail closed)"]
|
||||||
if op_reason == "forbidden":
|
if op_reason == "forbidden":
|
||||||
@@ -4507,12 +4385,6 @@ def gitea_resolve_task_capability(
|
|||||||
required_permission, active_allowed, active_forbidden
|
required_permission, active_allowed, active_forbidden
|
||||||
)
|
)
|
||||||
|
|
||||||
switching = gitea_config.is_runtime_switching_enabled()
|
|
||||||
available_in_session = allowed_in_current_session
|
|
||||||
configured = False
|
|
||||||
restart_required = False
|
|
||||||
reason_msg = None
|
|
||||||
|
|
||||||
# Find matching configured profiles
|
# Find matching configured profiles
|
||||||
matching_profiles = []
|
matching_profiles = []
|
||||||
if config and "profiles" in config:
|
if config and "profiles" in config:
|
||||||
@@ -4535,24 +4407,7 @@ def gitea_resolve_task_capability(
|
|||||||
if ok:
|
if ok:
|
||||||
matching_profiles.append(p_name)
|
matching_profiles.append(p_name)
|
||||||
|
|
||||||
if not allowed_in_current_session and switching:
|
switching = gitea_config.is_runtime_switching_enabled()
|
||||||
# Try to auto switch
|
|
||||||
if _try_auto_switch_for_operation(required_permission, h):
|
|
||||||
# Successfully switched!
|
|
||||||
profile = get_profile()
|
|
||||||
username = _authenticated_username(h) if h else None
|
|
||||||
active_allowed = profile.get("allowed_operations") or []
|
|
||||||
active_forbidden = profile.get("forbidden_operations") or []
|
|
||||||
allowed_in_current_session = True
|
|
||||||
available_in_session = True
|
|
||||||
else:
|
|
||||||
# Check if any profile in config allows it (meaning it is configured but unattached)
|
|
||||||
if matching_profiles:
|
|
||||||
configured = True
|
|
||||||
restart_required = True
|
|
||||||
available_in_session = False
|
|
||||||
reason_msg = f"{required_role.capitalize()} profile exists but MCP server was added after session startup and is not attached"
|
|
||||||
|
|
||||||
different_namespace_required = False
|
different_namespace_required = False
|
||||||
next_safe_action = "None; ready for operations."
|
next_safe_action = "None; ready for operations."
|
||||||
|
|
||||||
@@ -4630,19 +4485,14 @@ def gitea_resolve_task_capability(
|
|||||||
"active_identity": username,
|
"active_identity": username,
|
||||||
"active_profile_allowed_operations": active_allowed,
|
"active_profile_allowed_operations": active_allowed,
|
||||||
"allowed_in_current_session": allowed_in_current_session,
|
"allowed_in_current_session": allowed_in_current_session,
|
||||||
"available_in_session": available_in_session,
|
"stop_required": stop_required,
|
||||||
"configured": configured,
|
|
||||||
"restart_required": restart_required,
|
|
||||||
"stop_required": stop_required or restart_required,
|
|
||||||
"task_role_guidance": task_role_guidance,
|
"task_role_guidance": task_role_guidance,
|
||||||
"matching_configured_profile": matching_profiles,
|
"matching_configured_profile": matching_profiles,
|
||||||
"runtime_switching_supported": switching,
|
"runtime_switching_supported": switching,
|
||||||
"different_mcp_namespace_required": different_namespace_required,
|
"different_mcp_namespace_required": different_namespace_required,
|
||||||
"exact_safe_next_action": next_safe_action,
|
"exact_safe_next_action": next_safe_action,
|
||||||
}
|
}
|
||||||
if reason_msg:
|
if stop_required:
|
||||||
result["reason"] = reason_msg
|
|
||||||
if stop_required or restart_required:
|
|
||||||
terminal = capability_stop_terminal.enter_from_capability_result(result)
|
terminal = capability_stop_terminal.enter_from_capability_result(result)
|
||||||
if terminal:
|
if terminal:
|
||||||
result["terminal_mode"] = True
|
result["terminal_mode"] = True
|
||||||
|
|||||||
+1
-116
@@ -17,7 +17,6 @@ here weakens or replaces them.
|
|||||||
import re
|
import re
|
||||||
|
|
||||||
import issue_duplicate_gate
|
import issue_duplicate_gate
|
||||||
from reviewer_worktree import assess_reviewer_worktree_proof
|
|
||||||
|
|
||||||
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
|
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
|
||||||
|
|
||||||
@@ -638,7 +637,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
role_boundary=None, review_mutation=None,
|
role_boundary=None, review_mutation=None,
|
||||||
report_text=None, review_decision_lock=None,
|
report_text=None, review_decision_lock=None,
|
||||||
controller_handoff=None, capability_proof=None,
|
controller_handoff=None, capability_proof=None,
|
||||||
sweep_proof=None, worktree_proof=None):
|
sweep_proof=None):
|
||||||
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
|
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
|
||||||
|
|
||||||
Combines the individual proof verdicts into the final-report fields the
|
Combines the individual proof verdicts into the final-report fields the
|
||||||
@@ -708,14 +707,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
"downgraded": True,
|
"downgraded": True,
|
||||||
"reasons": ["review mutation proof not provided (#211)"],
|
"reasons": ["review mutation proof not provided (#211)"],
|
||||||
}
|
}
|
||||||
if worktree_proof is not None:
|
|
||||||
worktree = assess_reviewer_worktree_proof(worktree_proof)
|
|
||||||
else:
|
|
||||||
worktree = {
|
|
||||||
"proven": False,
|
|
||||||
"block": True,
|
|
||||||
"reasons": ["reviewer worktree proof not provided (#233)"],
|
|
||||||
}
|
|
||||||
|
|
||||||
capability_proven = bool(capability_evidence.get("proven"))
|
capability_proven = bool(capability_evidence.get("proven"))
|
||||||
sweep_proven = bool(sweep.get("proven"))
|
sweep_proven = bool(sweep.get("proven"))
|
||||||
@@ -728,7 +719,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
"reasons": ["review mutation proof missing"],
|
"reasons": ["review mutation proof missing"],
|
||||||
}
|
}
|
||||||
review_mutation_complete = bool(review_mutation.get("complete"))
|
review_mutation_complete = bool(review_mutation.get("complete"))
|
||||||
worktree_proven = bool(worktree.get("proven"))
|
|
||||||
|
|
||||||
downgrade_reasons = []
|
downgrade_reasons = []
|
||||||
if not identity_eligible:
|
if not identity_eligible:
|
||||||
@@ -782,11 +772,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
if not review_mutation_complete:
|
if not review_mutation_complete:
|
||||||
downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
|
downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
|
||||||
downgrade_reasons.extend(review_mutation.get("reasons", []))
|
downgrade_reasons.extend(review_mutation.get("reasons", []))
|
||||||
if not worktree_proven:
|
|
||||||
downgrade_reasons.append(
|
|
||||||
"reviewer worktree safety proof missing or failed (#233)"
|
|
||||||
)
|
|
||||||
downgrade_reasons.extend(worktree.get("reasons", []))
|
|
||||||
|
|
||||||
merge_allowed = (
|
merge_allowed = (
|
||||||
identity_eligible
|
identity_eligible
|
||||||
@@ -797,7 +782,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
and validation.get("verdict") != "invalid"
|
and validation.get("verdict") != "invalid"
|
||||||
# #179: no merge without a proven final live-state recheck.
|
# #179: no merge without a proven final live-state recheck.
|
||||||
and live_state_proven
|
and live_state_proven
|
||||||
and worktree_proven
|
|
||||||
)
|
)
|
||||||
|
|
||||||
violations = []
|
violations = []
|
||||||
@@ -841,11 +825,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
"live_state_recheck_proven": live_state_proven,
|
"live_state_recheck_proven": live_state_proven,
|
||||||
"role_boundary_clean": role_boundary_clean,
|
"role_boundary_clean": role_boundary_clean,
|
||||||
"review_mutation_complete": review_mutation_complete,
|
"review_mutation_complete": review_mutation_complete,
|
||||||
"worktree_proof_proven": worktree_proven,
|
|
||||||
"worktree_scratch_used": bool(worktree.get("scratch_used")),
|
|
||||||
"unrelated_mutations_avoided": bool(
|
|
||||||
worktree.get("unrelated_mutations_avoided")
|
|
||||||
),
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -992,12 +971,6 @@ HANDOFF_ROLE_FIELDS = {
|
|||||||
("Selected PR", ("selected pr",)),
|
("Selected PR", ("selected pr",)),
|
||||||
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
|
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
|
||||||
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
|
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
|
||||||
("Worktree path", ("worktree path", "starting worktree path")),
|
|
||||||
("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
|
|
||||||
("Scratch worktree used", ("scratch worktree used", "scratch clone used",
|
|
||||||
"scratch worktree")),
|
|
||||||
("Unrelated local mutations", ("unrelated local mutations",
|
|
||||||
"unrelated files modified")),
|
|
||||||
("Review decision", ("review decision", "decision")),
|
("Review decision", ("review decision", "decision")),
|
||||||
("Merge result", ("merge result",)),
|
("Merge result", ("merge result",)),
|
||||||
("Linked issue status", ("linked issue status", "linked issue")),
|
("Linked issue status", ("linked issue status", "linked issue")),
|
||||||
@@ -1005,7 +978,6 @@ HANDOFF_ROLE_FIELDS = {
|
|||||||
),
|
),
|
||||||
"author": (
|
"author": (
|
||||||
("Selected issue", ("selected issue",)),
|
("Selected issue", ("selected issue",)),
|
||||||
("Issue lock proof", ("issue lock proof", "lock before diff")),
|
|
||||||
("Claim/comment status", ("claim/comment status", "claim status",
|
("Claim/comment status", ("claim/comment status", "claim status",
|
||||||
"claim")),
|
"claim")),
|
||||||
("PR number opened", ("pr number opened", "pr opened", "pr number")),
|
("PR number opened", ("pr number opened", "pr opened", "pr number")),
|
||||||
@@ -1312,93 +1284,6 @@ def pr_inventory_trust_gate(
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]:
|
|
||||||
parts = (full_repo or "").split("/", 1)
|
|
||||||
if len(parts) == 2:
|
|
||||||
return parts[0].strip() or None, parts[1].strip() or None
|
|
||||||
return None, None
|
|
||||||
|
|
||||||
|
|
||||||
def assess_reviewer_queue_inventory(
|
|
||||||
repo_reports: list[dict] | None,
|
|
||||||
required_repos: list[str] | None = None,
|
|
||||||
*,
|
|
||||||
user_context: str | None = None,
|
|
||||||
) -> dict:
|
|
||||||
"""Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
|
|
||||||
|
|
||||||
Any repository reporting ``open_pr_count == 0`` must pass
|
|
||||||
``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue
|
|
||||||
claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient.
|
|
||||||
"""
|
|
||||||
required = list(required_repos or [
|
|
||||||
"Scaled-Tech-Consulting/Gitea-Tools",
|
|
||||||
"Scaled-Tech-Consulting/mcp-control-plane",
|
|
||||||
])
|
|
||||||
completeness = assess_inventory_completeness(repo_reports, required)
|
|
||||||
|
|
||||||
trust_gates: dict[str, dict] = {}
|
|
||||||
blockers: list[str] = []
|
|
||||||
can_claim_empty = bool(completeness.get("complete"))
|
|
||||||
|
|
||||||
for report in repo_reports or []:
|
|
||||||
repo = (report.get("repo") or "").strip()
|
|
||||||
count = report.get("open_pr_count")
|
|
||||||
if not isinstance(count, int) or count != 0:
|
|
||||||
continue
|
|
||||||
|
|
||||||
org, repo_name = _split_repo_slug(repo)
|
|
||||||
list_response = report.get("list_prs_response")
|
|
||||||
if list_response is None:
|
|
||||||
list_response = []
|
|
||||||
|
|
||||||
gate = pr_inventory_trust_gate(
|
|
||||||
list_response,
|
|
||||||
remote=report.get("remote"),
|
|
||||||
org=org,
|
|
||||||
repo=repo_name,
|
|
||||||
state=report.get("state_filter"),
|
|
||||||
authenticated_profile=report.get("authenticated_profile"),
|
|
||||||
local_remote_url=report.get("local_remote_url"),
|
|
||||||
user_context=user_context or report.get("user_context"),
|
|
||||||
corroboration_open_pr_counter=report.get(
|
|
||||||
"corroboration_open_pr_counter"
|
|
||||||
),
|
|
||||||
has_finality_metadata=report.get("pagination_complete") is True,
|
|
||||||
)
|
|
||||||
trust_gates[repo] = gate
|
|
||||||
status = gate.get("status")
|
|
||||||
if status != "trusted_empty":
|
|
||||||
can_claim_empty = False
|
|
||||||
blockers.append(
|
|
||||||
f"repository '{repo}': empty PR list trust gate is "
|
|
||||||
f"'{status}'; cannot claim 'no open PRs'"
|
|
||||||
)
|
|
||||||
blockers.extend(gate.get("reasons") or [])
|
|
||||||
|
|
||||||
return {
|
|
||||||
"complete": bool(completeness.get("complete")),
|
|
||||||
"can_claim_empty_queue": can_claim_empty,
|
|
||||||
"can_claim_exhaustive": (
|
|
||||||
bool(completeness.get("can_claim_exhaustive")) and can_claim_empty
|
|
||||||
),
|
|
||||||
"inventory_reasons": list(completeness.get("reasons") or []),
|
|
||||||
"trust_gates": trust_gates,
|
|
||||||
"blockers": blockers,
|
|
||||||
"reasons": list(completeness.get("reasons") or []) + blockers,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
|
|
||||||
"""Render trust-gate lines for MCP inventory output."""
|
|
||||||
lines = [f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}"]
|
|
||||||
if gate.get("corroborated"):
|
|
||||||
lines.append("pr_inventory_trust_gate.corroborated: true")
|
|
||||||
for reason in gate.get("reasons") or []:
|
|
||||||
lines.append(f"pr_inventory_trust_gate.reason: {reason}")
|
|
||||||
return lines
|
|
||||||
|
|
||||||
|
|
||||||
def assess_duplicate_search_proof(report_text, matches):
|
def assess_duplicate_search_proof(report_text, matches):
|
||||||
"""#207: reject LLM duplicate summaries that omit known title matches."""
|
"""#207: reject LLM duplicate summaries that omit known title matches."""
|
||||||
return issue_duplicate_gate.assess_duplicate_search_proof(
|
return issue_duplicate_gate.assess_duplicate_search_proof(
|
||||||
|
|||||||
@@ -1,219 +0,0 @@
|
|||||||
"""Fail-closed reviewer worktree and local-git safety proofs (#233).
|
|
||||||
|
|
||||||
Reviewer sessions must never stash, reset, or otherwise manipulate unrelated
|
|
||||||
local changes from another session. When the active worktree has dirty tracked
|
|
||||||
files outside the PR scope, the workflow must stop or switch to a disposable
|
|
||||||
scratch worktree (``scripts/worktree-review``).
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import re
|
|
||||||
import shlex
|
|
||||||
|
|
||||||
# Subcommands that mutate unrelated local state — forbidden for reviewers.
|
|
||||||
_FORBIDDEN_REVIEWER_GIT = re.compile(
|
|
||||||
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
|
|
||||||
r"(?:stash(?:\s+(?:push|pop|drop|apply|clear|list))?|"
|
|
||||||
r"checkout\s+--|"
|
|
||||||
r"restore\s+|"
|
|
||||||
r"reset(?:\s+(?:--hard|--soft|--mixed|--merge))?|"
|
|
||||||
r"clean(?:\s+(?:-f|-fd|-fdx|-x|-d|-n))*|"
|
|
||||||
r"cherry-pick|"
|
|
||||||
r"rebase|"
|
|
||||||
r"merge|"
|
|
||||||
r"commit(?:\s+(?:--amend|-a|-am))?|"
|
|
||||||
r"push(?:\s+(?:--force|--force-with-lease))?|"
|
|
||||||
r"branch\s+-[dD]|"
|
|
||||||
r"worktree\s+remove)",
|
|
||||||
re.IGNORECASE,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Read-only git operations reviewers may use for validation.
|
|
||||||
_READONLY_REVIEWER_GIT = re.compile(
|
|
||||||
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
|
|
||||||
r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|"
|
|
||||||
r"worktree\s+list|worktree\s+add)",
|
|
||||||
re.IGNORECASE,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def parse_dirty_tracked_files(porcelain: str) -> list[str]:
|
|
||||||
"""Return tracked paths with local modifications from ``git status --porcelain``.
|
|
||||||
|
|
||||||
Untracked entries (``??``) are ignored — they do not block reviewer work
|
|
||||||
when a scratch worktree is used, and authors may have unrelated untracked
|
|
||||||
files without implying reviewer interference.
|
|
||||||
"""
|
|
||||||
paths: list[str] = []
|
|
||||||
for line in (porcelain or "").splitlines():
|
|
||||||
if not line or len(line) < 4:
|
|
||||||
continue
|
|
||||||
if line.startswith("??"):
|
|
||||||
continue
|
|
||||||
path = line[3:].strip()
|
|
||||||
if " -> " in path:
|
|
||||||
path = path.split(" -> ", 1)[1].strip()
|
|
||||||
if path:
|
|
||||||
paths.append(path)
|
|
||||||
return paths
|
|
||||||
|
|
||||||
|
|
||||||
def files_outside_pr_scope(
|
|
||||||
dirty_files: list[str] | None,
|
|
||||||
pr_scope_files: list[str] | None,
|
|
||||||
) -> list[str]:
|
|
||||||
"""Dirty tracked files not explained by the PR diff file set."""
|
|
||||||
dirty = [p for p in (dirty_files or []) if p]
|
|
||||||
scope = {p for p in (pr_scope_files or []) if p}
|
|
||||||
if not dirty:
|
|
||||||
return []
|
|
||||||
if not scope:
|
|
||||||
return list(dirty)
|
|
||||||
return [path for path in dirty if path not in scope]
|
|
||||||
|
|
||||||
|
|
||||||
def is_forbidden_reviewer_git_command(command: str) -> bool:
|
|
||||||
"""True when a shell command would mutate unrelated local/remote git state."""
|
|
||||||
text = (command or "").strip()
|
|
||||||
if not text:
|
|
||||||
return False
|
|
||||||
return bool(_FORBIDDEN_REVIEWER_GIT.search(text))
|
|
||||||
|
|
||||||
|
|
||||||
def is_readonly_reviewer_git_command(command: str) -> bool:
|
|
||||||
"""True when the command is an explicitly allowed read-only git operation."""
|
|
||||||
text = (command or "").strip()
|
|
||||||
if not text:
|
|
||||||
return False
|
|
||||||
if is_forbidden_reviewer_git_command(text):
|
|
||||||
return False
|
|
||||||
return bool(_READONLY_REVIEWER_GIT.search(text))
|
|
||||||
|
|
||||||
|
|
||||||
def assess_reviewer_git_command_log(commands: list[str] | None) -> dict:
|
|
||||||
"""Fail closed when reviewer shell history includes forbidden git mutations."""
|
|
||||||
forbidden = [
|
|
||||||
cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd)
|
|
||||||
]
|
|
||||||
if forbidden:
|
|
||||||
return {
|
|
||||||
"proven": False,
|
|
||||||
"block": True,
|
|
||||||
"forbidden_commands": forbidden,
|
|
||||||
"reasons": [
|
|
||||||
"reviewer workflow executed forbidden local git mutation: "
|
|
||||||
f"{cmd!r}"
|
|
||||||
for cmd in forbidden
|
|
||||||
],
|
|
||||||
"safe_next_action": (
|
|
||||||
"stop; report worktree interference; do not stash/reset/checkout "
|
|
||||||
"unrelated files — use scripts/worktree-review instead"
|
|
||||||
),
|
|
||||||
}
|
|
||||||
return {
|
|
||||||
"proven": True,
|
|
||||||
"block": False,
|
|
||||||
"forbidden_commands": [],
|
|
||||||
"reasons": [],
|
|
||||||
"safe_next_action": "proceed",
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def assess_reviewer_worktree_proof(proof: dict | None) -> dict:
|
|
||||||
"""Evaluate reviewer worktree safety before checkout/diff/validation/review.
|
|
||||||
|
|
||||||
*proof* keys:
|
|
||||||
- ``worktree_path`` (required)
|
|
||||||
- ``porcelain_status`` or ``dirty_files``
|
|
||||||
- ``pr_scope_files`` (paths in the PR diff)
|
|
||||||
- ``scratch_used`` (bool)
|
|
||||||
- ``scratch_path`` (when scratch_used)
|
|
||||||
- ``git_commands`` (shell commands executed this session)
|
|
||||||
- ``unrelated_mutations_claimed`` (bool) — stash/reset/drop reported
|
|
||||||
"""
|
|
||||||
proof = dict(proof or {})
|
|
||||||
reasons: list[str] = []
|
|
||||||
worktree_path = (proof.get("worktree_path") or "").strip()
|
|
||||||
if not worktree_path:
|
|
||||||
reasons.append("reviewer worktree path not reported; fail closed")
|
|
||||||
|
|
||||||
if proof.get("dirty_files") is not None:
|
|
||||||
dirty_files = list(proof.get("dirty_files") or [])
|
|
||||||
else:
|
|
||||||
dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "")
|
|
||||||
|
|
||||||
pr_scope = list(proof.get("pr_scope_files") or [])
|
|
||||||
unrelated = files_outside_pr_scope(dirty_files, pr_scope)
|
|
||||||
scratch_used = bool(proof.get("scratch_used"))
|
|
||||||
scratch_path = (proof.get("scratch_path") or "").strip()
|
|
||||||
|
|
||||||
is_dirty = bool(dirty_files)
|
|
||||||
unrelated_dirty = bool(unrelated)
|
|
||||||
|
|
||||||
if unrelated_dirty and not scratch_used:
|
|
||||||
reasons.append(
|
|
||||||
"worktree has dirty tracked files outside PR scope "
|
|
||||||
f"({', '.join(unrelated)}); stop or use a scratch worktree"
|
|
||||||
)
|
|
||||||
if scratch_used and not scratch_path:
|
|
||||||
reasons.append(
|
|
||||||
"scratch worktree was used but scratch_path was not reported"
|
|
||||||
)
|
|
||||||
if proof.get("unrelated_mutations_claimed"):
|
|
||||||
reasons.append(
|
|
||||||
"reviewer reported stash/reset/checkout cleanup of unrelated "
|
|
||||||
"local changes; this is forbidden"
|
|
||||||
)
|
|
||||||
|
|
||||||
command_assessment = assess_reviewer_git_command_log(
|
|
||||||
list(proof.get("git_commands") or [])
|
|
||||||
)
|
|
||||||
if command_assessment["block"]:
|
|
||||||
reasons.extend(command_assessment["reasons"])
|
|
||||||
|
|
||||||
proven = not reasons
|
|
||||||
return {
|
|
||||||
"proven": proven,
|
|
||||||
"block": not proven,
|
|
||||||
"reasons": reasons,
|
|
||||||
"worktree_path": worktree_path or None,
|
|
||||||
"is_dirty": is_dirty,
|
|
||||||
"dirty_files": dirty_files,
|
|
||||||
"unrelated_dirty_files": unrelated,
|
|
||||||
"scratch_used": scratch_used,
|
|
||||||
"scratch_path": scratch_path or None,
|
|
||||||
"unrelated_mutations_avoided": not bool(
|
|
||||||
proof.get("unrelated_mutations_claimed")
|
|
||||||
or command_assessment.get("forbidden_commands")
|
|
||||||
),
|
|
||||||
"safe_next_action": (
|
|
||||||
"proceed"
|
|
||||||
if proven
|
|
||||||
else command_assessment.get("safe_next_action")
|
|
||||||
or "stop; use scripts/worktree-review or report dirty worktree"
|
|
||||||
),
|
|
||||||
"forbidden_commands": command_assessment.get("forbidden_commands", []),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def assess_author_worktree_continuity(proof: dict | None) -> dict:
|
|
||||||
"""Authors may keep dirty feature worktrees; reviewers may not manipulate them.
|
|
||||||
|
|
||||||
This helper only proves the task role is author when dirty unrelated files
|
|
||||||
exist — it does not grant reviewers an exception.
|
|
||||||
"""
|
|
||||||
proof = dict(proof or {})
|
|
||||||
role = (proof.get("task_role") or "").strip().lower()
|
|
||||||
dirty_files = list(proof.get("dirty_files") or [])
|
|
||||||
if role == "author" and dirty_files:
|
|
||||||
return {
|
|
||||||
"allowed": True,
|
|
||||||
"reasons": [
|
|
||||||
"author task may continue with dirty tracked files in its own "
|
|
||||||
"worktree; reviewer interference rules do not apply"
|
|
||||||
],
|
|
||||||
}
|
|
||||||
if role == "reviewer" and dirty_files:
|
|
||||||
return assess_reviewer_worktree_proof(proof)
|
|
||||||
return {"allowed": True, "reasons": []}
|
|
||||||
@@ -17,24 +17,9 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
|
|||||||
configured repos were not checked. This is not a complete queue inventory."
|
configured repos were not checked. This is not a complete queue inventory."
|
||||||
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
|
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
|
||||||
if the other configured repo was not inventoried.
|
if the other configured repo was not inventoried.
|
||||||
- PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty",
|
|
||||||
the workflow must run `pr_inventory_trust_gate` (via the live inventory path or
|
|
||||||
`review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a
|
|
||||||
clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
|
|
||||||
corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
|
|
||||||
sufficient proof.
|
|
||||||
|
|
||||||
Rules (llm-project-workflow):
|
Rules (llm-project-workflow):
|
||||||
- Review in a SEPARATE detached review worktree, never the author's folder.
|
- Review in a SEPARATE detached review worktree, never the author's folder.
|
||||||
- Worktree safety (#233): before checkout, diff, validation, review, or merge,
|
|
||||||
report the starting worktree path and whether it was dirty. If unrelated
|
|
||||||
tracked files exist outside the PR scope, STOP or run
|
|
||||||
`scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
|
|
||||||
NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
|
|
||||||
or `git clean` to manage another session's dirty files.
|
|
||||||
- Final report must state: Worktree path, Worktree dirty (yes/no),
|
|
||||||
Scratch worktree used (yes/no + path if yes), and confirm no unrelated local
|
|
||||||
files were modified, stashed, reset, or dropped.
|
|
||||||
- You must NOT be the PR author. If the authenticated user == PR author, stop.
|
- You must NOT be the PR author. If the authenticated user == PR author, stop.
|
||||||
A different LLM-Agent-SHA does NOT make you a different actor — only a
|
A different LLM-Agent-SHA does NOT make you a different actor — only a
|
||||||
different authenticated Gitea user does (docs/llm-agent-sha.md).
|
different authenticated Gitea user does (docs/llm-agent-sha.md).
|
||||||
|
|||||||
+13
-49
@@ -65,20 +65,6 @@ CREATE_PR_ENV = {
|
|||||||
),
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
|
|
||||||
|
|
||||||
|
|
||||||
def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
|
|
||||||
record = {
|
|
||||||
"issue_number": issue_number,
|
|
||||||
"branch_name": branch_name,
|
|
||||||
"remote": "dadeschools",
|
|
||||||
"org": "Scaled-Tech-Consulting",
|
|
||||||
"repo": "Gitea-Tools",
|
|
||||||
}
|
|
||||||
record.update(overrides)
|
|
||||||
return record
|
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Create Issue
|
# Create Issue
|
||||||
@@ -141,19 +127,15 @@ class TestCreatePR(unittest.TestCase):
|
|||||||
@patch("os.path.exists", return_value=True)
|
@patch("os.path.exists", return_value=True)
|
||||||
@patch("builtins.open")
|
@patch("builtins.open")
|
||||||
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
|
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
|
||||||
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
|
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
|
||||||
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
|
|
||||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||||
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
||||||
self.assertEqual(result["number"], 3)
|
self.assertEqual(result["number"], 3)
|
||||||
self.assertNotIn("url", result)
|
self.assertNotIn("url", result)
|
||||||
mock_exists.assert_called_with(ISSUE_LOCK_FILE)
|
|
||||||
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
|
|
||||||
payload = mock_api.call_args[0][3]
|
payload = mock_api.call_args[0][3]
|
||||||
self.assertEqual(payload["head"], "feat/x")
|
self.assertEqual(payload["head"], "feat/x")
|
||||||
self.assertEqual(payload["base"], "main")
|
self.assertEqual(payload["base"], "main")
|
||||||
self.assertIn("Closes #123", payload["title"])
|
|
||||||
|
|
||||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||||
return_value=(True, []))
|
return_value=(True, []))
|
||||||
@@ -162,28 +144,13 @@ class TestCreatePR(unittest.TestCase):
|
|||||||
@patch("os.path.exists", return_value=True)
|
@patch("os.path.exists", return_value=True)
|
||||||
@patch("builtins.open")
|
@patch("builtins.open")
|
||||||
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
|
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
|
||||||
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
|
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
|
||||||
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
|
|
||||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||||
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
|
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
|
||||||
with patch.dict(os.environ, env, clear=True):
|
with patch.dict(os.environ, env, clear=True):
|
||||||
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
||||||
self.assertIn("pulls/3", result["url"])
|
self.assertIn("pulls/3", result["url"])
|
||||||
|
|
||||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
|
||||||
return_value=(True, []))
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
@patch("os.path.exists", return_value=True)
|
|
||||||
@patch("builtins.open")
|
|
||||||
def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
|
|
||||||
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
|
|
||||||
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
|
|
||||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
|
||||||
with self.assertRaises(ValueError) as ctx:
|
|
||||||
gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
|
|
||||||
self.assertIn("Closes #123", str(ctx.exception))
|
|
||||||
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
|
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Close Issue
|
# Close Issue
|
||||||
@@ -2803,8 +2770,8 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
"""Test issue locking and PR gating constraints."""
|
"""Test issue locking and PR gating constraints."""
|
||||||
|
|
||||||
def tearDown(self):
|
def tearDown(self):
|
||||||
if os.path.exists(ISSUE_LOCK_FILE):
|
if os.path.exists("/tmp/gitea_issue_lock.json"):
|
||||||
os.remove(ISSUE_LOCK_FILE)
|
os.remove("/tmp/gitea_issue_lock.json")
|
||||||
|
|
||||||
@patch("mcp_server.api_get_all")
|
@patch("mcp_server.api_get_all")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
@@ -2812,7 +2779,7 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
mock_api.return_value = [] # no open PRs
|
mock_api.return_value = [] # no open PRs
|
||||||
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
||||||
self.assertTrue(res["success"])
|
self.assertTrue(res["success"])
|
||||||
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
|
self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
|
||||||
|
|
||||||
def test_lock_issue_mismatch_branch_fails(self):
|
def test_lock_issue_mismatch_branch_fails(self):
|
||||||
with self.assertRaises(ValueError) as ctx:
|
with self.assertRaises(ValueError) as ctx:
|
||||||
@@ -2849,8 +2816,8 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
return_value=(True, []))
|
return_value=(True, []))
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
def test_create_pr_missing_lock_fails(self, _auth, _role):
|
def test_create_pr_missing_lock_fails(self, _auth, _role):
|
||||||
if os.path.exists(ISSUE_LOCK_FILE):
|
if os.path.exists("/tmp/gitea_issue_lock.json"):
|
||||||
os.remove(ISSUE_LOCK_FILE)
|
os.remove("/tmp/gitea_issue_lock.json")
|
||||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
with self.assertRaises(RuntimeError) as ctx:
|
||||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
|
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
|
||||||
@@ -2860,9 +2827,8 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
return_value=(True, []))
|
return_value=(True, []))
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
|
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
|
||||||
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
||||||
json.dump(_sample_issue_lock(
|
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
||||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
|
||||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||||
with self.assertRaises(ValueError) as ctx:
|
with self.assertRaises(ValueError) as ctx:
|
||||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
|
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
|
||||||
@@ -2872,9 +2838,8 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
return_value=(True, []))
|
return_value=(True, []))
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
|
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
|
||||||
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
||||||
json.dump(_sample_issue_lock(
|
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
||||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
|
||||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||||
for term in ("equivalent to #196", "related to #196", "same as #196"):
|
for term in ("equivalent to #196", "related to #196", "same as #196"):
|
||||||
with self.assertRaises(ValueError) as ctx:
|
with self.assertRaises(ValueError) as ctx:
|
||||||
@@ -2885,9 +2850,8 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
return_value=(True, []))
|
return_value=(True, []))
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
|
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
|
||||||
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
||||||
json.dump(_sample_issue_lock(
|
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
||||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
|
||||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||||
with self.assertRaises(ValueError) as ctx:
|
with self.assertRaises(ValueError) as ctx:
|
||||||
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
|
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
|
||||||
|
|||||||
@@ -1,181 +0,0 @@
|
|||||||
"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
|
|
||||||
import os
|
|
||||||
import sys
|
|
||||||
import json
|
|
||||||
import tempfile
|
|
||||||
import unittest
|
|
||||||
from unittest.mock import patch, MagicMock
|
|
||||||
|
|
||||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
||||||
|
|
||||||
import gitea_config
|
|
||||||
import gitea_auth
|
|
||||||
import mcp_server
|
|
||||||
from reviewer_worktree import assess_author_worktree_continuity
|
|
||||||
|
|
||||||
CONFIG_TEST = {
|
|
||||||
"version": 2,
|
|
||||||
"contexts": {
|
|
||||||
"ctx": {
|
|
||||||
"enabled": True,
|
|
||||||
"gitea": {
|
|
||||||
"enabled": True,
|
|
||||||
"base_url": "https://gitea.example.com"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"profiles": {
|
|
||||||
"author-profile": {
|
|
||||||
"enabled": True,
|
|
||||||
"context": "ctx",
|
|
||||||
"role": "author",
|
|
||||||
"username": "author-user",
|
|
||||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
|
||||||
"allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
|
|
||||||
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
|
|
||||||
"execution_profile": "author-profile"
|
|
||||||
},
|
|
||||||
"reviewer-profile": {
|
|
||||||
"enabled": True,
|
|
||||||
"context": "ctx",
|
|
||||||
"role": "reviewer",
|
|
||||||
"username": "reviewer-user",
|
|
||||||
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
|
||||||
"allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"],
|
|
||||||
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
|
||||||
"execution_profile": "reviewer-profile"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
class TestOperationScopedRoles(unittest.TestCase):
|
|
||||||
def setUp(self):
|
|
||||||
self._remotes_patch = patch.dict(mcp_server.REMOTES, {
|
|
||||||
"dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
|
|
||||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
|
|
||||||
})
|
|
||||||
self._remotes_patch.start()
|
|
||||||
mcp_server._IDENTITY_CACHE.clear()
|
|
||||||
gitea_config._active_profile_override = None
|
|
||||||
mcp_server._MUTATION_AUTHORITY = None
|
|
||||||
self._dir = tempfile.TemporaryDirectory()
|
|
||||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
|
||||||
self._write_config(CONFIG_TEST)
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
self._remotes_patch.stop()
|
|
||||||
mcp_server._IDENTITY_CACHE.clear()
|
|
||||||
gitea_config._active_profile_override = None
|
|
||||||
mcp_server._MUTATION_AUTHORITY = None
|
|
||||||
self._dir.cleanup()
|
|
||||||
|
|
||||||
def _write_config(self, obj):
|
|
||||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
|
||||||
fh.write(json.dumps(obj))
|
|
||||||
|
|
||||||
def _env(self, profile="author-profile", with_reviewer_token=True):
|
|
||||||
env = {
|
|
||||||
"GITEA_MCP_CONFIG": self.config_path,
|
|
||||||
"GITEA_MCP_PROFILE": profile,
|
|
||||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
|
||||||
}
|
|
||||||
if with_reviewer_token:
|
|
||||||
env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass"
|
|
||||||
return env
|
|
||||||
|
|
||||||
@patch("mcp_server.api_request")
|
|
||||||
def test_auto_switch_to_reviewer(self, mock_api):
|
|
||||||
# mock identity resolution to return username matching profile
|
|
||||||
mock_api.side_effect = lambda method, url, header: (
|
|
||||||
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
|
|
||||||
)
|
|
||||||
with patch.dict(os.environ, self._env("author-profile")):
|
|
||||||
# initially we are author-profile
|
|
||||||
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
|
|
||||||
self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
|
|
||||||
|
|
||||||
# resolve a reviewer task (review_pr)
|
|
||||||
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
|
|
||||||
|
|
||||||
# verify it automatically switched to reviewer-profile
|
|
||||||
self.assertTrue(res["allowed_in_current_session"])
|
|
||||||
self.assertTrue(res["available_in_session"])
|
|
||||||
self.assertEqual(res["active_profile"], "reviewer-profile")
|
|
||||||
self.assertEqual(res["active_identity"], "reviewer-user")
|
|
||||||
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
|
|
||||||
|
|
||||||
@patch("mcp_server.api_request")
|
|
||||||
def test_auto_switch_to_author(self, mock_api):
|
|
||||||
mock_api.side_effect = lambda method, url, header: (
|
|
||||||
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
|
|
||||||
)
|
|
||||||
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
||||||
# initially we are reviewer-profile
|
|
||||||
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
|
|
||||||
|
|
||||||
# resolve an author task (create_issue)
|
|
||||||
res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
|
|
||||||
|
|
||||||
# verify it automatically switched to author-profile
|
|
||||||
self.assertTrue(res["allowed_in_current_session"])
|
|
||||||
self.assertTrue(res["available_in_session"])
|
|
||||||
self.assertEqual(res["active_profile"], "author-profile")
|
|
||||||
self.assertEqual(res["active_identity"], "author-user")
|
|
||||||
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
|
|
||||||
|
|
||||||
@patch("mcp_server.api_request")
|
|
||||||
def test_restart_required_when_unattached(self, mock_api):
|
|
||||||
mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
|
|
||||||
# launch without reviewer token in env
|
|
||||||
with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
|
|
||||||
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
|
|
||||||
|
|
||||||
# resolve a reviewer task (review_pr)
|
|
||||||
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
|
|
||||||
|
|
||||||
# verify it did NOT switch and reports restart_required
|
|
||||||
self.assertFalse(res["allowed_in_current_session"])
|
|
||||||
self.assertFalse(res["available_in_session"])
|
|
||||||
self.assertTrue(res["configured"])
|
|
||||||
self.assertTrue(res["restart_required"])
|
|
||||||
self.assertTrue(res["stop_required"])
|
|
||||||
self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
|
|
||||||
|
|
||||||
def test_author_continuity_dirty_worktree(self):
|
|
||||||
# author is allowed to keep dirty worktree
|
|
||||||
res = assess_author_worktree_continuity({
|
|
||||||
"task_role": "author",
|
|
||||||
"dirty_files": ["review_proofs.py"],
|
|
||||||
})
|
|
||||||
self.assertTrue(res["allowed"])
|
|
||||||
|
|
||||||
# reviewer is blocked by reviewer worktree proof
|
|
||||||
res2 = assess_author_worktree_continuity({
|
|
||||||
"task_role": "reviewer",
|
|
||||||
"worktree_path": "/repo",
|
|
||||||
"dirty_files": ["review_proofs.py"],
|
|
||||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
|
||||||
"scratch_used": False,
|
|
||||||
})
|
|
||||||
self.assertFalse(res2["proven"])
|
|
||||||
|
|
||||||
@patch("mcp_server.api_request")
|
|
||||||
def test_mutating_actions_auto_switch(self, mock_api):
|
|
||||||
mock_api.side_effect = lambda method, url, header: (
|
|
||||||
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
|
|
||||||
)
|
|
||||||
with patch.dict(os.environ, self._env("author-profile")):
|
|
||||||
# verify we are author-profile
|
|
||||||
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
|
|
||||||
|
|
||||||
# call a reviewer mutation check helper (like _profile_permission_block with reviewer permission)
|
|
||||||
blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs")
|
|
||||||
self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block)
|
|
||||||
|
|
||||||
# verify we dynamically switched to reviewer-profile
|
|
||||||
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
@@ -261,79 +261,12 @@ class TestPRQueueInventory(unittest.TestCase):
|
|||||||
for call in mock_api.call_args_list:
|
for call in mock_api.call_args_list:
|
||||||
self.assertEqual(call.args[0], "GET")
|
self.assertEqual(call.args[0], "GET")
|
||||||
|
|
||||||
@patch("mcp_server._local_git_remote_url")
|
|
||||||
@patch("mcp_server.api_get_all")
|
@patch("mcp_server.api_get_all")
|
||||||
@patch("mcp_server.api_request")
|
@patch("mcp_server.api_request")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
@patch("mcp_server.get_profile")
|
@patch("mcp_server.get_profile")
|
||||||
def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty(
|
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
|
||||||
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
|
|
||||||
):
|
|
||||||
mock_get_profile.return_value = {
|
|
||||||
"profile_name": "gitea-author",
|
|
||||||
"allowed_operations": ["read"],
|
|
||||||
"forbidden_operations": [],
|
|
||||||
"base_url": None,
|
|
||||||
}
|
|
||||||
mock_get_all.return_value = []
|
|
||||||
mock_api.return_value = {"login": "jcwalker3"}
|
|
||||||
mock_local_url.return_value = None
|
|
||||||
|
|
||||||
result = gitea_review_pr(
|
|
||||||
pr_number=1,
|
|
||||||
event="APPROVE",
|
|
||||||
remote="prgs",
|
|
||||||
org="Scaled-Tech-Consulting",
|
|
||||||
repo="Gitea-Tools",
|
|
||||||
)
|
|
||||||
self.assertFalse(result["success"])
|
|
||||||
msg = result["message"]
|
|
||||||
self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg)
|
|
||||||
self.assertIn("Empty-queue claim blocked", msg)
|
|
||||||
self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg)
|
|
||||||
|
|
||||||
@patch("mcp_server._local_git_remote_url")
|
|
||||||
@patch("mcp_server.api_get_all")
|
|
||||||
@patch("mcp_server.api_request")
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
@patch("mcp_server.get_profile")
|
|
||||||
def test_empty_inventory_trusted_empty_when_gate_passes(
|
|
||||||
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
|
|
||||||
):
|
|
||||||
mock_get_profile.return_value = {
|
|
||||||
"profile_name": "gitea-author",
|
|
||||||
"allowed_operations": ["read"],
|
|
||||||
"forbidden_operations": [],
|
|
||||||
"base_url": None,
|
|
||||||
}
|
|
||||||
mock_get_all.return_value = []
|
|
||||||
mock_api.return_value = {"login": "jcwalker3"}
|
|
||||||
mock_local_url.return_value = (
|
|
||||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
|
||||||
)
|
|
||||||
|
|
||||||
result = gitea_review_pr(
|
|
||||||
pr_number=1,
|
|
||||||
event="APPROVE",
|
|
||||||
remote="prgs",
|
|
||||||
org="Scaled-Tech-Consulting",
|
|
||||||
repo="Gitea-Tools",
|
|
||||||
)
|
|
||||||
self.assertFalse(result["success"])
|
|
||||||
msg = result["message"]
|
|
||||||
self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg)
|
|
||||||
self.assertIn("Open PRs found: 0 (trusted_empty)", msg)
|
|
||||||
|
|
||||||
@patch("mcp_server._local_git_remote_url")
|
|
||||||
@patch("mcp_server.api_get_all")
|
|
||||||
@patch("mcp_server.api_request")
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
@patch("mcp_server.get_profile")
|
|
||||||
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url):
|
|
||||||
"""Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
|
"""Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
|
||||||
mock_local_url.return_value = (
|
|
||||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
|
||||||
)
|
|
||||||
for event in ["APPROVE", "REQUEST_CHANGES"]:
|
for event in ["APPROVE", "REQUEST_CHANGES"]:
|
||||||
mock_get_profile.return_value = {
|
mock_get_profile.return_value = {
|
||||||
"profile_name": "gitea-author",
|
"profile_name": "gitea-author",
|
||||||
|
|||||||
+18
-142
@@ -26,7 +26,6 @@ from review_proofs import ( # noqa: E402
|
|||||||
assess_capability_proof,
|
assess_capability_proof,
|
||||||
assess_controller_handoff,
|
assess_controller_handoff,
|
||||||
assess_inventory_completeness,
|
assess_inventory_completeness,
|
||||||
assess_reviewer_queue_inventory,
|
|
||||||
assess_live_state_recheck,
|
assess_live_state_recheck,
|
||||||
assess_review_mutation_final_report,
|
assess_review_mutation_final_report,
|
||||||
assess_role_boundary,
|
assess_role_boundary,
|
||||||
@@ -184,24 +183,6 @@ def _good_review_mutation():
|
|||||||
return assess_review_mutation_final_report(report, lock)
|
return assess_review_mutation_final_report(report, lock)
|
||||||
|
|
||||||
|
|
||||||
def _good_worktree(**overrides):
|
|
||||||
proof = {
|
|
||||||
"worktree_path": "/repo/branches/review-feat-issue-224",
|
|
||||||
"porcelain_status": "",
|
|
||||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
|
||||||
"scratch_used": True,
|
|
||||||
"scratch_path": "/repo/branches/review-feat-issue-224",
|
|
||||||
"git_commands": [
|
|
||||||
"git fetch prgs master feat/issue-224-wiki-proof-refresh",
|
|
||||||
"git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
|
|
||||||
],
|
|
||||||
}
|
|
||||||
proof.update(overrides)
|
|
||||||
from reviewer_worktree import assess_reviewer_worktree_proof # noqa: E402
|
|
||||||
|
|
||||||
return assess_reviewer_worktree_proof(proof)
|
|
||||||
|
|
||||||
|
|
||||||
def _good_role_boundary_179(**overrides):
|
def _good_role_boundary_179(**overrides):
|
||||||
kwargs = {
|
kwargs = {
|
||||||
"task_role": "reviewer",
|
"task_role": "reviewer",
|
||||||
@@ -614,13 +595,6 @@ class TestFinalReport(unittest.TestCase):
|
|||||||
"controller_handoff": _good_handoff(),
|
"controller_handoff": _good_handoff(),
|
||||||
"capability_proof": _good_capability_proof(),
|
"capability_proof": _good_capability_proof(),
|
||||||
"sweep_proof": _good_secret_sweep(),
|
"sweep_proof": _good_secret_sweep(),
|
||||||
"worktree_proof": {
|
|
||||||
"worktree_path": "/repo/branches/review-feat-issue-224",
|
|
||||||
"porcelain_status": "",
|
|
||||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
|
||||||
"scratch_used": True,
|
|
||||||
"scratch_path": "/repo/branches/review-feat-issue-224",
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
kwargs.update(overrides)
|
kwargs.update(overrides)
|
||||||
return build_final_report(**kwargs)
|
return build_final_report(**kwargs)
|
||||||
@@ -925,17 +899,12 @@ class TestControllerHandoff(unittest.TestCase):
|
|||||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
|
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
|
||||||
self.assertEqual(result["verdict"], "incomplete")
|
self.assertEqual(result["verdict"], "incomplete")
|
||||||
self.assertIn("Pinned reviewed head", result["missing_fields"])
|
self.assertIn("Pinned reviewed head", result["missing_fields"])
|
||||||
self.assertIn("Worktree path", result["missing_fields"])
|
|
||||||
self.assertIn("Merge result", result["missing_fields"])
|
self.assertIn("Merge result", result["missing_fields"])
|
||||||
|
|
||||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||||
"- Selected PR: #999",
|
"- Selected PR: #999",
|
||||||
"- Reviewer eligibility: passed",
|
"- Reviewer eligibility: passed",
|
||||||
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||||
"- Worktree path: /repo/branches/review-pr-999",
|
|
||||||
"- Worktree dirty: no",
|
|
||||||
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
|
|
||||||
"- Unrelated local mutations: none",
|
|
||||||
"- Review decision: approve",
|
"- Review decision: approve",
|
||||||
"- Merge result: merged",
|
"- Merge result: merged",
|
||||||
"- Linked issue status: closed",
|
"- Linked issue status: closed",
|
||||||
@@ -944,46 +913,24 @@ class TestControllerHandoff(unittest.TestCase):
|
|||||||
result = assess_controller_handoff(complete, role="review")
|
result = assess_controller_handoff(complete, role="review")
|
||||||
self.assertEqual(result["verdict"], "complete")
|
self.assertEqual(result["verdict"], "complete")
|
||||||
|
|
||||||
def _author_role_fields(self, issue_number=182, pr_number=999):
|
|
||||||
return [
|
|
||||||
f"- Selected issue: #{issue_number}",
|
|
||||||
"- Issue lock proof: lock before diff on feat/x @ master",
|
|
||||||
"- Claim/comment status: comment-claimed",
|
|
||||||
f"- PR number opened: #{pr_number}",
|
|
||||||
"- No review/merge: confirmed",
|
|
||||||
]
|
|
||||||
|
|
||||||
def test_handoff_role_fields_author_includes_issue_lock_proof(self):
|
|
||||||
from review_proofs import HANDOFF_ROLE_FIELDS
|
|
||||||
names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
|
|
||||||
self.assertIn("Issue lock proof", names)
|
|
||||||
|
|
||||||
def test_author_role_requires_author_fields(self):
|
def test_author_role_requires_author_fields(self):
|
||||||
complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
|
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||||
result = assess_controller_handoff(complete, role="author")
|
|
||||||
self.assertEqual(result["verdict"], "complete")
|
|
||||||
|
|
||||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
|
|
||||||
self.assertEqual(result["verdict"], "incomplete")
|
|
||||||
self.assertIn("Issue lock proof", result["missing_fields"])
|
|
||||||
self.assertIn("No review/merge confirmation", result["missing_fields"])
|
|
||||||
|
|
||||||
def test_author_role_requires_issue_lock_proof(self):
|
|
||||||
without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
|
|
||||||
"- Selected issue: #182",
|
"- Selected issue: #182",
|
||||||
"- Claim/comment status: comment-claimed",
|
"- Claim/comment status: comment-claimed",
|
||||||
"- PR number opened: #999",
|
"- PR number opened: #999",
|
||||||
"- No review/merge: confirmed",
|
"- No review/merge: confirmed",
|
||||||
])
|
])
|
||||||
result = assess_controller_handoff(without_lock, role="author")
|
result = assess_controller_handoff(complete, role="author")
|
||||||
|
self.assertEqual(result["verdict"], "complete")
|
||||||
|
|
||||||
|
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
|
||||||
self.assertEqual(result["verdict"], "incomplete")
|
self.assertEqual(result["verdict"], "incomplete")
|
||||||
self.assertIn("Issue lock proof", result["missing_fields"])
|
self.assertIn("No review/merge confirmation", result["missing_fields"])
|
||||||
|
|
||||||
def test_author_role_rejects_equivalent_or_multiple_issues(self):
|
def test_author_role_rejects_equivalent_or_multiple_issues(self):
|
||||||
# 1. equivalent reference blocked
|
# 1. equivalent reference blocked
|
||||||
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
|
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||||
"- Selected issue: Issue #194 / #196 equivalent",
|
"- Selected issue: Issue #194 / #196 equivalent",
|
||||||
"- Issue lock proof: lock before diff on feat/x @ master",
|
|
||||||
"- Claim/comment status: comment-claimed",
|
"- Claim/comment status: comment-claimed",
|
||||||
"- PR number opened: #999",
|
"- PR number opened: #999",
|
||||||
"- No review/merge: confirmed",
|
"- No review/merge: confirmed",
|
||||||
@@ -995,7 +942,6 @@ class TestControllerHandoff(unittest.TestCase):
|
|||||||
# 2. multiple issues blocked
|
# 2. multiple issues blocked
|
||||||
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
|
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||||
"- Selected issue: #194, #196",
|
"- Selected issue: #194, #196",
|
||||||
"- Issue lock proof: lock before diff on feat/x @ master",
|
|
||||||
"- Claim/comment status: comment-claimed",
|
"- Claim/comment status: comment-claimed",
|
||||||
"- PR number opened: #999",
|
"- PR number opened: #999",
|
||||||
"- No review/merge: confirmed",
|
"- No review/merge: confirmed",
|
||||||
@@ -1007,7 +953,6 @@ class TestControllerHandoff(unittest.TestCase):
|
|||||||
def test_author_role_rejects_fuzzy_pr_number(self):
|
def test_author_role_rejects_fuzzy_pr_number(self):
|
||||||
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
|
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||||
"- Selected issue: #196",
|
"- Selected issue: #196",
|
||||||
"- Issue lock proof: lock before diff on feat/issue-196 @ master",
|
|
||||||
"- Claim/comment status: comment-claimed",
|
"- Claim/comment status: comment-claimed",
|
||||||
"- PR number opened: PR #203 / #204 equivalent",
|
"- PR number opened: PR #203 / #204 equivalent",
|
||||||
"- No review/merge: confirmed",
|
"- No review/merge: confirmed",
|
||||||
@@ -1039,17 +984,23 @@ class TestControllerHandoff(unittest.TestCase):
|
|||||||
|
|
||||||
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
|
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
|
||||||
# 1. Workspace mutations: none is rejected when local_edits is True
|
# 1. Workspace mutations: none is rejected when local_edits is True
|
||||||
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
|
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||||
self._author_role_fields(issue_number=196, pr_number=203))
|
"- Selected issue: #196",
|
||||||
|
"- Claim/comment status: comment-claimed",
|
||||||
|
"- PR number opened: #203",
|
||||||
|
"- No review/merge: confirmed",
|
||||||
|
])
|
||||||
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
|
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
|
||||||
self.assertEqual(res["verdict"], "incomplete")
|
self.assertEqual(res["verdict"], "incomplete")
|
||||||
self.assertIn("Workspace mutations", res["missing_fields"])
|
self.assertIn("Workspace mutations", res["missing_fields"])
|
||||||
|
|
||||||
# 2. Workspace mutations: edited files is allowed when local_edits is True
|
# 2. Workspace mutations: edited files is allowed when local_edits is True
|
||||||
complete_eq = self.BASE_HANDOFF.replace(
|
complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
|
||||||
"- Workspace mutations: none",
|
"- Selected issue: #196",
|
||||||
"- Workspace mutations: edited review_proofs.py",
|
"- Claim/comment status: comment-claimed",
|
||||||
) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
|
"- PR number opened: #203",
|
||||||
|
"- No review/merge: confirmed",
|
||||||
|
])
|
||||||
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
|
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
|
||||||
self.assertEqual(res2["verdict"], "complete")
|
self.assertEqual(res2["verdict"], "complete")
|
||||||
|
|
||||||
@@ -1212,74 +1163,6 @@ class TestPRInventoryTrustGate(unittest.TestCase):
|
|||||||
self.assertTrue(res["corroborated"])
|
self.assertTrue(res["corroborated"])
|
||||||
|
|
||||||
|
|
||||||
class TestAssessReviewerQueueInventory(unittest.TestCase):
|
|
||||||
"""Issue #196: trust gate wired into canonical queue inventory."""
|
|
||||||
|
|
||||||
def _repo_report(self, **overrides):
|
|
||||||
report = {
|
|
||||||
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
|
|
||||||
"state_filter": "open",
|
|
||||||
"pagination_complete": True,
|
|
||||||
"open_pr_count": 0,
|
|
||||||
"list_prs_response": [],
|
|
||||||
"remote": "prgs",
|
|
||||||
"authenticated_profile": {
|
|
||||||
"profile_name": "prgs-reviewer",
|
|
||||||
"allowed_operations": ["read", "gitea.read"],
|
|
||||||
},
|
|
||||||
"local_remote_url": (
|
|
||||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
|
||||||
),
|
|
||||||
}
|
|
||||||
report.update(overrides)
|
|
||||||
return report
|
|
||||||
|
|
||||||
def test_empty_without_corroboration_blocks_empty_queue_claim(self):
|
|
||||||
result = assess_reviewer_queue_inventory([
|
|
||||||
self._repo_report(pagination_complete=False),
|
|
||||||
self._repo_report(
|
|
||||||
repo="Scaled-Tech-Consulting/mcp-control-plane",
|
|
||||||
open_pr_count=0,
|
|
||||||
pagination_complete=False,
|
|
||||||
),
|
|
||||||
])
|
|
||||||
self.assertFalse(result["can_claim_empty_queue"])
|
|
||||||
self.assertIn("untrusted_empty", str(result["trust_gates"]))
|
|
||||||
|
|
||||||
def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
|
|
||||||
result = assess_reviewer_queue_inventory([
|
|
||||||
self._repo_report(),
|
|
||||||
self._repo_report(
|
|
||||||
repo="Scaled-Tech-Consulting/mcp-control-plane",
|
|
||||||
local_remote_url=(
|
|
||||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/"
|
|
||||||
"mcp-control-plane.git"
|
|
||||||
),
|
|
||||||
),
|
|
||||||
])
|
|
||||||
self.assertTrue(result["can_claim_empty_queue"])
|
|
||||||
self.assertTrue(result["can_claim_exhaustive"])
|
|
||||||
|
|
||||||
def test_user_context_indicating_open_prs_blocks_empty_claim(self):
|
|
||||||
result = assess_reviewer_queue_inventory(
|
|
||||||
[self._repo_report()],
|
|
||||||
user_context="please review open PR #195 in the queue",
|
|
||||||
)
|
|
||||||
self.assertFalse(result["can_claim_empty_queue"])
|
|
||||||
self.assertTrue(result["blockers"])
|
|
||||||
|
|
||||||
def test_nonempty_inventory_skips_empty_trust_gate_block(self):
|
|
||||||
result = assess_reviewer_queue_inventory([
|
|
||||||
self._repo_report(open_pr_count=2),
|
|
||||||
self._repo_report(
|
|
||||||
repo="Scaled-Tech-Consulting/mcp-control-plane",
|
|
||||||
open_pr_count=1,
|
|
||||||
),
|
|
||||||
])
|
|
||||||
self.assertTrue(result["complete"])
|
|
||||||
self.assertEqual(result["trust_gates"], {})
|
|
||||||
|
|
||||||
|
|
||||||
class TestCapabilityEvidence(unittest.TestCase):
|
class TestCapabilityEvidence(unittest.TestCase):
|
||||||
"""#179 gap 1: capability claims need exact evidence."""
|
"""#179 gap 1: capability claims need exact evidence."""
|
||||||
|
|
||||||
@@ -1423,13 +1306,6 @@ class TestFinalReport179Bar(unittest.TestCase):
|
|||||||
"controller_handoff": _good_handoff(),
|
"controller_handoff": _good_handoff(),
|
||||||
"capability_proof": _good_capability_proof(),
|
"capability_proof": _good_capability_proof(),
|
||||||
"sweep_proof": _good_secret_sweep(),
|
"sweep_proof": _good_secret_sweep(),
|
||||||
"worktree_proof": {
|
|
||||||
"worktree_path": "/repo/branches/review-feat-issue-224",
|
|
||||||
"porcelain_status": "",
|
|
||||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
|
||||||
"scratch_used": True,
|
|
||||||
"scratch_path": "/repo/branches/review-feat-issue-224",
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
kwargs.update(overrides)
|
kwargs.update(overrides)
|
||||||
return build_final_report(**kwargs)
|
return build_final_report(**kwargs)
|
||||||
|
|||||||
@@ -1,166 +0,0 @@
|
|||||||
"""Tests for reviewer worktree safety proofs (Issue #233)."""
|
|
||||||
import sys
|
|
||||||
import unittest
|
|
||||||
|
|
||||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
||||||
|
|
||||||
from reviewer_worktree import ( # noqa: E402
|
|
||||||
assess_author_worktree_continuity,
|
|
||||||
assess_reviewer_git_command_log,
|
|
||||||
assess_reviewer_worktree_proof,
|
|
||||||
files_outside_pr_scope,
|
|
||||||
is_forbidden_reviewer_git_command,
|
|
||||||
is_readonly_reviewer_git_command,
|
|
||||||
parse_dirty_tracked_files,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class TestParseDirtyTrackedFiles(unittest.TestCase):
|
|
||||||
def test_ignores_untracked_files(self):
|
|
||||||
porcelain = "?? untracked.txt\n M tracked.py\n"
|
|
||||||
self.assertEqual(parse_dirty_tracked_files(porcelain), ["tracked.py"])
|
|
||||||
|
|
||||||
def test_parses_renamed_paths(self):
|
|
||||||
porcelain = "R old.py -> new.py\n"
|
|
||||||
self.assertEqual(parse_dirty_tracked_files(porcelain), ["new.py"])
|
|
||||||
|
|
||||||
|
|
||||||
class TestFilesOutsidePrScope(unittest.TestCase):
|
|
||||||
def test_all_dirty_in_scope_is_clean(self):
|
|
||||||
self.assertEqual(
|
|
||||||
files_outside_pr_scope(
|
|
||||||
["docs/wiki/Repositories.md"],
|
|
||||||
["docs/wiki/Repositories.md"],
|
|
||||||
),
|
|
||||||
[],
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_unrelated_dirty_files_detected(self):
|
|
||||||
self.assertEqual(
|
|
||||||
files_outside_pr_scope(
|
|
||||||
["review_proofs.py", "docs/wiki/Repositories.md"],
|
|
||||||
["docs/wiki/Repositories.md"],
|
|
||||||
),
|
|
||||||
["review_proofs.py"],
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class TestForbiddenGitCommands(unittest.TestCase):
|
|
||||||
def test_blocks_stash_operations(self):
|
|
||||||
self.assertTrue(is_forbidden_reviewer_git_command("git stash"))
|
|
||||||
self.assertTrue(
|
|
||||||
is_forbidden_reviewer_git_command(
|
|
||||||
'git stash push -m "reviewer-temp-stash" -- tests/test_mcp_server.py'
|
|
||||||
)
|
|
||||||
)
|
|
||||||
self.assertTrue(is_forbidden_reviewer_git_command("git stash pop"))
|
|
||||||
self.assertTrue(is_forbidden_reviewer_git_command("git stash drop"))
|
|
||||||
|
|
||||||
def test_blocks_checkout_reset_and_clean(self):
|
|
||||||
self.assertTrue(
|
|
||||||
is_forbidden_reviewer_git_command(
|
|
||||||
"git checkout -- review_proofs.py tests/test_mcp_server.py"
|
|
||||||
)
|
|
||||||
)
|
|
||||||
self.assertTrue(is_forbidden_reviewer_git_command("git reset --hard"))
|
|
||||||
self.assertTrue(is_forbidden_reviewer_git_command("git clean -fd"))
|
|
||||||
|
|
||||||
def test_allows_readonly_commands(self):
|
|
||||||
for cmd in (
|
|
||||||
"git fetch prgs master",
|
|
||||||
"git status --porcelain",
|
|
||||||
"git diff prgs/master...HEAD",
|
|
||||||
"git rev-parse HEAD",
|
|
||||||
"git -C /repo log -1",
|
|
||||||
):
|
|
||||||
with self.subTest(cmd=cmd):
|
|
||||||
self.assertFalse(is_forbidden_reviewer_git_command(cmd))
|
|
||||||
self.assertTrue(is_readonly_reviewer_git_command(cmd))
|
|
||||||
|
|
||||||
|
|
||||||
class TestAssessReviewerWorktreeProof(unittest.TestCase):
|
|
||||||
def test_clean_worktree_proceeds(self):
|
|
||||||
result = assess_reviewer_worktree_proof({
|
|
||||||
"worktree_path": "/repo/branches/review-pr-231",
|
|
||||||
"porcelain_status": "",
|
|
||||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
|
||||||
"scratch_used": False,
|
|
||||||
"git_commands": ["git fetch prgs master", "git diff prgs/master...HEAD"],
|
|
||||||
})
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
self.assertFalse(result["block"])
|
|
||||||
|
|
||||||
def test_dirty_unrelated_without_scratch_blocks(self):
|
|
||||||
result = assess_reviewer_worktree_proof({
|
|
||||||
"worktree_path": "/repo",
|
|
||||||
"dirty_files": ["review_proofs.py", "tests/test_review_proofs.py"],
|
|
||||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
|
||||||
"scratch_used": False,
|
|
||||||
})
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
self.assertTrue(result["block"])
|
|
||||||
self.assertIn("review_proofs.py", result["unrelated_dirty_files"][0])
|
|
||||||
|
|
||||||
def test_scratch_worktree_allows_dirty_main_repo(self):
|
|
||||||
result = assess_reviewer_worktree_proof({
|
|
||||||
"worktree_path": "/repo",
|
|
||||||
"dirty_files": ["review_proofs.py"],
|
|
||||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
|
||||||
"scratch_used": True,
|
|
||||||
"scratch_path": "/repo/branches/review-feat-issue-224",
|
|
||||||
})
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
|
|
||||||
def test_forbidden_command_history_blocks(self):
|
|
||||||
result = assess_reviewer_worktree_proof({
|
|
||||||
"worktree_path": "/repo/branches/review-pr-231",
|
|
||||||
"porcelain_status": "",
|
|
||||||
"git_commands": ["git stash push -m temp -- review_proofs.py"],
|
|
||||||
})
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
self.assertTrue(result["forbidden_commands"])
|
|
||||||
|
|
||||||
def test_unrelated_mutations_claimed_blocks(self):
|
|
||||||
result = assess_reviewer_worktree_proof({
|
|
||||||
"worktree_path": "/repo",
|
|
||||||
"porcelain_status": "",
|
|
||||||
"unrelated_mutations_claimed": True,
|
|
||||||
})
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
|
|
||||||
|
|
||||||
class TestAuthorContinuity(unittest.TestCase):
|
|
||||||
def test_author_may_keep_dirty_worktree(self):
|
|
||||||
result = assess_author_worktree_continuity({
|
|
||||||
"task_role": "author",
|
|
||||||
"dirty_files": ["feat.py"],
|
|
||||||
})
|
|
||||||
self.assertTrue(result["allowed"])
|
|
||||||
|
|
||||||
def test_reviewer_dirty_worktree_uses_reviewer_gate(self):
|
|
||||||
result = assess_author_worktree_continuity({
|
|
||||||
"task_role": "reviewer",
|
|
||||||
"worktree_path": "/repo",
|
|
||||||
"dirty_files": ["other.py"],
|
|
||||||
"pr_scope_files": ["docs/a.md"],
|
|
||||||
"scratch_used": False,
|
|
||||||
})
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
|
|
||||||
|
|
||||||
class TestAssessReviewerGitCommandLog(unittest.TestCase):
|
|
||||||
def test_empty_log_is_clean(self):
|
|
||||||
result = assess_reviewer_git_command_log([])
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
|
|
||||||
def test_mixed_log_blocks_on_forbidden(self):
|
|
||||||
result = assess_reviewer_git_command_log([
|
|
||||||
"git fetch prgs",
|
|
||||||
"git stash",
|
|
||||||
])
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
self.assertEqual(len(result["forbidden_commands"]), 1)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
Reference in New Issue
Block a user