Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
  sysadminandClaude Opus 4.8
|
b36eb201d4 |
@@ -1,20 +0,0 @@
|
||||
## Description
|
||||
|
||||
[Summary of changes and issue number closed.]
|
||||
|
||||
Closes #[Issue Number]
|
||||
|
||||
## Checklist
|
||||
|
||||
- [ ] I have verified my identity matches the required role.
|
||||
- [ ] No secrets, tokens, keychain IDs, or raw service URLs are committed.
|
||||
- [ ] All tests pass for touched code.
|
||||
- [ ] `git diff --check` is clean.
|
||||
|
||||
## Documentation and Wiki
|
||||
|
||||
- [ ] Does this change require a wiki update (workflows, tools, profiles, runbooks)?
|
||||
- [ ] If yes, has `docs/wiki/` been updated accordingly?
|
||||
- [ ] If wiki pages changed, plan the Gitea Wiki sync after merge (`scripts/sync-gitea-wiki.sh`, see Runbooks).
|
||||
- [ ] Readiness gate (#224): the Gitea Wiki is populated and current for this repo — verify the repo **Wiki tab**, not `docs/wiki/`. If stale or empty, record the required sync as a follow-up before approval.
|
||||
- [ ] If this PR closes a wiki-related issue: closure requires live Gitea Wiki proof links (Wiki Home plus page listing or wiki git log). Markdown in `docs/wiki/`, sync-helper code, or policy docs alone are not sufficient to close a wiki issue.
|
||||
@@ -172,14 +172,6 @@ Recognized environment fields (see [`.env.example`](.env.example) for placeholde
|
||||
| `GITEA_MCP_CONFIG` | Optional path to a JSON file defining multiple named runtime profiles. Unset ⇒ pure env behaviour. |
|
||||
| `GITEA_MCP_PROFILE` | Name of the profile (from `GITEA_MCP_CONFIG`) to activate for this runtime. |
|
||||
|
||||
#### External MCP Control Plane servers
|
||||
|
||||
Jenkins and GlitchTip are separate MCP trust boundaries, not tools inside this
|
||||
Gitea MCP runtime. Register them as `jenkins-mcp` and `glitchtip-mcp` in the
|
||||
client that will use them, then reconnect or reload the client and verify the
|
||||
expected tools are visible before claiming readiness. See
|
||||
[`docs/mcp-client-registration.md`](docs/mcp-client-registration.md).
|
||||
|
||||
Notes:
|
||||
|
||||
- This provides **one token + one profile per process**. It does not implement
|
||||
|
||||
@@ -120,11 +120,6 @@ def assess_capability_stop_report(
|
||||
capability_denied: bool = True,
|
||||
) -> dict:
|
||||
"""Validate final report purity after reviewer capability denial."""
|
||||
from review_proofs import (
|
||||
assess_empty_queue_report,
|
||||
parse_trust_gate_status_from_report,
|
||||
)
|
||||
|
||||
text = report_text or ""
|
||||
lower = text.lower()
|
||||
violations = []
|
||||
@@ -153,19 +148,13 @@ def assess_capability_stop_report(
|
||||
r"inventory empty",
|
||||
re.I,
|
||||
)
|
||||
parsed_status = parse_trust_gate_status_from_report(text)
|
||||
effective_status = trust_gate_status or parsed_status
|
||||
if empty_queue_patterns.search(text):
|
||||
if effective_status != "trusted_empty":
|
||||
if trust_gate_status != "trusted_empty":
|
||||
violations.append(
|
||||
"empty-queue claim after capability stop without "
|
||||
"pr_inventory_trust_gate.status == trusted_empty"
|
||||
)
|
||||
|
||||
empty_queue = assess_empty_queue_report(text)
|
||||
if empty_queue.get("claimed") and not empty_queue.get("proven"):
|
||||
violations.extend(empty_queue.get("reasons") or [])
|
||||
|
||||
ok, elig_violations = validate_eligibility_wording(text)
|
||||
violations.extend(elig_violations)
|
||||
|
||||
|
||||
@@ -22,13 +22,13 @@ Strictly read-only, per ADR-0001:
|
||||
tools — never one dual-credential server).
|
||||
- **This server never holds Gitea write credentials.**
|
||||
|
||||
## 2. Boundary placement
|
||||
## 2. Boundary placement (namespace pending)
|
||||
|
||||
These tools belong to the GlitchTip observability boundary of the MCP Control
|
||||
Plane. The canonical MCP server name is `glitchtip-mcp`; client registration
|
||||
and reload instructions live in
|
||||
[`../mcp-client-registration.md`](../mcp-client-registration.md). Tool names
|
||||
below use the `glitchtip_` prefix.
|
||||
Plane — `glitchtip-mcp` (ADR-0001's recommendation), `observability-mcp`, or
|
||||
folded into `ops-mcp`. **ADR-0001 open owner decision #2 picks the name; this
|
||||
design does not assume it.** Tool names below use the `glitchtip_` prefix for
|
||||
readability and rename mechanically with the decision.
|
||||
|
||||
Fixed regardless of the name (per `tool-boundaries.md`,
|
||||
`credential-isolation.md`):
|
||||
@@ -160,13 +160,11 @@ Mocked-GlitchTip unit tests only, per `docs/developer-testing-guidelines.md`:
|
||||
|
||||
## 10. Implementation-readiness checklist
|
||||
|
||||
Ready to operate once:
|
||||
Ready to implement once:
|
||||
|
||||
1. The MCP client registers the external server under the exact `glitchtip-mcp`
|
||||
name and is reconnected/reloaded.
|
||||
2. Tool discoverability proves the expected `glitchtip_*` read tools are
|
||||
visible; if the server is enabled but exposes no usable tools, report
|
||||
`SKIPPED` and stop.
|
||||
1. ADR-0001 owner decision #2 (namespace/placement) is made — mechanical
|
||||
rename of the `glitchtip_` prefix if needed.
|
||||
2. ADR-0001 owner decision #1 (repo home) is made.
|
||||
3. #76 profile schema exists (or a minimal `glitchtip-readonly` profile is
|
||||
hand-rolled to the same rules).
|
||||
4. A pinned GlitchTip version is chosen for API-subset testing (§3).
|
||||
|
||||
@@ -6,8 +6,6 @@
|
||||
- **Date:** 2026-07-02
|
||||
|
||||
Note on naming: This design used historical `jenkins-readonly` skill name in Gitea-Tools. Actual package/server is `jenkins-mcp` (see mcp-control-plane registration in #55). The server boundary now contains gated trigger (see #56), but read tools remain as designed.
|
||||
Client registration and reload instructions live in
|
||||
[`../mcp-client-registration.md`](../mcp-client-registration.md).
|
||||
|
||||
## 1. Purpose and scope
|
||||
|
||||
@@ -143,16 +141,12 @@ repo's conventions (`docs/developer-testing-guidelines.md`):
|
||||
|
||||
## 10. Implementation-readiness checklist
|
||||
|
||||
Ready to operate through `jenkins-mcp` once:
|
||||
Ready to implement in `jenkins-mcp` once:
|
||||
|
||||
1. The MCP client registers the external server under the exact `jenkins-mcp`
|
||||
name and is reconnected/reloaded.
|
||||
2. Tool discoverability proves the expected `jenkins_*` read tools are visible;
|
||||
if the server is enabled but exposes no usable tools, report `SKIPPED` and
|
||||
stop.
|
||||
3. #76 profile schema exists (or a minimal `jenkins-readonly` profile is
|
||||
1. ADR-0001 owner decision #1 (where `jenkins-mcp` lives) is made.
|
||||
2. #76 profile schema exists (or a minimal `jenkins-readonly` profile is
|
||||
hand-rolled to the same rules).
|
||||
4. #77 mapping design is accepted (or tools ship path-addressed only, mapping
|
||||
3. #77 mapping design is accepted (or tools ship path-addressed only, mapping
|
||||
deferred).
|
||||
|
||||
Explicitly **not** unlocked by this document: build triggers, deploys,
|
||||
|
||||
@@ -298,17 +298,6 @@ When dynamic profile switching is enabled and a profile is activated via `gitea_
|
||||
2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
|
||||
This guarantees the active profile operations align with the actual Gitea authenticated user credential.
|
||||
|
||||
## Gitea MCP Runtime Isolation and Worktree Safety
|
||||
|
||||
To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:
|
||||
|
||||
- **Startup Conflict Check:** The MCP server (`mcp_server.py`) acts as a conflict-free loader. On startup, it scans all Python files in the directory for unresolved git merge conflicts (`<<<<<<<`, `=======`, `>>>>>>>`). If any are found, it prints an `infra_stop` message and exits immediately.
|
||||
- **Workflow Guard:** Before starting reviewer work, task routing checks if the MCP runtime source is mid-merge (by checking for `.git/MERGE_HEAD` or conflict markers). If dirty, it returns `infra_stop` (never `wrong_role_stop` or empty queue) to prevent unsafe mutations.
|
||||
- **Recovery Instructions:** To recover from an `infra_stop` state:
|
||||
1. Resolve all merge conflicts in the local repository or abort the merge (`git merge --abort` / `git rebase --abort`).
|
||||
2. Restart the Gitea MCP server process.
|
||||
3. Retry the task.
|
||||
|
||||
## Relationship to roadmap issues
|
||||
|
||||
This document defines the **model only**. Related work is tracked separately
|
||||
|
||||
@@ -30,11 +30,6 @@ audit logging). See [Related documents](#related-documents).
|
||||
> the standard rules; operator prompts still control task-specific scope.
|
||||
> See issue #129 for the skill registry design.
|
||||
|
||||
Jenkins and GlitchTip workflows use separate MCP servers, not this Gitea MCP
|
||||
runtime. Register them as `jenkins-mcp` and `glitchtip-mcp`, reconnect or
|
||||
reload the client, and verify visible tools before claiming either integration
|
||||
is usable. See [`mcp-client-registration.md`](mcp-client-registration.md).
|
||||
|
||||
For cross-project use, copy the portable workflow skill at
|
||||
[`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
|
||||
It extracts the issue-first, isolated-worktree, no-self-review, profile-safety,
|
||||
|
||||
@@ -1,87 +0,0 @@
|
||||
# MCP Client Registration for External Control Plane Servers
|
||||
|
||||
Issue #151 fixes the registration and naming contract for the Jenkins and
|
||||
GlitchTip MCP servers that live outside this Gitea MCP runtime.
|
||||
|
||||
## Canonical Server Names
|
||||
|
||||
Use these exact MCP server names in clients:
|
||||
|
||||
| Server name | Boundary | Default capability |
|
||||
|---|---|---|
|
||||
| `jenkins-mcp` | Jenkins CI inspection | Read-only build/job inspection |
|
||||
| `glitchtip-mcp` | GlitchTip observability inspection | Read-only issue/event inspection |
|
||||
|
||||
Historical names such as `jenkins-readonly` and `glitchtip-readonly` are
|
||||
descriptive profile labels only. They are not the canonical MCP server names
|
||||
unless an operator intentionally creates aliases and documents them.
|
||||
|
||||
## Registration Pattern
|
||||
|
||||
Register each external server as its own MCP entry. Do not add Jenkins or
|
||||
GlitchTip credentials to the Gitea MCP server. Also, do not add Gitea write
|
||||
credentials to the GlitchTip server.
|
||||
|
||||
```json
|
||||
{
|
||||
"mcpServers": {
|
||||
"jenkins-mcp": {
|
||||
"command": "/path/to/mcp-control-plane/venv/bin/python3",
|
||||
"args": ["-m", "jenkins_mcp"],
|
||||
"env": {
|
||||
"JENKINS_MCP_CONFIG": "/path/to/mcp-control-plane/profiles.json",
|
||||
"JENKINS_MCP_PROFILE": "jenkins-readonly"
|
||||
}
|
||||
},
|
||||
"glitchtip-mcp": {
|
||||
"command": "/path/to/mcp-control-plane/venv/bin/python3",
|
||||
"args": ["-m", "glitchtip_mcp"],
|
||||
"env": {
|
||||
"GLITCHTIP_MCP_CONFIG": "/path/to/mcp-control-plane/profiles.json",
|
||||
"GLITCHTIP_MCP_PROFILE": "glitchtip-readonly"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Client-specific wrappers may differ, but the server names, trust boundaries,
|
||||
and profile separation must remain the same. After adding or changing either
|
||||
entry, reconnect or reload the MCP client before claiming the tools are usable.
|
||||
|
||||
## Discoverability Validation
|
||||
|
||||
Before using either server in a task, prove the expected tools are visible in
|
||||
the client. It is not enough for the config entry to exist.
|
||||
|
||||
Expected Jenkins tools:
|
||||
|
||||
- `jenkins_whoami`
|
||||
- `jenkins_list_jobs`
|
||||
- `jenkins_latest_build`
|
||||
- `jenkins_build_status`
|
||||
- `jenkins_get_build`
|
||||
|
||||
Expected GlitchTip tools:
|
||||
|
||||
- `glitchtip_whoami`
|
||||
- `glitchtip_list_projects`
|
||||
- `glitchtip_list_unresolved`
|
||||
- `glitchtip_get_issue`
|
||||
- `glitchtip_recent_events`
|
||||
- `glitchtip_search`
|
||||
|
||||
If a client reports the server as enabled but exposes no usable tools, report
|
||||
`SKIPPED: server enabled but no usable tools visible`, then stop. Do not fall
|
||||
back to shell commands, raw service APIs, or unrelated MCP servers.
|
||||
|
||||
## Boundary Rules
|
||||
|
||||
- `jenkins-mcp` read profiles must not expose build trigger tools.
|
||||
- Jenkins build triggers require a separately named write profile, exact
|
||||
confirmation, and fail-closed mutation audit.
|
||||
- `glitchtip-mcp` remains read-only. It must not file or mutate Gitea issues.
|
||||
- GlitchTip-to-Gitea filing is a separate orchestrator that composes GlitchTip
|
||||
read tools with Gitea issue-write tools.
|
||||
- Gitea credentials never enter Jenkins or GlitchTip runtimes.
|
||||
- Jenkins and GlitchTip credentials never enter the Gitea MCP runtime.
|
||||
@@ -1,15 +0,0 @@
|
||||
# Project History
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
## 2026-07-06
|
||||
|
||||
- **Wiki bootstrap (#224)** — Added repo-tracked `docs/wiki/` (10 pages), `scripts/sync-gitea-wiki.sh`, PR template gate, and sync safety tests for Gitea-Tools publication.
|
||||
|
||||
## Prior milestones
|
||||
|
||||
- Gated review/merge path (#16), task capability resolver (#69), issue-write tool gates.
|
||||
- Canonical JSON execution profiles (#19) and thin MCP launchers.
|
||||
- Role session router (#206) and review decision lock (#211).
|
||||
@@ -1,35 +0,0 @@
|
||||
# Gitea-Tools Project Wiki
|
||||
|
||||
Welcome to the version-controlled wiki for the Gitea-Tools MCP server and CLI tooling.
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md)
|
||||
- [Operator Guide](Operator-Guide.md)
|
||||
- [Repositories Map](Repositories.md)
|
||||
- [Identity and Profiles](Identity-and-Profiles.md)
|
||||
- [Workflow Model](Workflow.md)
|
||||
- [Safety and Gates](Safety-and-Gates.md)
|
||||
- [MCP Tools Reference](MCP-Tools.md)
|
||||
- [Operator Runbooks](Runbooks.md)
|
||||
- [Open Decisions](Open-Decisions.md)
|
||||
- [Project History](History.md)
|
||||
|
||||
## Gitea Wiki mirror
|
||||
|
||||
These repo-tracked pages are the **source of truth**. The Gitea native Wiki
|
||||
for this repository is a read-only convenience mirror generated from this
|
||||
directory with `scripts/sync-gitea-wiki.sh` (dry-run by default; see
|
||||
[Runbooks](Runbooks.md)). Never edit the Gitea Wiki directly — change the
|
||||
pages here through a PR, then sync.
|
||||
|
||||
## Project overview
|
||||
|
||||
Gitea-Tools provides the Gitea MCP server, execution-profile configuration,
|
||||
identity handling, and CLI helpers used across MCP Control Plane workflows.
|
||||
It enforces author/reviewer separation, gated review/merge, task-capability
|
||||
resolution, and fail-closed safety rails in code.
|
||||
|
||||
Canonical long-form docs also live under `docs/` in the repository (for example
|
||||
`docs/gitea-execution-profiles.md` and `docs/llm-workflow-runbooks.md`). The
|
||||
wiki summarizes operator-facing rules; the repo docs carry implementation detail.
|
||||
@@ -1,39 +0,0 @@
|
||||
# Identity and Profiles
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
The LLM is not the role — the **MCP execution profile** is the role. Profiles
|
||||
bind an authenticated Gitea identity to an allowed operation set.
|
||||
|
||||
## Reference profiles (prgs)
|
||||
|
||||
### Author / implementer
|
||||
|
||||
- **Profile:** `prgs-author`
|
||||
- **Typical identity:** `jcwalker3`
|
||||
- **Allowed:** branch create/push, PR create, issue comment/create/close, repo commit, read.
|
||||
- **Forbidden:** PR approve, merge, request_changes.
|
||||
|
||||
### Reviewer / merger
|
||||
|
||||
- **Profile:** `prgs-reviewer`
|
||||
- **Typical identity:** `sysadmin`
|
||||
- **Allowed:** PR review/approve/merge/request_changes, issue comment, read.
|
||||
- **Forbidden:** branch push, PR create, repo commit.
|
||||
|
||||
## Configuration
|
||||
|
||||
Profiles are defined in the canonical JSON config (`GITEA_MCP_CONFIG`, typically
|
||||
`~/.config/gitea-tools/profiles.json`). Launchers are thin: they set
|
||||
`GITEA_MCP_PROFILE` and point at the config file. Credentials resolve from
|
||||
keychain or env references — never inline in client configs.
|
||||
|
||||
See `docs/gitea-execution-profiles.md` in the repository for the full model.
|
||||
|
||||
## Profile switching
|
||||
|
||||
Use separate MCP server namespaces (`gitea-tools` author vs `gitea-reviewer`)
|
||||
or distinct launcher entries. Runtime in-place profile switching is disabled by
|
||||
default (fail closed).
|
||||
@@ -1,34 +0,0 @@
|
||||
# MCP Tools Reference
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
## Identity and capability
|
||||
|
||||
- `gitea_whoami` — authenticated user and active profile metadata.
|
||||
- `gitea_get_profile` / `gitea_get_runtime_context` — allowed and forbidden operations.
|
||||
- `gitea_resolve_task_capability` — required pre-flight for gated mutations.
|
||||
- `gitea_route_task_session` — role/session router before task execution.
|
||||
|
||||
## Author tools
|
||||
|
||||
- `gitea_create_issue`, `gitea_create_issue_comment`, `gitea_close_issue`
|
||||
- `gitea_mark_issue`, `gitea_set_issue_labels`, `gitea_lock_issue`
|
||||
- `gitea_create_pr`, `gitea_edit_pr`, `gitea_commit_files`
|
||||
- `gitea_delete_branch`
|
||||
|
||||
## Reviewer tools
|
||||
|
||||
- `gitea_check_pr_eligibility` — read-only eligibility check.
|
||||
- `gitea_dry_run_pr_review` — validation-phase review mechanics.
|
||||
- `gitea_mark_final_review_decision` — mark validation complete.
|
||||
- `gitea_submit_pr_review` / `gitea_review_pr` — gated live review.
|
||||
- `gitea_merge_pr` — gated merge (only merge path).
|
||||
|
||||
## Read tools
|
||||
|
||||
- `gitea_list_prs`, `gitea_view_pr`, `gitea_list_issues`, `gitea_view_issue`
|
||||
- `gitea_get_file`, `gitea_list_labels`, `gitea_mirror_refs`
|
||||
|
||||
See the repository `README.md` for the full tool table and client setup.
|
||||
@@ -1,11 +0,0 @@
|
||||
# Open Decisions
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
Pending architectural and workflow decisions:
|
||||
|
||||
- **Operation-scoped role selection (#228)** — Move from launcher-profile-dependent routing to per-operation profile resolution.
|
||||
- **Gitea-Tools wiki for dadeschools remote** — This bootstrap covers `prgs`; dadeschools instance wiki parity is undecided.
|
||||
- **Server-side self-merge block** — Complement tool gates with Gitea branch protection where available.
|
||||
@@ -1,31 +0,0 @@
|
||||
# Operator Guide
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
Handbook for LLM operators and human developers using the Gitea-Tools MCP server.
|
||||
|
||||
## Key rules
|
||||
|
||||
1. **Verify identity first** — Run `gitea_whoami` and confirm the active profile before any mutation.
|
||||
2. **Resolve task capability** — Call `gitea_resolve_task_capability` for the intended task before gated tools.
|
||||
3. **One unit of work per session** — Implement one claimed issue *or* review/merge one PR; do not mix author and reviewer mutations in one session.
|
||||
4. **No self-review / no self-merge** — The authenticated Gitea user must not approve or merge a PR they authored.
|
||||
5. **Follow the gates** — Prompts express intent; MCP tools enforce safety. Never bypass gates via prompt instructions.
|
||||
|
||||
## Supported Gitea instances
|
||||
|
||||
| Remote | Host | Default org/repo |
|
||||
|--------|------|------------------|
|
||||
| `dadeschools` | `gitea.dadeschools.net` | `Contractor / Timesheet` |
|
||||
| `prgs` | `gitea.prgs.cc` | `Scaled-Tech-Consulting / Timesheet` |
|
||||
|
||||
Always pass `remote` explicitly on tool calls. The server default is `dadeschools`; forgetting `remote` on a `prgs` task hits the wrong host.
|
||||
|
||||
## New session checklist
|
||||
|
||||
1. `gitea_whoami` — confirm authenticated user and profile.
|
||||
2. `gitea_get_runtime_context` — allowed/forbidden operations for this session.
|
||||
3. `gitea_resolve_task_capability` — prove the session may perform the planned task.
|
||||
4. For reviewer work: dry-run validation (`gitea_dry_run_pr_review`) before live review mutations.
|
||||
@@ -1,38 +0,0 @@
|
||||
# Repositories Map
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
Active MCP Control Plane repositories (authoritative inventory for wiki publication gate #224 / #87):
|
||||
|
||||
| Repository | Purpose | Wiki required |
|
||||
|------------|---------|---------------|
|
||||
| `Scaled-Tech-Consulting/Gitea-Tools` | Gitea MCP server, execution profiles, workflow tooling | Yes — live Gitea Wiki on Wiki tab |
|
||||
| `Scaled-Tech-Consulting/mcp-control-plane` | Orchestrators, audit trails, multi-service controller workflows | Yes — live Gitea Wiki on Wiki tab |
|
||||
|
||||
Repo-tracked `docs/wiki/` is the source of truth; the Gitea Wiki is a mirror.
|
||||
Wiki-related issues cannot be closed until the live Wiki is verified — see
|
||||
[Safety and Gates](Safety-and-Gates.md) and [Runbooks](Runbooks.md).
|
||||
|
||||
### Gitea-Tools
|
||||
|
||||
- **Repository:** `Scaled-Tech-Consulting/Gitea-Tools`
|
||||
- **Purpose:** Gitea MCP server, profile configuration, CLI scripts, safety gates.
|
||||
- **Base branch:** `master`
|
||||
|
||||
### mcp-control-plane
|
||||
|
||||
- **Repository:** `Scaled-Tech-Consulting/mcp-control-plane`
|
||||
- **Purpose:** Controller workflows, Jenkins/GlitchTip integrations, audit orchestration.
|
||||
- **Base branch:** `master`
|
||||
|
||||
## Wiki publication status (#224 readiness gate)
|
||||
|
||||
| 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/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` |
|
||||
|
||||
|
||||
Update this table whenever a wiki is published, re-synced, or found stale.
|
||||
@@ -1,40 +0,0 @@
|
||||
# Operator Runbooks
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
## PR review and merge
|
||||
|
||||
1. `gitea_resolve_task_capability(task="review_pr")`
|
||||
2. `gitea_check_pr_eligibility` for review and merge actions.
|
||||
3. Validate locally: tests, `py_compile`, `git diff --check`.
|
||||
4. `gitea_mark_final_review_decision` → approve via `gitea_review_pr`.
|
||||
5. `gitea_merge_pr` with pinned head SHA and `confirmation="MERGE PR <n>"`.
|
||||
|
||||
## Gitea Wiki sync
|
||||
|
||||
The Gitea Wiki mirrors `docs/wiki/` (source of truth). After merging wiki changes:
|
||||
|
||||
1. Preview (no network):
|
||||
```bash
|
||||
scripts/sync-gitea-wiki.sh
|
||||
```
|
||||
2. Operator-confirmed push:
|
||||
```bash
|
||||
GITEA_WIKI_SYNC_CONFIRM="SYNC WIKI Gitea-Tools" scripts/sync-gitea-wiki.sh --push
|
||||
```
|
||||
3. If clone fails on first run, bootstrap Home via Gitea API or UI, then re-run.
|
||||
|
||||
The script mirrors only `docs/wiki/*.md`, never deletes wiki pages, and never
|
||||
prints credentials.
|
||||
|
||||
## Wiki Publication Readiness Gate (#224)
|
||||
|
||||
Actual Gitea Wiki publication is a **required repo readiness gate**.
|
||||
|
||||
1. **Closure prevention** — No wiki issue may close on markdown/helper work alone.
|
||||
Closing requires live-Wiki proof: Wiki Home link plus page listing or wiki git log.
|
||||
2. **Reviewer checklist** — PR template requires verifying the repo **Wiki tab**.
|
||||
3. **Publication authority** — `--push` requires exact `GITEA_WIKI_SYNC_CONFIRM` phrase.
|
||||
4. **Per-repo status** — Update [Repositories Map](Repositories.md) when published.
|
||||
@@ -1,19 +0,0 @@
|
||||
# Safety and Gates
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
*Prompts express intent; MCP tools enforce safety.*
|
||||
|
||||
## Primary gates
|
||||
|
||||
1. **Fail closed** — Unknown tasks, missing capability resolution, or profile mismatches block mutations.
|
||||
2. **Task capability map** — `gitea_resolve_task_capability` must precede gated issue/PR mutations.
|
||||
3. **Issue lock** — `gitea_lock_issue` required before author implementation on a tracked issue.
|
||||
4. **Head SHA pinning** — Reviews and merges refuse when the PR head moved.
|
||||
5. **Explicit merge confirmation** — `gitea_merge_pr` requires `confirmation="MERGE PR <n>"`.
|
||||
6. **No self-review / self-merge** — Authenticated user must differ from PR author for approve/merge.
|
||||
7. **Review decision lock** — Live review mutations require validation-phase dry-run and `gitea_mark_final_review_decision`.
|
||||
8. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
|
||||
9. **Wiki publication (#224)** — `docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).
|
||||
@@ -1,28 +0,0 @@
|
||||
# Workflow Model
|
||||
|
||||
## Navigation
|
||||
|
||||
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
|
||||
|
||||
## Step 1: Review queue first (reviewer profile)
|
||||
|
||||
1. `gitea_resolve_task_capability(task="review_pr")`
|
||||
2. List open PRs; pick the oldest eligible PR (not self-authored, mergeable).
|
||||
3. Pin head SHA; validate diff scope and run tests locally.
|
||||
4. `gitea_mark_final_review_decision` → `gitea_review_pr` / `gitea_submit_pr_review`
|
||||
5. `gitea_merge_pr` only with `confirmation="MERGE PR <n>"` and pinned head SHA.
|
||||
|
||||
## Step 2: Implement issues (author profile)
|
||||
|
||||
1. `gitea_resolve_task_capability` for the author task.
|
||||
2. `gitea_lock_issue` before implementation mutations.
|
||||
3. Claim with `gitea_mark_issue` / `status:in-progress` label.
|
||||
4. Branch, implement, test, push, `gitea_create_pr`.
|
||||
5. Release claim when done; never self-review the PR.
|
||||
|
||||
## Step 3: Operator follow-ups
|
||||
|
||||
Wiki publication, credential provisioning, and cross-repo mirror operations are
|
||||
operator-confirmed actions — see [Runbooks](Runbooks.md).
|
||||
|
||||
Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.
|
||||
+2
-10
@@ -163,13 +163,12 @@ def audit_enabled():
|
||||
def build_event(*, action, result, remote=None, server=None, repository=None,
|
||||
issue_number=None, pr_number=None, profile_name=None,
|
||||
audit_label=None, authenticated_username=None, target_branch=None,
|
||||
head_sha=None, reason=None, request_metadata=None, now=None,
|
||||
mcp_namespace=None, task_role=None, operation=None):
|
||||
head_sha=None, reason=None, request_metadata=None, now=None):
|
||||
"""Build a redacted, JSON-able audit record for a mutating action."""
|
||||
ts = now or datetime.datetime.now(datetime.timezone.utc)
|
||||
if isinstance(ts, datetime.datetime):
|
||||
ts = ts.isoformat()
|
||||
event = {
|
||||
return {
|
||||
"timestamp": ts,
|
||||
"action": action,
|
||||
"action_type": "mutating",
|
||||
@@ -187,13 +186,6 @@ def build_event(*, action, result, remote=None, server=None, repository=None,
|
||||
"reason": _redact_str(reason) if reason else reason,
|
||||
"request_metadata": redact(request_metadata) if request_metadata is not None else None,
|
||||
}
|
||||
if mcp_namespace is not None:
|
||||
event["mcp_namespace"] = mcp_namespace
|
||||
if task_role is not None:
|
||||
event["task_role"] = task_role
|
||||
if operation is not None:
|
||||
event["operation"] = operation
|
||||
return event
|
||||
|
||||
|
||||
def write_event(event, path=None):
|
||||
|
||||
-4572
File diff suppressed because it is too large
Load Diff
+4328
-48
File diff suppressed because it is too large
Load Diff
+11
-870
File diff suppressed because it is too large
Load Diff
@@ -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": []}
|
||||
@@ -1,89 +0,0 @@
|
||||
"""Block author mutations from reviewer-bound MCP namespaces (#209).
|
||||
|
||||
Every mutation must prove that the active profile role, inferred MCP
|
||||
namespace, and declared task role align. Reviewer sessions cannot silently
|
||||
perform author-side work (especially PR creation).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import gitea_config
|
||||
import role_session_router
|
||||
|
||||
|
||||
def infer_mcp_namespace(profile_name: str | None) -> str:
|
||||
"""Map a runtime profile name to its MCP namespace label."""
|
||||
lower = (profile_name or "").strip().lower()
|
||||
if "reviewer" in lower and "author" not in lower:
|
||||
return "gitea-reviewer"
|
||||
if "author" in lower:
|
||||
return "gitea-author"
|
||||
return profile_name or "gitea-default"
|
||||
|
||||
|
||||
def derive_role_kind(allowed, forbidden=()) -> str:
|
||||
"""Classify the active profile the same way as ``mcp_server._role_kind``."""
|
||||
|
||||
def can(op):
|
||||
return gitea_config.check_operation(op, allowed, forbidden)[0]
|
||||
|
||||
review = can("gitea.pr.approve") or can("gitea.pr.merge")
|
||||
author = can("gitea.pr.create") or can("gitea.branch.push")
|
||||
if review and author:
|
||||
return "mixed"
|
||||
if review:
|
||||
return "reviewer"
|
||||
if author:
|
||||
return "author"
|
||||
return "limited"
|
||||
|
||||
|
||||
def check_author_mutation_namespace(
|
||||
mutation_task: str,
|
||||
profile: dict,
|
||||
) -> tuple[bool, list[str]]:
|
||||
"""Return (allowed, reasons). Fail closed on reviewer/author mismatch."""
|
||||
required_role = role_session_router.required_role_for_task(mutation_task)
|
||||
if required_role != "author":
|
||||
return True, []
|
||||
|
||||
allowed = profile.get("allowed_operations") or []
|
||||
forbidden = profile.get("forbidden_operations") or []
|
||||
active_role = derive_role_kind(allowed, forbidden)
|
||||
profile_name = profile.get("profile_name") or ""
|
||||
namespace = infer_mcp_namespace(profile_name)
|
||||
|
||||
if mutation_task == "create_pr":
|
||||
if active_role == "reviewer" or namespace == "gitea-reviewer":
|
||||
return False, [
|
||||
"author mutation 'create_pr' blocked in reviewer MCP namespace "
|
||||
f"({namespace}); launch gitea-author",
|
||||
]
|
||||
|
||||
if active_role == "reviewer" or namespace == "gitea-reviewer":
|
||||
if mutation_task == "create_issue":
|
||||
ok, _ = gitea_config.check_operation(
|
||||
"gitea.issue.create", allowed, forbidden)
|
||||
if ok:
|
||||
return True, []
|
||||
return False, [
|
||||
f"author mutation '{mutation_task}' blocked: active session is "
|
||||
f"reviewer-bound ({profile_name} / {namespace})",
|
||||
]
|
||||
|
||||
return True, []
|
||||
|
||||
|
||||
def mutation_audit_context(mutation_task: str, profile: dict, *,
|
||||
remote=None, repository=None) -> dict:
|
||||
"""Structured mutation metadata for audit records (#209)."""
|
||||
allowed = profile.get("allowed_operations") or []
|
||||
forbidden = profile.get("forbidden_operations") or []
|
||||
return {
|
||||
"mcp_namespace": infer_mcp_namespace(profile.get("profile_name")),
|
||||
"profile_name": profile.get("profile_name"),
|
||||
"task_role": role_session_router.required_role_for_task(mutation_task),
|
||||
"operation": mutation_task,
|
||||
"remote": remote,
|
||||
"repository": repository,
|
||||
}
|
||||
+1
-53
@@ -5,14 +5,12 @@ returns a route result before any downstream mutation tools run.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
import os
|
||||
|
||||
ROUTE_ALLOWED = "allowed_current_session"
|
||||
ROUTE_WRONG_ROLE = "wrong_role_stop"
|
||||
ROUTE_TO_AUTHOR = "route_to_author_session"
|
||||
ROUTE_TO_REVIEWER = "route_to_reviewer_session"
|
||||
ROUTE_AMBIGUOUS = "ambiguous_task_stop"
|
||||
ROUTE_INFRA_STOP = "infra_stop"
|
||||
|
||||
REVIEWER_TASKS = frozenset({
|
||||
"review_pr",
|
||||
@@ -95,23 +93,6 @@ def route_task_session(
|
||||
_record_route(result)
|
||||
return result
|
||||
|
||||
if required_role == "reviewer" and check_mid_merge():
|
||||
result = {
|
||||
"task_type": task_type,
|
||||
"required_role": required_role,
|
||||
"active_role": active_role_kind,
|
||||
"active_profile": active_profile,
|
||||
"route_result": ROUTE_INFRA_STOP,
|
||||
"downstream_allowed": False,
|
||||
"reasons": [
|
||||
"infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
|
||||
"Please resolve all conflicts manually, finish/abort the merge, restart the MCP server, and retry."
|
||||
],
|
||||
"message": "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
|
||||
}
|
||||
_record_route(result)
|
||||
return result
|
||||
|
||||
if allowed_in_current_session:
|
||||
result = {
|
||||
"task_type": task_type,
|
||||
@@ -211,37 +192,4 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool,
|
||||
"task and relaunches an author MCP session.",
|
||||
f"Attempted fallback mutation: {mutation_task}",
|
||||
]
|
||||
return True, []
|
||||
|
||||
|
||||
def check_mid_merge() -> bool:
|
||||
"""Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
|
||||
project_root = os.path.dirname(os.path.abspath(__file__))
|
||||
git_dir = os.path.join(project_root, ".git")
|
||||
if os.path.exists(git_dir):
|
||||
if (os.path.exists(os.path.join(git_dir, "MERGE_HEAD"))
|
||||
or os.path.exists(os.path.join(git_dir, "rebase-merge"))
|
||||
or os.path.exists(os.path.join(git_dir, "rebase-apply"))):
|
||||
return True
|
||||
|
||||
# Scan python files for conflict markers
|
||||
conflict_patterns = [
|
||||
b"<" * 7 + b" ",
|
||||
b"=" * 7 + b"\n",
|
||||
b"=" * 7 + b"\r\n",
|
||||
b">" * 7 + b" "
|
||||
]
|
||||
for root, dirs, files in os.walk(project_root):
|
||||
if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
|
||||
continue
|
||||
for file in files:
|
||||
if file.endswith(".py"):
|
||||
file_path = os.path.join(root, file)
|
||||
try:
|
||||
with open(file_path, "rb") as f:
|
||||
content = f.read()
|
||||
if any(pattern in content for pattern in conflict_patterns):
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
return True, []
|
||||
@@ -1,86 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
# sync-gitea-wiki.sh — mirror the repo-tracked docs/wiki/ pages into the
|
||||
# Gitea native wiki. docs/wiki/ remains the source of truth; the Gitea Wiki
|
||||
# is a read convenience mirrored FROM it, never edited directly.
|
||||
#
|
||||
# scripts/sync-gitea-wiki.sh dry-run (default): print the plan
|
||||
# scripts/sync-gitea-wiki.sh --push actually sync, ONLY with the exact
|
||||
# confirmation below
|
||||
#
|
||||
# Safety contract:
|
||||
# - Dry-run is the default and performs no network or repository-mutating
|
||||
# operation (only a local read of the origin remote URL).
|
||||
# - A push requires GITEA_WIKI_SYNC_CONFIRM to equal exactly
|
||||
# "SYNC WIKI <repo-name>" (repo name derived from the origin remote).
|
||||
# Anything else refuses before any clone happens.
|
||||
# - The wiki remote is derived from the local origin remote; nothing is
|
||||
# hardcoded here and no credential material is read, printed, or stored
|
||||
# by this script — git's own configured auth is used as-is.
|
||||
# - Only *.md pages from docs/wiki/ are mirrored; nothing else is touched
|
||||
# and no page is deleted from the wiki by this script.
|
||||
|
||||
here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
repo_root="$(cd "$here/.." && pwd)"
|
||||
wiki_src="$repo_root/docs/wiki"
|
||||
|
||||
[ -d "$wiki_src" ] || { echo "error: $wiki_src not found" >&2; exit 1; }
|
||||
|
||||
if git -C "$repo_root" remote get-url origin >/dev/null 2>&1; then
|
||||
origin_url="$(git -C "$repo_root" remote get-url origin)"
|
||||
elif git -C "$repo_root" remote get-url prgs >/dev/null 2>&1; then
|
||||
origin_url="$(git -C "$repo_root" remote get-url prgs)"
|
||||
else
|
||||
echo "error: configure an origin or prgs git remote" >&2
|
||||
exit 1
|
||||
fi
|
||||
repo_name="$(basename "$origin_url" .git)"
|
||||
wiki_remote="${origin_url%.git}.wiki.git"
|
||||
expected_confirm="SYNC WIKI $repo_name"
|
||||
|
||||
pages=()
|
||||
while IFS= read -r -d '' f; do
|
||||
pages+=("$(basename "$f")")
|
||||
done < <(find "$wiki_src" -maxdepth 1 -name '*.md' -print0 | sort -z)
|
||||
|
||||
mode="${1:-}"
|
||||
|
||||
if [ "$mode" != "--push" ]; then
|
||||
echo "dry-run: would sync ${#pages[@]} pages from docs/wiki/ to the '$repo_name' Gitea Wiki:"
|
||||
for p in "${pages[@]}"; do
|
||||
echo " $p"
|
||||
done
|
||||
echo "dry-run: no git or network operation performed."
|
||||
echo "To sync for real: GITEA_WIKI_SYNC_CONFIRM=\"$expected_confirm\" $0 --push"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [ "${GITEA_WIKI_SYNC_CONFIRM:-}" != "$expected_confirm" ]; then
|
||||
echo "refused: --push requires GITEA_WIKI_SYNC_CONFIRM to equal exactly:" >&2
|
||||
echo " $expected_confirm" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
workdir="$(mktemp -d)"
|
||||
trap 'rm -rf "$workdir"' EXIT
|
||||
|
||||
echo "cloning wiki repository for '$repo_name'"
|
||||
if ! git clone --quiet -- "$wiki_remote" "$workdir/wiki" 2>/dev/null; then
|
||||
echo "error: could not clone the wiki repository. If this repo's wiki has" >&2
|
||||
echo "never been initialized, create its first page once in the Gitea UI" >&2
|
||||
echo "(Wiki tab -> New Page), then re-run this script." >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
cp "$wiki_src"/*.md "$workdir/wiki/"
|
||||
|
||||
if git -C "$workdir/wiki" status --porcelain | grep -q .; then
|
||||
git -C "$workdir/wiki" add -A
|
||||
git -C "$workdir/wiki" commit --quiet -m "docs: sync from repo docs/wiki (source of truth)"
|
||||
echo "pushing wiki update for '$repo_name'"
|
||||
git -C "$workdir/wiki" push --quiet
|
||||
echo "done: ${#pages[@]} pages synced."
|
||||
else
|
||||
echo "done: wiki already up to date; nothing pushed."
|
||||
fi
|
||||
@@ -341,13 +341,11 @@ Ready-to-copy templates live in [`templates/`](templates/):
|
||||
|
||||
## K. Controller Handoff (required, every task)
|
||||
|
||||
Every LLM task **must end with a `Controller Handoff`** (exact title) — whether the
|
||||
Every LLM task **must end with a `Controller Handoff`** — whether the
|
||||
task was implementation, review, merge, issue triage, documentation,
|
||||
discussion-only, or blocked planning. It lets a controller LLM understand the
|
||||
current state immediately, without rereading the conversation.
|
||||
|
||||
The section title must be exactly "Controller Handoff" (or "Controller Handoff Summary" for long form). Reports without it are downgraded (see review_proofs.assess_controller_handoff).
|
||||
|
||||
**The compact format is the default.** It is written for controller-LLM
|
||||
readability, not as a full human status report. PR bodies still carry the
|
||||
full review detail — the handoff never replaces PR documentation.
|
||||
@@ -379,17 +377,6 @@ Role-specific fields (append to the compact block):
|
||||
`Linked issue status:`, `Cleanup status:`
|
||||
- author tasks: `Selected issue:`, `Claim/comment status:`,
|
||||
`PR number opened:`, `No review/merge:` (explicit confirmation)
|
||||
- continuation tasks (#188/#189): `Continuation mode:`, `Existing PR:`,
|
||||
`PR author:`, `Issue claim status:`, `Branch:`, `Old PR head:`,
|
||||
`New PR head:`, `Session authored PR:`, `Why continuation allowed:` —
|
||||
when rebasing with `git push --force-with-lease`, also record
|
||||
`Old remote head:`, `Lease head:`, `Pushed head:`, and confirm
|
||||
`Push branch only:` (feature branch only). Use the canonical secret sweep
|
||||
from `review_proofs.CANONICAL_SECRET_SWEEP_COMMAND` so runs are
|
||||
comparable. Issues with open PRs are excluded from fresh selection unless
|
||||
the operator explicitly requests continuation
|
||||
(`review_proofs.classify_issue_for_selection`,
|
||||
`assess_issue_selection_final_report`)
|
||||
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
|
||||
`Selected PR or reason none selected:`, `Inventory completeness:`
|
||||
|
||||
|
||||
@@ -17,30 +17,9 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
|
||||
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"
|
||||
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.
|
||||
- Empty-queue report wall (#198): if the final report claims "no open PRs",
|
||||
"queue empty", or "nothing to review", it must include verbatim:
|
||||
`pr_inventory_trust_gate.status`, trust-gate reasons, corroboration,
|
||||
remote/owner/repo/state filter, and the inventory MCP profile. A recent merge
|
||||
commit is not valid corroboration. Author-bound sessions must not present
|
||||
reviewer queue inventory as a reviewer decision.
|
||||
|
||||
Rules (llm-project-workflow):
|
||||
- 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.
|
||||
A different LLM-Agent-SHA does NOT make you a different actor — only a
|
||||
different authenticated Gitea user does (docs/llm-agent-sha.md).
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
"""Shared task→permission map for resolver and tool gates (#69).
|
||||
|
||||
``gitea_resolve_task_capability`` and issue-mutating MCP tools must agree on
|
||||
which profile operation each task requires. This module is the single source
|
||||
of truth; regression tests assert tool gates cannot drift from it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
||||
"create_issue": {
|
||||
"permission": "gitea.issue.create",
|
||||
"role": "author",
|
||||
},
|
||||
"comment_issue": {
|
||||
"permission": "gitea.issue.comment",
|
||||
"role": "author",
|
||||
},
|
||||
"close_issue": {
|
||||
"permission": "gitea.issue.close",
|
||||
"role": "author",
|
||||
},
|
||||
"claim_issue": {
|
||||
"permission": "gitea.issue.comment",
|
||||
"role": "author",
|
||||
},
|
||||
"mark_issue": {
|
||||
"permission": "gitea.issue.comment",
|
||||
"role": "author",
|
||||
},
|
||||
"set_issue_labels": {
|
||||
"permission": "gitea.issue.comment",
|
||||
"role": "author",
|
||||
},
|
||||
"create_branch": {
|
||||
"permission": "gitea.branch.create",
|
||||
"role": "author",
|
||||
},
|
||||
"push_branch": {
|
||||
"permission": "gitea.branch.push",
|
||||
"role": "author",
|
||||
},
|
||||
"create_pr": {
|
||||
"permission": "gitea.pr.create",
|
||||
"role": "author",
|
||||
},
|
||||
"comment_pr": {
|
||||
"permission": "gitea.pr.comment",
|
||||
"role": "author",
|
||||
},
|
||||
"close_pr": {
|
||||
"permission": "gitea.pr.close",
|
||||
"role": "author",
|
||||
},
|
||||
"address_pr_change_requests": {
|
||||
"permission": "gitea.branch.push",
|
||||
"role": "author",
|
||||
},
|
||||
"review_pr": {
|
||||
"permission": "gitea.pr.review",
|
||||
"role": "reviewer",
|
||||
},
|
||||
"merge_pr": {
|
||||
"permission": "gitea.pr.merge",
|
||||
"role": "reviewer",
|
||||
},
|
||||
"blind_pr_queue_review": {
|
||||
"permission": "gitea.pr.review",
|
||||
"role": "reviewer",
|
||||
},
|
||||
"request_changes_pr": {
|
||||
"permission": "gitea.pr.request_changes",
|
||||
"role": "reviewer",
|
||||
},
|
||||
"approve_pr": {
|
||||
"permission": "gitea.pr.approve",
|
||||
"role": "reviewer",
|
||||
},
|
||||
"delete_branch": {
|
||||
"permission": "gitea.branch.delete",
|
||||
"role": "author",
|
||||
},
|
||||
}
|
||||
|
||||
# Issue-mutating MCP tools and their resolver task keys.
|
||||
ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
|
||||
"gitea_create_issue": "create_issue",
|
||||
"gitea_close_issue": "close_issue",
|
||||
"gitea_create_issue_comment": "comment_issue",
|
||||
"gitea_mark_issue": "mark_issue",
|
||||
"gitea_set_issue_labels": "set_issue_labels",
|
||||
}
|
||||
|
||||
|
||||
def required_permission(task: str) -> str:
|
||||
"""Return the canonical operation a *task* requires (fail closed)."""
|
||||
try:
|
||||
return TASK_CAPABILITY_MAP[task]["permission"]
|
||||
except KeyError as exc:
|
||||
raise KeyError(f"Unknown task/action: {task!r} (fail closed)") from exc
|
||||
|
||||
|
||||
def required_role(task: str) -> str:
|
||||
"""Return author/reviewer role kind for *task* (fail closed)."""
|
||||
try:
|
||||
return TASK_CAPABILITY_MAP[task]["role"]
|
||||
except KeyError as exc:
|
||||
raise KeyError(f"Unknown task/action: {task!r} (fail closed)") from exc
|
||||
|
||||
|
||||
def tool_required_permission(tool_name: str) -> str:
|
||||
"""Return the operation an issue-mutating tool must gate on."""
|
||||
return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name])
|
||||
+2
-6
@@ -162,8 +162,7 @@ class _AuditWiringBase(unittest.TestCase):
|
||||
def _env(self, **extra):
|
||||
env = {"GITEA_AUDIT_LOG": self.audit_path,
|
||||
"GITEA_PROFILE_NAME": "gitea-author",
|
||||
"GITEA_ALLOWED_OPERATIONS": (
|
||||
"read,merge,gitea.issue.create,gitea.issue.close")}
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
|
||||
env.update(extra)
|
||||
return env
|
||||
|
||||
@@ -238,10 +237,7 @@ class TestSimpleToolAudit(_AuditWiringBase):
|
||||
def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role):
|
||||
# No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
|
||||
mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
|
||||
with patch.dict(os.environ, {
|
||||
"GITEA_PROFILE_NAME": "gitea-author",
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
|
||||
}, clear=True):
|
||||
with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
|
||||
gitea_create_issue(title="x", remote="prgs")
|
||||
issue_posts = [
|
||||
c for c in mock_api.call_args_list if c.args[0] == "POST"
|
||||
|
||||
@@ -148,19 +148,6 @@ class TestCapabilityStopTerminal(unittest.TestCase):
|
||||
)
|
||||
self.assertFalse(result["pure"])
|
||||
|
||||
def test_empty_queue_with_parsed_trusted_status_passes_gate_check(self):
|
||||
report = (
|
||||
"Cannot perform reviewer task under current profile. "
|
||||
"No reviewer mutations performed.\n"
|
||||
"Repository: Scaled-Tech-Consulting/Gitea-Tools\n"
|
||||
"pr_inventory_trust_gate.status: trusted_empty\n"
|
||||
"pr_inventory_trust_gate.corroborated: true\n"
|
||||
"Inventory profile: prgs-reviewer\n"
|
||||
"No open PRs in queue."
|
||||
)
|
||||
result = assess_capability_stop_terminal_report(report)
|
||||
self.assertTrue(result["pure"])
|
||||
|
||||
def test_pure_terminal_report_passes(self):
|
||||
report = (
|
||||
"Cannot perform reviewer task under current profile. "
|
||||
|
||||
@@ -107,24 +107,6 @@ class TestGetCredentials(unittest.TestCase):
|
||||
class TestGetAuthHeader(unittest.TestCase):
|
||||
"""Test the get_auth_header function."""
|
||||
|
||||
def setUp(self):
|
||||
self._saved_env = os.environ.copy()
|
||||
self._saved_configs = gitea_auth.DYNAMIC_CONFIGS.copy()
|
||||
for key in (
|
||||
"GITEA_MCP_CONFIG",
|
||||
"GITEA_MCP_PROFILE",
|
||||
"GITEA_TOKEN",
|
||||
"GITEA_TOKEN_PRGS",
|
||||
):
|
||||
os.environ.pop(key, None)
|
||||
gitea_auth.DYNAMIC_CONFIGS.clear()
|
||||
|
||||
def tearDown(self):
|
||||
os.environ.clear()
|
||||
os.environ.update(self._saved_env)
|
||||
gitea_auth.DYNAMIC_CONFIGS.clear()
|
||||
gitea_auth.DYNAMIC_CONFIGS.update(self._saved_configs)
|
||||
|
||||
@patch("gitea_auth.get_credentials", return_value=("user", "pass"))
|
||||
def test_returns_basic_header(self, _cred):
|
||||
header = gitea_auth.get_auth_header("example.com")
|
||||
|
||||
@@ -1,82 +0,0 @@
|
||||
"""Documentation checks for external Jenkins/GlitchTip MCP registration (#151)."""
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
DOC = REPO_ROOT / "docs" / "mcp-client-registration.md"
|
||||
|
||||
|
||||
def _doc_text():
|
||||
assert DOC.is_file(), "missing docs/mcp-client-registration.md"
|
||||
return DOC.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_registration_doc_names_canonical_servers():
|
||||
text = _doc_text()
|
||||
assert "jenkins-mcp" in text
|
||||
assert "glitchtip-mcp" in text
|
||||
assert "Historical names" in text
|
||||
assert "jenkins-readonly" in text
|
||||
assert "glitchtip-readonly" in text
|
||||
|
||||
|
||||
def test_registration_doc_lists_expected_tool_visibility():
|
||||
text = _doc_text()
|
||||
for tool in (
|
||||
"jenkins_whoami",
|
||||
"jenkins_list_jobs",
|
||||
"jenkins_latest_build",
|
||||
"jenkins_build_status",
|
||||
"jenkins_get_build",
|
||||
"glitchtip_whoami",
|
||||
"glitchtip_list_projects",
|
||||
"glitchtip_list_unresolved",
|
||||
"glitchtip_get_issue",
|
||||
"glitchtip_recent_events",
|
||||
"glitchtip_search",
|
||||
):
|
||||
assert tool in text
|
||||
|
||||
|
||||
def test_registration_doc_requires_reload_and_negative_case():
|
||||
text = _doc_text()
|
||||
lower = text.lower()
|
||||
assert "reconnect" in lower
|
||||
assert "reload" in lower
|
||||
assert "enabled but no usable tools" in lower
|
||||
assert "SKIPPED" in text
|
||||
|
||||
|
||||
def test_registration_doc_preserves_boundaries():
|
||||
text = " ".join(_doc_text().split())
|
||||
for phrase in (
|
||||
"Do not add Jenkins or GlitchTip credentials to the Gitea MCP server",
|
||||
"do not add Gitea write credentials to the GlitchTip server",
|
||||
"must not expose build trigger tools",
|
||||
"remains read-only",
|
||||
):
|
||||
assert phrase in text
|
||||
|
||||
|
||||
def test_registration_doc_has_no_secret_material_or_live_urls():
|
||||
text = _doc_text()
|
||||
for marker in (
|
||||
"https://",
|
||||
"http://",
|
||||
"Authorization:",
|
||||
"BEGIN PRIVATE KEY",
|
||||
"ghp_",
|
||||
"token-value",
|
||||
):
|
||||
assert marker not in text
|
||||
|
||||
|
||||
def test_related_docs_link_registration_doc():
|
||||
for name in (
|
||||
"README.md",
|
||||
"docs/llm-workflow-runbooks.md",
|
||||
"docs/architecture/jenkins-readonly-build-status-design.md",
|
||||
"docs/architecture/glitchtip-readonly-tools-design.md",
|
||||
):
|
||||
text = (REPO_ROOT / name).read_text(encoding="utf-8")
|
||||
assert "mcp-client-registration.md" in text, (
|
||||
f"{name} does not link docs/mcp-client-registration.md")
|
||||
@@ -1,58 +0,0 @@
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
import role_session_router
|
||||
from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
|
||||
|
||||
class TestMCPHealth(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
self.temp_file = os.path.join(self.project_root, "test_temp_conflict.py")
|
||||
if os.path.exists(self.temp_file):
|
||||
os.remove(self.temp_file)
|
||||
|
||||
def tearDown(self):
|
||||
if os.path.exists(self.temp_file):
|
||||
os.remove(self.temp_file)
|
||||
|
||||
def test_startup_conflict_detection(self):
|
||||
# Create a Python file with conflict markers constructed dynamically
|
||||
with open(self.temp_file, "w") as f:
|
||||
f.write("<" * 7 + " HEAD\n")
|
||||
f.write("print('hello')\n")
|
||||
f.write("=" * 7 + "\n")
|
||||
f.write("print('world')\n")
|
||||
f.write(">" * 7 + " main\n")
|
||||
|
||||
# Run mcp_server.py
|
||||
venv_python = os.path.join(self.project_root, "venv", "bin", "python")
|
||||
script_path = os.path.join(self.project_root, "mcp_server.py")
|
||||
res = subprocess.run(
|
||||
[venv_python, script_path],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
cwd=self.project_root
|
||||
)
|
||||
|
||||
self.assertEqual(res.returncode, 1)
|
||||
stderr = res.stderr.decode()
|
||||
self.assertIn("infra_stop", stderr)
|
||||
self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
|
||||
|
||||
@patch("role_session_router.check_mid_merge", return_value=True)
|
||||
@patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
|
||||
def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
|
||||
res = gitea_route_task_session("review_pr")
|
||||
self.assertEqual(res["route_result"], "infra_stop")
|
||||
self.assertIn("infra_stop", res["message"])
|
||||
|
||||
@patch("role_session_router.check_mid_merge", return_value=True)
|
||||
@patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
|
||||
def test_resolve_task_capability_blocks_during_merge(self, mock_profile, mock_check):
|
||||
res = gitea_resolve_task_capability("review_pr")
|
||||
self.assertTrue(res["infra_stop"])
|
||||
self.assertFalse(res["allowed_in_current_session"])
|
||||
self.assertIn("infra_stop", res["exact_safe_next_action"])
|
||||
@@ -19,9 +19,6 @@ from mcp_server import gitea_create_issue # noqa: E402
|
||||
from review_proofs import assess_duplicate_search_proof as proof_assess # noqa: E402
|
||||
|
||||
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
||||
ISSUE_WRITE_ENV = {
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
|
||||
}
|
||||
CANONICAL_TITLE = (
|
||||
"Add hard queue-target resolution wall before PR inventory "
|
||||
"or empty-queue claims"
|
||||
@@ -121,8 +118,7 @@ class TestCreateIssueMCPGate(unittest.TestCase):
|
||||
mock_get_all.return_value = [
|
||||
{"number": 201, "title": CANONICAL_TITLE, "state": "open"},
|
||||
]
|
||||
with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
|
||||
result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
|
||||
result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
|
||||
self.assertFalse(result["performed"])
|
||||
self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE)
|
||||
mock_api.assert_not_called()
|
||||
@@ -139,14 +135,13 @@ class TestCreateIssueMCPGate(unittest.TestCase):
|
||||
{"number": 201, "title": CANONICAL_TITLE, "state": "open"},
|
||||
]
|
||||
mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"}
|
||||
with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
|
||||
result = gitea_create_issue(
|
||||
title=CANONICAL_TITLE,
|
||||
body="Follow-up scope",
|
||||
remote="prgs",
|
||||
allow_duplicate_override=True,
|
||||
split_from_issue=201,
|
||||
)
|
||||
result = gitea_create_issue(
|
||||
title=CANONICAL_TITLE,
|
||||
body="Follow-up scope",
|
||||
remote="prgs",
|
||||
allow_duplicate_override=True,
|
||||
split_from_issue=201,
|
||||
)
|
||||
self.assertEqual(result["number"], 210)
|
||||
payload = mock_api.call_args[0][3]
|
||||
self.assertIn("Operator-approved split from #201.", payload["body"])
|
||||
@@ -161,8 +156,7 @@ class TestCreateIssueMCPGate(unittest.TestCase):
|
||||
):
|
||||
mock_role_check.return_value = (True, [])
|
||||
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
|
||||
with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
|
||||
result = gitea_create_issue(title="Unique new issue", body="body text")
|
||||
result = gitea_create_issue(title="Unique new issue", body="body text")
|
||||
self.assertEqual(result["number"], 1)
|
||||
|
||||
|
||||
|
||||
@@ -1,210 +0,0 @@
|
||||
"""Regression tests: issue-write tool gates match gitea_resolve_task_capability (#69)."""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server
|
||||
import task_capability_map
|
||||
|
||||
CONFIG = {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"ctx": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"comment-only": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "commenter",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": ["gitea.read", "gitea.issue.comment"],
|
||||
"forbidden_operations": [
|
||||
"gitea.issue.create", "gitea.issue.close"],
|
||||
"execution_profile": "comment-only",
|
||||
},
|
||||
"full-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "author-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.issue.create", "gitea.issue.close",
|
||||
"gitea.issue.comment"],
|
||||
"forbidden_operations": [],
|
||||
"execution_profile": "full-author",
|
||||
},
|
||||
},
|
||||
"rules": {"allow_runtime_switching": False},
|
||||
}
|
||||
|
||||
ISSUE_WRITE_TASKS = (
|
||||
"create_issue", "close_issue", "comment_issue", "mark_issue",
|
||||
"set_issue_labels",
|
||||
)
|
||||
|
||||
TOOL_CALLS = {
|
||||
"create_issue": lambda: mcp_server.gitea_create_issue(
|
||||
title="New issue", remote="prgs"),
|
||||
"close_issue": lambda: mcp_server.gitea_close_issue(
|
||||
issue_number=9, remote="prgs"),
|
||||
"comment_issue": lambda: mcp_server.gitea_create_issue_comment(
|
||||
issue_number=9, body="hello", remote="prgs"),
|
||||
"mark_issue": lambda: mcp_server.gitea_mark_issue(
|
||||
issue_number=9, action="start", remote="prgs"),
|
||||
"set_issue_labels": lambda: mcp_server.gitea_set_issue_labels(
|
||||
issue_number=9, labels=["bug"], remote="prgs"),
|
||||
}
|
||||
|
||||
|
||||
class IssueWriteGateBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
self._dir.cleanup()
|
||||
|
||||
def _env(self, profile: str) -> dict:
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
}
|
||||
|
||||
|
||||
class TestResolverToolAlignment(unittest.TestCase):
|
||||
def test_issue_mutation_tools_match_resolver_permissions(self):
|
||||
for tool_name, task_key in (
|
||||
task_capability_map.ISSUE_MUTATION_TOOL_TASKS.items()
|
||||
):
|
||||
self.assertEqual(
|
||||
task_capability_map.tool_required_permission(tool_name),
|
||||
task_capability_map.required_permission(task_key),
|
||||
msg=f"{tool_name} gate drift from resolver task {task_key}",
|
||||
)
|
||||
|
||||
|
||||
class TestDeniedIssueWritesFailClosed(IssueWriteGateBase):
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolver_denied_tasks_block_raw_tools(
|
||||
self, _auth, mock_api, _get_all, _role
|
||||
):
|
||||
mock_api.return_value = {"number": 1}
|
||||
with patch.dict(os.environ, self._env("comment-only"), clear=True):
|
||||
for task in ISSUE_WRITE_TASKS:
|
||||
with self.subTest(task=task):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task=task, remote="prgs")
|
||||
if resolve["allowed_in_current_session"]:
|
||||
continue
|
||||
result = TOOL_CALLS[task]()
|
||||
self.assertFalse(result.get("success", True), task)
|
||||
self.assertFalse(result.get("performed", True), task)
|
||||
self.assertTrue(result.get("reasons"), task)
|
||||
self.assertIn("permission_report", result, task)
|
||||
self.assertEqual(
|
||||
result["permission_report"]["missing_permission"],
|
||||
resolve["required_operation_permission"],
|
||||
)
|
||||
|
||||
|
||||
class TestAllowedIssueWritesProceed(IssueWriteGateBase):
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolver_allowed_create_issue_proceeds(
|
||||
self, _auth, mock_api, _get_all, _role
|
||||
):
|
||||
mock_api.return_value = {"number": 42, "html_url": "https://x/42"}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task="create_issue", remote="prgs")
|
||||
self.assertTrue(resolve["allowed_in_current_session"])
|
||||
result = mcp_server.gitea_create_issue(title="Allowed", remote="prgs")
|
||||
self.assertEqual(result.get("number"), 42)
|
||||
post_calls = [
|
||||
c for c in mock_api.call_args_list if c.args[0] == "POST"
|
||||
]
|
||||
self.assertTrue(post_calls)
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolver_allowed_close_issue_proceeds(self, _auth, mock_api):
|
||||
mock_api.return_value = {"state": "closed"}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task="close_issue", remote="prgs")
|
||||
self.assertTrue(resolve["allowed_in_current_session"])
|
||||
result = mcp_server.gitea_close_issue(issue_number=9, remote="prgs")
|
||||
self.assertTrue(result.get("success"))
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "GET" and url.endswith("/labels?limit=100"):
|
||||
return [{"id": 10, "name": "status:in-progress"}]
|
||||
if method == "POST" and "/issues/9/labels" in url:
|
||||
return [{"name": "status:in-progress"}]
|
||||
if method == "GET" and url.endswith("/user"):
|
||||
return {"login": "author-user"}
|
||||
return {}
|
||||
mock_api.side_effect = api_side_effect
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task="mark_issue", remote="prgs")
|
||||
self.assertTrue(resolve["allowed_in_current_session"])
|
||||
result = mcp_server.gitea_mark_issue(
|
||||
issue_number=9, action="start", remote="prgs")
|
||||
self.assertTrue(result.get("success"))
|
||||
|
||||
|
||||
class TestCreateIssueMissingPermissionReport(IssueWriteGateBase):
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_create_issue_without_create_permission_blocked(
|
||||
self, _auth, mock_api, _get_all, _role
|
||||
):
|
||||
with patch.dict(os.environ, self._env("comment-only"), clear=True):
|
||||
result = mcp_server.gitea_create_issue(title="Blocked", remote="prgs")
|
||||
self.assertFalse(result["success"])
|
||||
self.assertFalse(result["performed"])
|
||||
self.assertIsNone(result["number"])
|
||||
self.assertEqual(
|
||||
result["permission_report"]["missing_permission"],
|
||||
"gitea.issue.create",
|
||||
)
|
||||
mock_api.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+45
-208
@@ -47,38 +47,6 @@ import mcp_server
|
||||
|
||||
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
||||
|
||||
# Issue-write tools are profile-gated (#69).
|
||||
ISSUE_WRITE_ENV = {
|
||||
"GITEA_ALLOWED_OPERATIONS": (
|
||||
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
|
||||
),
|
||||
}
|
||||
|
||||
# create_pr is gated on author profile + namespace (#69, #209).
|
||||
CREATE_PR_ENV = {
|
||||
"GITEA_PROFILE_NAME": "author-test",
|
||||
"GITEA_ALLOWED_OPERATIONS": (
|
||||
"gitea.read,gitea.pr.create,gitea.branch.push"
|
||||
),
|
||||
"GITEA_FORBIDDEN_OPERATIONS": (
|
||||
"gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
|
||||
),
|
||||
}
|
||||
|
||||
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
|
||||
@@ -92,7 +60,7 @@ class TestCreateIssue(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_creates_issue(self, _auth, _get_all, mock_api, _role):
|
||||
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
result = gitea_create_issue(title="Test issue", body="body text")
|
||||
self.assertEqual(result["number"], 1)
|
||||
self.assertNotIn("url", result)
|
||||
@@ -109,8 +77,7 @@ class TestCreateIssue(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
|
||||
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
|
||||
env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||
result = gitea_create_issue(title="Test issue", body="body text")
|
||||
self.assertIn("issues/1", result["url"])
|
||||
|
||||
@@ -121,8 +88,7 @@ class TestCreateIssue(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
|
||||
mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
result = gitea_create_issue(title="Test", remote="prgs")
|
||||
result = gitea_create_issue(title="Test", remote="prgs")
|
||||
self.assertEqual(result["number"], 5)
|
||||
url = mock_api.call_args[0][1]
|
||||
self.assertIn("gitea.prgs.cc", url)
|
||||
@@ -134,56 +100,32 @@ class TestCreateIssue(unittest.TestCase):
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestCreatePR(unittest.TestCase):
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("os.path.exists", return_value=True)
|
||||
@patch("builtins.open")
|
||||
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 = lock_json
|
||||
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
|
||||
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
|
||||
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, {}, clear=True):
|
||||
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
||||
self.assertEqual(result["number"], 3)
|
||||
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]
|
||||
self.assertEqual(payload["head"], "feat/x")
|
||||
self.assertEqual(payload["base"], "main")
|
||||
self.assertIn("Closes #123", payload["title"])
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("os.path.exists", return_value=True)
|
||||
@patch("builtins.open")
|
||||
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 = lock_json
|
||||
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
|
||||
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
|
||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
||||
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
|
||||
@@ -194,8 +136,7 @@ class TestCloseIssue(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_closes_issue(self, _auth, mock_api):
|
||||
mock_api.return_value = {"state": "closed"}
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
result = gitea_close_issue(issue_number=42)
|
||||
result = gitea_close_issue(issue_number=42)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertIn("42", result["message"])
|
||||
patch_call = next(call for call in mock_api.call_args_list if call[0][0] == "PATCH")
|
||||
@@ -294,8 +235,7 @@ class TestMarkIssue(unittest.TestCase):
|
||||
[{"id": 10, "name": "status:in-progress"}],
|
||||
[{"name": "status:in-progress"}],
|
||||
]
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
result = gitea_mark_issue(issue_number=5, action="start")
|
||||
result = gitea_mark_issue(issue_number=5, action="start")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertIn("claimed", result["message"])
|
||||
|
||||
@@ -306,8 +246,7 @@ class TestMarkIssue(unittest.TestCase):
|
||||
[{"id": 10, "name": "status:in-progress"}],
|
||||
None,
|
||||
]
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
result = gitea_mark_issue(issue_number=5, action="done")
|
||||
result = gitea_mark_issue(issue_number=5, action="done")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertIn("released", result["message"])
|
||||
|
||||
@@ -319,9 +258,8 @@ class TestMarkIssue(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_missing_label_raises(self, _auth, mock_api):
|
||||
mock_api.return_value = [] # no labels exist
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
with self.assertRaises(RuntimeError):
|
||||
gitea_mark_issue(issue_number=5, action="start")
|
||||
with self.assertRaises(RuntimeError):
|
||||
gitea_mark_issue(issue_number=5, action="start")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -331,14 +269,12 @@ class TestAuthErrors(unittest.TestCase):
|
||||
|
||||
@patch("mcp_server.get_auth_header", return_value=None)
|
||||
def test_no_credentials_raises(self, _auth):
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
with self.assertRaises(RuntimeError):
|
||||
gitea_create_issue(title="test")
|
||||
with self.assertRaises(RuntimeError):
|
||||
gitea_create_issue(title="test")
|
||||
|
||||
def test_unknown_remote_raises(self):
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
with self.assertRaises(ValueError):
|
||||
gitea_create_issue(title="test", remote="nonexistent")
|
||||
with self.assertRaises(ValueError):
|
||||
gitea_create_issue(title="test", remote="nonexistent")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -2003,7 +1939,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
|
||||
patch("gitea_audit.audit_enabled", return_value=True).start()
|
||||
self.mock_audit = patch("gitea_audit.write_event").start()
|
||||
# gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
|
||||
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
|
||||
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
@@ -2803,8 +2739,8 @@ class TestIssueLocking(unittest.TestCase):
|
||||
"""Test issue locking and PR gating constraints."""
|
||||
|
||||
def tearDown(self):
|
||||
if os.path.exists(ISSUE_LOCK_FILE):
|
||||
os.remove(ISSUE_LOCK_FILE)
|
||||
if os.path.exists("/tmp/gitea_issue_lock.json"):
|
||||
os.remove("/tmp/gitea_issue_lock.json")
|
||||
|
||||
@patch("mcp_server.api_get_all")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@@ -2812,7 +2748,7 @@ class TestIssueLocking(unittest.TestCase):
|
||||
mock_api.return_value = [] # no open PRs
|
||||
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
||||
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):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
@@ -2845,134 +2781,35 @@ class TestIssueLocking(unittest.TestCase):
|
||||
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("already tied to an open PR", str(ctx.exception))
|
||||
|
||||
@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)
|
||||
def test_create_pr_missing_lock_fails(self, _auth, _role):
|
||||
if os.path.exists(ISSUE_LOCK_FILE):
|
||||
os.remove(ISSUE_LOCK_FILE)
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
|
||||
def test_create_pr_missing_lock_fails(self, _auth):
|
||||
if os.path.exists("/tmp/gitea_issue_lock.json"):
|
||||
os.remove("/tmp/gitea_issue_lock.json")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("Issue lock is missing", str(ctx.exception))
|
||||
|
||||
@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)
|
||||
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
|
||||
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(_sample_issue_lock(
|
||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
|
||||
def test_create_pr_branch_mismatch_fails(self, _auth):
|
||||
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
||||
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
|
||||
self.assertIn("does not match locked branch", str(ctx.exception))
|
||||
|
||||
@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)
|
||||
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
|
||||
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(_sample_issue_lock(
|
||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
for term in ("equivalent to #196", "related to #196", "same as #196"):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("contains forbidden term", str(ctx.exception))
|
||||
|
||||
@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)
|
||||
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
|
||||
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(_sample_issue_lock(
|
||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
def test_create_pr_forbidden_terms_fails(self, _auth):
|
||||
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
||||
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
||||
for term in ("equivalent to #196", "related to #196", "same as #196"):
|
||||
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=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("contains forbidden term", str(ctx.exception))
|
||||
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_pr_missing_closes_ref_fails(self, _auth):
|
||||
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
||||
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pre-flight ordering and workspace edit block (#210)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestPreflightVerification(unittest.TestCase):
|
||||
"""Assert that pre-flight verification gates order correctly."""
|
||||
|
||||
def setUp(self):
|
||||
# Save real global variables
|
||||
import mcp_server
|
||||
self.orig_whoami_called = mcp_server._preflight_whoami_called
|
||||
self.orig_capability_called = mcp_server._preflight_capability_called
|
||||
self.orig_whoami_violation = mcp_server._preflight_whoami_violation
|
||||
self.orig_capability_violation = mcp_server._preflight_capability_violation
|
||||
self.orig_resolved_role = mcp_server._preflight_resolved_role
|
||||
|
||||
# Reset state for each test
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
mcp_server._preflight_whoami_violation = False
|
||||
mcp_server._preflight_capability_violation = False
|
||||
mcp_server._preflight_resolved_role = None
|
||||
|
||||
def tearDown(self):
|
||||
# Restore real global variables
|
||||
import mcp_server
|
||||
mcp_server._preflight_whoami_called = self.orig_whoami_called
|
||||
mcp_server._preflight_capability_called = self.orig_capability_called
|
||||
mcp_server._preflight_whoami_violation = self.orig_whoami_violation
|
||||
mcp_server._preflight_capability_violation = self.orig_capability_violation
|
||||
mcp_server._preflight_resolved_role = self.orig_resolved_role
|
||||
if "GITEA_TEST_FORCE_DIRTY" in os.environ:
|
||||
del os.environ["GITEA_TEST_FORCE_DIRTY"]
|
||||
|
||||
def test_preflight_whoami_violation(self):
|
||||
import mcp_server
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
mcp_server._preflight_capability_called = True
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertTrue(mcp_server._preflight_whoami_violation)
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
|
||||
|
||||
def test_preflight_capability_violation(self):
|
||||
import mcp_server
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
mcp_server._preflight_whoami_called = True
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
self.assertTrue(mcp_server._preflight_capability_violation)
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
|
||||
|
||||
def test_preflight_not_called_fails_closed(self):
|
||||
import mcp_server
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
|
||||
|
||||
mcp_server._preflight_whoami_called = True
|
||||
with self.assertRaises(RuntimeError) as ctx2:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception))
|
||||
|
||||
def test_preflight_reviewer_mutation_violation(self):
|
||||
import mcp_server
|
||||
mcp_server._preflight_whoami_called = True
|
||||
mcp_server._preflight_capability_called = True
|
||||
mcp_server._preflight_resolved_role = "reviewer"
|
||||
|
||||
# When dirty, reviewer edits are blocked
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
|
||||
|
||||
# When clean, reviewer is allowed
|
||||
del os.environ["GITEA_TEST_FORCE_DIRTY"]
|
||||
mcp_server.verify_preflight_purity()
|
||||
|
||||
@@ -234,8 +234,8 @@ class TestProjectSkills(GuideTestBase):
|
||||
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
||||
r = mcp_list_project_skills()
|
||||
by_name = {s["name"]: s for s in r["skills"]}
|
||||
self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
|
||||
self.assertEqual(by_name["glitchtip-mcp"]["status"], "external-mcp")
|
||||
self.assertNotEqual(by_name["jenkins-mcp"]["status"], "available")
|
||||
self.assertNotEqual(by_name["glitchtip-mcp"]["status"], "available")
|
||||
|
||||
def test_no_urls_in_registry(self):
|
||||
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
||||
@@ -246,29 +246,16 @@ class TestProjectSkills(GuideTestBase):
|
||||
self.assertNotIn("keychain:", blob)
|
||||
|
||||
def test_enabled_but_no_usable_tools_negative_assertion(self):
|
||||
"""Negative assertion for 'enabled but no usable tools' (per issue #151)."""
|
||||
"""Negative assertion for 'enabled but no usable tools' (per issue #146)."""
|
||||
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
||||
r = mcp_list_project_skills()
|
||||
by_name = {s["name"]: s for s in r["skills"]}
|
||||
# jenkins-mcp is an external boundary; Gitea profile discovery must not
|
||||
# treat a local config entry as a usable Jenkins tool surface.
|
||||
# jenkins-mcp is designed-not-implemented; even if "enabled" in config,
|
||||
# it should not be usable/available to current profile without tools.
|
||||
self.assertIn("jenkins-mcp", by_name)
|
||||
self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
|
||||
self.assertEqual(by_name["jenkins-mcp"]["status"], "designed-not-implemented")
|
||||
self.assertFalse(by_name["jenkins-mcp"].get("available_to_current_profile", False))
|
||||
|
||||
def test_external_mcp_skill_guides_name_expected_tools(self):
|
||||
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
||||
jenkins = mcp_get_skill_guide("jenkins-mcp")
|
||||
glitchtip = mcp_get_skill_guide("glitchtip-mcp")
|
||||
self.assertEqual(jenkins["skill"]["status"], "external-mcp")
|
||||
self.assertIn("jenkins_whoami", " ".join(jenkins["steps"]))
|
||||
self.assertIn("jenkins_get_build", " ".join(jenkins["steps"]))
|
||||
self.assertIn("SKIPPED", " ".join(jenkins["steps"]))
|
||||
self.assertEqual(glitchtip["skill"]["status"], "external-mcp")
|
||||
self.assertIn("glitchtip_whoami", " ".join(glitchtip["steps"]))
|
||||
self.assertIn("glitchtip_search", " ".join(glitchtip["steps"]))
|
||||
self.assertIn("SKIPPED", " ".join(glitchtip["steps"]))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# mcp_get_skill_guide
|
||||
|
||||
@@ -261,79 +261,12 @@ class TestPRQueueInventory(unittest.TestCase):
|
||||
for call in mock_api.call_args_list:
|
||||
self.assertEqual(call.args[0], "GET")
|
||||
|
||||
@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_runs_trust_gate_and_blocks_without_trusted_empty(
|
||||
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):
|
||||
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
|
||||
"""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"]:
|
||||
mock_get_profile.return_value = {
|
||||
"profile_name": "gitea-author",
|
||||
|
||||
@@ -130,17 +130,6 @@ class TestResolveTaskCapability(unittest.TestCase):
|
||||
self.assertIn("author-profile", res["matching_configured_profile"])
|
||||
self.assertIn("reviewer-profile", res["matching_configured_profile"])
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolve_mark_issue_author_profile_allowed(self, _auth, _api):
|
||||
# #183: mark_issue must map to gitea.issue.comment (prgs-author allowed op).
|
||||
with patch.dict(os.environ, self._env("author-profile")):
|
||||
res = mcp_server.gitea_resolve_task_capability(task="mark_issue", remote="prgs")
|
||||
self.assertEqual(res["requested_task"], "mark_issue")
|
||||
self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
|
||||
self.assertTrue(res["allowed_in_current_session"])
|
||||
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
||||
|
||||
def test_resolve_unknown_task_fails_closed(self):
|
||||
with patch.dict(os.environ, self._env("author-profile")):
|
||||
with self.assertRaises(ValueError):
|
||||
|
||||
+6
-646
@@ -21,32 +21,16 @@ import unittest
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import ( # noqa: E402
|
||||
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
|
||||
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
|
||||
assess_author_pr_report,
|
||||
assess_capability_evidence,
|
||||
assess_capability_proof,
|
||||
assess_contradictory_no_pr_claim,
|
||||
assess_continuation_mode_report,
|
||||
assess_force_with_lease_push_report,
|
||||
canonical_secret_sweep_report,
|
||||
CANONICAL_SECRET_SWEEP_COMMAND,
|
||||
assess_controller_handoff,
|
||||
assess_edited_pr_inventory_coverage,
|
||||
assess_empty_queue_report,
|
||||
assess_fresh_issue_selection,
|
||||
assess_inventory_completeness,
|
||||
assess_issue_selection_final_report,
|
||||
assess_reviewer_queue_inventory,
|
||||
assess_live_state_recheck,
|
||||
assess_review_mutation_final_report,
|
||||
assess_role_boundary,
|
||||
assess_secret_sweep,
|
||||
assess_self_review_contamination,
|
||||
assess_sweep_evidence,
|
||||
assess_validation_report,
|
||||
build_final_report,
|
||||
classify_issue_for_selection,
|
||||
pr_inventory_trust_gate,
|
||||
resolve_repos_from_user_reference,
|
||||
verify_pinned_head_checkout,
|
||||
@@ -196,24 +180,6 @@ def _good_review_mutation():
|
||||
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):
|
||||
kwargs = {
|
||||
"task_role": "reviewer",
|
||||
@@ -224,50 +190,6 @@ def _good_role_boundary_179(**overrides):
|
||||
return assess_role_boundary(**kwargs)
|
||||
|
||||
|
||||
GOOD_HANDOFF = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"",
|
||||
"- Task: review PR #999",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: reviewer",
|
||||
"- Identity: sysadmin / prgs-reviewer",
|
||||
"- Issue/PR: #182 / PR #999",
|
||||
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: 700 passed, 6 skipped",
|
||||
"- Mutations: review only",
|
||||
"- Workspace mutations: none",
|
||||
"- Current status: PR open",
|
||||
"- Blockers: none",
|
||||
"- Next: merge if approved",
|
||||
"- Safety: no self-review; no self-merge; no secrets",
|
||||
])
|
||||
|
||||
|
||||
def _good_handoff():
|
||||
return assess_controller_handoff(GOOD_HANDOFF)
|
||||
|
||||
|
||||
def _good_capability_proof():
|
||||
return assess_capability_proof({
|
||||
"comment_issue": {
|
||||
"requested_task": "comment_issue",
|
||||
"required_operation_permission": "gitea.issue.comment",
|
||||
"allowed_in_current_session": True,
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
def _good_secret_sweep(**overrides):
|
||||
sweep = {
|
||||
"method": "git diff prgs/master...HEAD | grep -iE 'token|secret'",
|
||||
"scope": "full PR diff relative to prgs/master",
|
||||
"clean": True,
|
||||
}
|
||||
sweep.update(overrides)
|
||||
return assess_secret_sweep(sweep)
|
||||
|
||||
|
||||
class TestCheckoutProof(unittest.TestCase):
|
||||
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
|
||||
|
||||
@@ -623,16 +545,6 @@ class TestFinalReport(unittest.TestCase):
|
||||
"live_state": _good_live_state(),
|
||||
"role_boundary": _good_role_boundary(),
|
||||
"review_mutation": _good_review_mutation(),
|
||||
"controller_handoff": _good_handoff(),
|
||||
"capability_proof": _good_capability_proof(),
|
||||
"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)
|
||||
return build_final_report(**kwargs)
|
||||
@@ -750,28 +662,6 @@ class TestFinalReport(unittest.TestCase):
|
||||
self.assertEqual(report["grade"], "blocked")
|
||||
self.assertFalse(report["merge_allowed"])
|
||||
|
||||
def test_failed_capability_proof_downgrades(self):
|
||||
report = self._report(
|
||||
capability_proof={
|
||||
"proven": False,
|
||||
"reasons": ["task mark_issue not allowed"],
|
||||
}
|
||||
)
|
||||
self.assertNotEqual(report["grade"], "A")
|
||||
self.assertTrue(
|
||||
any("capability" in r for r in report["downgrade_reasons"])
|
||||
)
|
||||
|
||||
def test_failed_sweep_proof_downgrades(self):
|
||||
report = self._report(
|
||||
sweep_proof={
|
||||
"proven": False,
|
||||
"reasons": ["method missing"],
|
||||
}
|
||||
)
|
||||
self.assertNotEqual(report["grade"], "A")
|
||||
self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"]))
|
||||
|
||||
|
||||
class TestStdoutIsolation(unittest.TestCase):
|
||||
"""Regression test for #178: tests must not close or corrupt stdout/stderr
|
||||
@@ -899,7 +789,6 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: 700 passed, 6 skipped",
|
||||
"- Mutations: one PR opened",
|
||||
"- Workspace mutations: none",
|
||||
"- Current status: PR open",
|
||||
"- Blockers: none",
|
||||
"- Next: review PR #999",
|
||||
@@ -937,17 +826,12 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertIn("Pinned reviewed head", result["missing_fields"])
|
||||
self.assertIn("Worktree path", result["missing_fields"])
|
||||
self.assertIn("Merge result", result["missing_fields"])
|
||||
|
||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected PR: #999",
|
||||
"- Reviewer eligibility: passed",
|
||||
"- 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",
|
||||
"- Merge result: merged",
|
||||
"- Linked issue status: closed",
|
||||
@@ -956,46 +840,24 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
result = assess_controller_handoff(complete, role="review")
|
||||
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):
|
||||
complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
|
||||
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([
|
||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #182",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #999",
|
||||
"- 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.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):
|
||||
# 1. equivalent reference blocked
|
||||
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: Issue #194 / #196 equivalent",
|
||||
"- Issue lock proof: lock before diff on feat/x @ master",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #999",
|
||||
"- No review/merge: confirmed",
|
||||
@@ -1007,7 +869,6 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
# 2. multiple issues blocked
|
||||
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #194, #196",
|
||||
"- Issue lock proof: lock before diff on feat/x @ master",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #999",
|
||||
"- No review/merge: confirmed",
|
||||
@@ -1019,7 +880,6 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
def test_author_role_rejects_fuzzy_pr_number(self):
|
||||
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #196",
|
||||
"- Issue lock proof: lock before diff on feat/issue-196 @ master",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: PR #203 / #204 equivalent",
|
||||
"- No review/merge: confirmed",
|
||||
@@ -1032,10 +892,6 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Repositories checked: Gitea-Tools, mcp-control-plane",
|
||||
"- Open PR counts: 2 / 0",
|
||||
"- PR inventory trust gate: trusted_nonempty / trusted_empty",
|
||||
"- Trust gate reasons: none",
|
||||
"- Trust gate corroborated: true",
|
||||
"- Inventory profile: prgs-reviewer",
|
||||
"- Selected PR or reason: none eligible (self-authored)",
|
||||
"- Inventory completeness: complete, no pagination needed",
|
||||
])
|
||||
@@ -1053,23 +909,6 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
self.assertIn("assess_controller_handoff", skill)
|
||||
self.assertIn("issue #182", skill)
|
||||
|
||||
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
|
||||
# 1. Workspace mutations: none is rejected when local_edits is True
|
||||
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
|
||||
self._author_role_fields(issue_number=196, pr_number=203))
|
||||
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
|
||||
self.assertEqual(res["verdict"], "incomplete")
|
||||
self.assertIn("Workspace mutations", res["missing_fields"])
|
||||
|
||||
# 2. Workspace mutations: edited files is allowed when local_edits is True
|
||||
complete_eq = self.BASE_HANDOFF.replace(
|
||||
"- Workspace mutations: none",
|
||||
"- Workspace mutations: edited review_proofs.py",
|
||||
) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
|
||||
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
|
||||
self.assertEqual(res2["verdict"], "complete")
|
||||
|
||||
|
||||
|
||||
class TestReviewMutationFinalReport(unittest.TestCase):
|
||||
"""Final reports must list exactly one live review mutation."""
|
||||
@@ -1228,157 +1067,6 @@ class TestPRInventoryTrustGate(unittest.TestCase):
|
||||
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 TestAssessEmptyQueueReport(unittest.TestCase):
|
||||
"""Issue #198: empty-queue reports require formal trust-gate proof."""
|
||||
|
||||
def _trusted_report(self, **extra):
|
||||
lines = [
|
||||
"Queue inventory complete.",
|
||||
"Repository: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"Open PR count: 0",
|
||||
"pr_inventory_trust_gate.status: trusted_empty",
|
||||
"pr_inventory_trust_gate.corroborated: true",
|
||||
"Inventory profile: prgs-reviewer",
|
||||
"Workflow correctly stops with nothing to review.",
|
||||
]
|
||||
lines.extend(extra)
|
||||
return "\n".join(lines)
|
||||
|
||||
def test_non_empty_report_not_claimed(self):
|
||||
result = assess_empty_queue_report("Reviewed PR #236 and merged.")
|
||||
self.assertFalse(result["claimed"])
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_empty_claim_without_trust_gate_blocked(self):
|
||||
report = (
|
||||
"Open PR count: 0\n"
|
||||
"Pagination complete: yes\n"
|
||||
"Queue cleared."
|
||||
)
|
||||
result = assess_empty_queue_report(report)
|
||||
self.assertTrue(result["claimed"])
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_trusted_empty_report_with_required_fields_passes(self):
|
||||
result = assess_empty_queue_report(self._trusted_report())
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_weak_merge_commit_corroboration_blocked(self):
|
||||
report = (
|
||||
"Open PR count: 0\n"
|
||||
"Master latest commit is merge of PR #79 so queue is empty."
|
||||
)
|
||||
result = assess_empty_queue_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("weak corroboration" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_author_session_reviewer_queue_wording_blocked(self):
|
||||
result = assess_empty_queue_report(
|
||||
self._trusted_report(),
|
||||
task_role="author",
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_build_final_report_downgrades_weak_empty_queue(self):
|
||||
final = build_final_report(
|
||||
checkout_proof=_good_checkout(),
|
||||
inventory=_good_inventory(),
|
||||
validation=_good_validation(),
|
||||
contamination=_good_contamination(),
|
||||
identity_eligible=True,
|
||||
merge_performed=False,
|
||||
issue_status_verified=True,
|
||||
capability_evidence=_good_capability_evidence(),
|
||||
sweep=_good_sweep(),
|
||||
live_state=_good_live_state(),
|
||||
role_boundary=_good_role_boundary(),
|
||||
review_mutation=_good_review_mutation(),
|
||||
controller_handoff=_good_handoff(),
|
||||
capability_proof=_good_capability_proof(),
|
||||
sweep_proof=_good_secret_sweep(),
|
||||
worktree_proof={
|
||||
"worktree_path": "/repo/branches/review-pr-1",
|
||||
"porcelain_status": "",
|
||||
"scratch_used": True,
|
||||
"scratch_path": "/repo/branches/review-pr-1",
|
||||
},
|
||||
report_text="Open PR count: 0. Queue cleared.",
|
||||
)
|
||||
self.assertNotEqual(final["grade"], "A")
|
||||
self.assertFalse(final["empty_queue_trust_gate_proven"])
|
||||
|
||||
|
||||
class TestCapabilityEvidence(unittest.TestCase):
|
||||
"""#179 gap 1: capability claims need exact evidence."""
|
||||
|
||||
@@ -1519,16 +1207,6 @@ class TestFinalReport179Bar(unittest.TestCase):
|
||||
"live_state": _good_live_state(),
|
||||
"role_boundary": _good_role_boundary(),
|
||||
"review_mutation": _good_review_mutation(),
|
||||
"controller_handoff": _good_handoff(),
|
||||
"capability_proof": _good_capability_proof(),
|
||||
"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)
|
||||
return build_final_report(**kwargs)
|
||||
@@ -1604,323 +1282,5 @@ class TestFinalReport179Bar(unittest.TestCase):
|
||||
self.assertFalse(report["review_mutation_complete"])
|
||||
|
||||
|
||||
class TestAuthorReporting(unittest.TestCase):
|
||||
"""Harness assertions for author reporting and capability/sweep proofs."""
|
||||
|
||||
def test_good_capability_proof_passes(self):
|
||||
result = _good_capability_proof()
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_missing_capability_proof_fails_closed(self):
|
||||
result = assess_capability_proof({})
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("fail closed" in r for r in result["reasons"]))
|
||||
|
||||
def test_unresolved_capability_proof_fails_closed(self):
|
||||
result = assess_capability_proof({
|
||||
"mark_issue": {
|
||||
"requested_task": "unknown_task",
|
||||
"required_operation_permission": None,
|
||||
"allowed_in_current_session": None,
|
||||
}
|
||||
})
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("could not be resolved" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_good_secret_sweep_passes(self):
|
||||
result = _good_secret_sweep()
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertEqual(result["verdict"], "strong")
|
||||
|
||||
def test_vague_secret_sweep_without_method_is_weak(self):
|
||||
result = _good_secret_sweep(method="")
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertEqual(result["verdict"], "weak")
|
||||
self.assertTrue(
|
||||
any("method" in r or "scan" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_unconfirmed_secret_sweep_is_weak(self):
|
||||
result = _good_secret_sweep(clean=False)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertEqual(result["verdict"], "weak")
|
||||
|
||||
def test_good_author_pr_report_passes(self):
|
||||
result = assess_author_pr_report({
|
||||
"pr_number": 185,
|
||||
"branch": "feat/issue-184-repo-name-disambiguation",
|
||||
"head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae",
|
||||
})
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_author_pr_report_without_head_sha_is_incomplete(self):
|
||||
result = assess_author_pr_report({
|
||||
"pr_number": 185,
|
||||
"branch": "feat/issue-184-repo-name-disambiguation",
|
||||
"head_sha": "",
|
||||
})
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(any("head SHA" in r for r in result["reasons"]))
|
||||
|
||||
def test_author_pr_report_rejects_short_sha(self):
|
||||
result = assess_author_pr_report({
|
||||
"pr_number": 185,
|
||||
"branch": "feat/issue-184-repo-name-disambiguation",
|
||||
"head_sha": "e2bccba",
|
||||
})
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
|
||||
class TestIssueSelectionContinuation(unittest.TestCase):
|
||||
"""Issue #188: continuation mode wall for issues with open PRs."""
|
||||
|
||||
OLD_SHA = PINNED
|
||||
NEW_SHA = OTHER
|
||||
OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
|
||||
|
||||
def test_open_pr_issue_excluded_from_fresh_selection(self):
|
||||
classified = classify_issue_for_selection(
|
||||
183, open_prs=self.OPEN_PR,
|
||||
)
|
||||
self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR)
|
||||
self.assertFalse(classified["selectable_for_fresh_work"])
|
||||
blocked = assess_fresh_issue_selection([classified])
|
||||
self.assertTrue(blocked["downgraded"])
|
||||
|
||||
def test_explicit_continuation_allows_represented_issue(self):
|
||||
classified = classify_issue_for_selection(
|
||||
183,
|
||||
open_prs=self.OPEN_PR,
|
||||
operator_continuation_requested=True,
|
||||
)
|
||||
self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT)
|
||||
blocked = assess_fresh_issue_selection([classified])
|
||||
self.assertFalse(blocked["downgraded"])
|
||||
|
||||
def test_contradictory_no_pr_claim_downgrades(self):
|
||||
report = (
|
||||
"Selected issue #183; no duplicate PR open. "
|
||||
"Updated PR #187 on branch feat/issue-183-harden-author-run-reporting."
|
||||
)
|
||||
result = assess_contradictory_no_pr_claim(
|
||||
report, edited_pr_numbers=[187], issue_open_pr_map={183: 187},
|
||||
)
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
def test_edited_pr_must_appear_in_inventory(self):
|
||||
report = "Open PR inventory: PR #195 only."
|
||||
result = assess_edited_pr_inventory_coverage(
|
||||
report,
|
||||
edited_pr_numbers=[187],
|
||||
inventoried_pr_numbers=[195],
|
||||
)
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
def test_continuation_report_requires_old_and_new_head(self):
|
||||
report = (
|
||||
"Issue #182 continuation mode. PR #186. "
|
||||
f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. "
|
||||
"PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. "
|
||||
"Session authored PR: yes. Continuation allowed: operator requested."
|
||||
)
|
||||
result = assess_continuation_mode_report(
|
||||
report,
|
||||
pr_number=186,
|
||||
pr_author="jcwalker3",
|
||||
branch="feat/issue-182-controller-handoff",
|
||||
old_head_sha=self.OLD_SHA,
|
||||
new_head_sha=self.NEW_SHA,
|
||||
session_authored_pr=True,
|
||||
continuation_allowed_reason="operator requested continuation",
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_issue_selection_final_report_continuation_earns_a(self):
|
||||
report = "\n".join([
|
||||
"Issue #182 continuation mode — no new issue claimed.",
|
||||
f"PR #186 updated: old head {self.OLD_SHA}, "
|
||||
f"new head {self.NEW_SHA}.",
|
||||
"PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.",
|
||||
"Session authored PR: yes.",
|
||||
"Continuation allowed: operator requested rebase.",
|
||||
"Open PR inventory included PR #186.",
|
||||
"## Controller Handoff",
|
||||
"- Task: continuation",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: author",
|
||||
"- Identity: prgs-author",
|
||||
"- Issue/PR: #182 / PR #186",
|
||||
"- Branch/SHA: feat/issue-182-controller-handoff",
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: tests passed",
|
||||
"- Mutations: push_branch",
|
||||
"- Workspace mutations: none",
|
||||
"- Current status: PR mergeable",
|
||||
"- Blockers: none",
|
||||
"- Next: review",
|
||||
"- Safety: no review/merge",
|
||||
"- Continuation mode: issue #182 continuation",
|
||||
"- Existing PR: #186",
|
||||
"- PR author: jcwalker3",
|
||||
"- Issue claim status: status:in-progress",
|
||||
"- Branch: feat/issue-182-controller-handoff",
|
||||
f"- Old PR head: {self.OLD_SHA}",
|
||||
f"- New PR head: {self.NEW_SHA}",
|
||||
"- Session authored PR: yes",
|
||||
"- Why continuation allowed: operator requested rebase",
|
||||
])
|
||||
result = assess_issue_selection_final_report(
|
||||
report,
|
||||
mode="continuation",
|
||||
continuation_proof={
|
||||
"pr_number": 186,
|
||||
"pr_author": "jcwalker3",
|
||||
"issue_number": 182,
|
||||
"issue_claim_status": "status:in-progress",
|
||||
"branch": "feat/issue-182-controller-handoff",
|
||||
"old_head_sha": self.OLD_SHA,
|
||||
"new_head_sha": self.NEW_SHA,
|
||||
"session_authored_pr": True,
|
||||
"continuation_allowed_reason": "operator requested rebase",
|
||||
},
|
||||
edited_pr_numbers=[186],
|
||||
inventoried_pr_numbers=[186],
|
||||
)
|
||||
self.assertEqual(result["grade"], "A")
|
||||
|
||||
|
||||
class TestContinuationModeProofs(unittest.TestCase):
|
||||
"""Issue #189: formal continuation proofs beyond #188 selection wall."""
|
||||
|
||||
OLD_SHA = PINNED
|
||||
NEW_SHA = OTHER
|
||||
REMOTE_OLD = "601c608c00000000000000000000000000000000"
|
||||
LEASE_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
|
||||
PUSHED_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
|
||||
|
||||
def test_force_with_lease_requires_lease_evidence(self):
|
||||
report = (
|
||||
"Issue #182 continuation. force-with-lease push to "
|
||||
"feat/issue-182-controller-handoff-enforcement."
|
||||
)
|
||||
result = assess_force_with_lease_push_report(
|
||||
report,
|
||||
old_remote_head=self.REMOTE_OLD,
|
||||
expected_lease_head=self.LEASE_SHA,
|
||||
new_pushed_head=self.PUSHED_SHA,
|
||||
branch_pushed="feat/issue-182-controller-handoff-enforcement",
|
||||
used_force_with_lease=True,
|
||||
)
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
def test_force_with_lease_complete_with_full_evidence(self):
|
||||
report = (
|
||||
"Issue #182 continuation with git push --force-with-lease. "
|
||||
f"Old remote head {self.REMOTE_OLD}. "
|
||||
f"Lease head {self.LEASE_SHA}. "
|
||||
f"Pushed head {self.PUSHED_SHA}. "
|
||||
"Branch feat/issue-182-controller-handoff-enforcement. "
|
||||
"Push branch only: only the feature branch was pushed."
|
||||
)
|
||||
result = assess_force_with_lease_push_report(
|
||||
report,
|
||||
old_remote_head=self.REMOTE_OLD,
|
||||
expected_lease_head=self.LEASE_SHA,
|
||||
new_pushed_head=self.PUSHED_SHA,
|
||||
branch_pushed="feat/issue-182-controller-handoff-enforcement",
|
||||
used_force_with_lease=True,
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_continuation_requires_issue_claim_status(self):
|
||||
report = (
|
||||
"Issue #182 continuation mode. PR #186. "
|
||||
f"old head {self.OLD_SHA} new head {self.NEW_SHA}. "
|
||||
"PR author jcwalker3. Branch feat/issue-182-test."
|
||||
)
|
||||
result = assess_continuation_mode_report(
|
||||
report,
|
||||
pr_number=186,
|
||||
pr_author="jcwalker3",
|
||||
issue_number=182,
|
||||
issue_claim_status="status:in-progress",
|
||||
branch="feat/issue-182-test",
|
||||
old_head_sha=self.OLD_SHA,
|
||||
new_head_sha=self.NEW_SHA,
|
||||
)
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
def test_canonical_secret_sweep_helper(self):
|
||||
sweep = canonical_secret_sweep_report(clean=True)
|
||||
self.assertEqual(sweep["method"], CANONICAL_SECRET_SWEEP_COMMAND)
|
||||
self.assertTrue(sweep["clean"])
|
||||
|
||||
def test_continuation_final_report_with_push_and_sweep_earns_a(self):
|
||||
branch = "feat/issue-182-controller-handoff-enforcement"
|
||||
report = "\n".join([
|
||||
"Issue #182 continuation mode — status:in-progress already set.",
|
||||
f"PR #186 updated via git push --force-with-lease.",
|
||||
f"Old PR head {self.OLD_SHA}; new PR head {self.NEW_SHA}.",
|
||||
f"Old remote head {self.REMOTE_OLD}.",
|
||||
f"Lease head {self.LEASE_SHA}. Pushed head {self.PUSHED_SHA}.",
|
||||
f"Branch {branch}. Push branch only: only the feature branch.",
|
||||
"PR author: jcwalker3. Session authored PR: yes.",
|
||||
"Continuation allowed: operator requested rebase.",
|
||||
"Open PR inventory included PR #186.",
|
||||
"## Controller Handoff",
|
||||
"- Task: continuation",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: author",
|
||||
"- Identity: prgs-author",
|
||||
"- Issue/PR: #182 / PR #186",
|
||||
f"- Branch/SHA: {branch}",
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: tests passed",
|
||||
"- Mutations: push_branch",
|
||||
"- Workspace mutations: none",
|
||||
"- Current status: PR mergeable",
|
||||
"- Blockers: none",
|
||||
"- Next: review",
|
||||
"- Safety: no review/merge",
|
||||
"- Continuation mode: issue #182 continuation",
|
||||
"- Existing PR: #186",
|
||||
"- PR author: jcwalker3",
|
||||
"- Issue claim status: status:in-progress",
|
||||
f"- Branch: {branch}",
|
||||
f"- Old PR head: {self.OLD_SHA}",
|
||||
f"- New PR head: {self.NEW_SHA}",
|
||||
"- Session authored PR: yes",
|
||||
"- Why continuation allowed: operator requested rebase",
|
||||
])
|
||||
result = assess_issue_selection_final_report(
|
||||
report,
|
||||
mode="continuation",
|
||||
continuation_proof={
|
||||
"pr_number": 186,
|
||||
"pr_author": "jcwalker3",
|
||||
"issue_number": 182,
|
||||
"issue_claim_status": "status:in-progress",
|
||||
"branch": branch,
|
||||
"old_head_sha": self.OLD_SHA,
|
||||
"new_head_sha": self.NEW_SHA,
|
||||
"session_authored_pr": True,
|
||||
"continuation_allowed_reason": "operator requested rebase",
|
||||
"push_proof": {
|
||||
"old_remote_head": self.REMOTE_OLD,
|
||||
"expected_lease_head": self.LEASE_SHA,
|
||||
"new_pushed_head": self.PUSHED_SHA,
|
||||
"branch_pushed": branch,
|
||||
"used_force_with_lease": True,
|
||||
},
|
||||
"secret_sweep": canonical_secret_sweep_report(clean=True),
|
||||
},
|
||||
edited_pr_numbers=[186],
|
||||
inventoried_pr_numbers=[186],
|
||||
)
|
||||
self.assertEqual(result["grade"], "A")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -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()
|
||||
@@ -1,215 +0,0 @@
|
||||
"""Tests for reviewer/author namespace mismatch walls (#209)."""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_audit
|
||||
import mcp_server
|
||||
import role_namespace_gate
|
||||
from review_proofs import assess_mutation_namespace_handoff
|
||||
|
||||
CONFIG = {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"ctx": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"prgs-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "jcwalker3",
|
||||
"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", "gitea.pr.review",
|
||||
],
|
||||
"execution_profile": "prgs-author",
|
||||
},
|
||||
"prgs-reviewer": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "reviewer",
|
||||
"username": "sysadmin",
|
||||
"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", "gitea.issue.create",
|
||||
],
|
||||
"execution_profile": "prgs-reviewer",
|
||||
},
|
||||
"prgs-reviewer-issue-writer": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "reviewer",
|
||||
"username": "sysadmin",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.pr.review", "gitea.issue.create",
|
||||
],
|
||||
"forbidden_operations": ["gitea.pr.create"],
|
||||
"execution_profile": "prgs-reviewer-issue-writer",
|
||||
},
|
||||
},
|
||||
"rules": {"allow_runtime_switching": False},
|
||||
}
|
||||
|
||||
ISSUE_WRITE_ENV = {"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create"}
|
||||
|
||||
|
||||
class NamespaceGateBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {
|
||||
"host": "gitea.example.com",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Gitea-Tools",
|
||||
},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
self.audit_path = os.path.join(self._dir.name, "audit.log")
|
||||
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
self._dir.cleanup()
|
||||
|
||||
def _env(self, profile, *, audit=False):
|
||||
env = {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
||||
}
|
||||
if audit:
|
||||
env["GITEA_AUDIT_LOG"] = self.audit_path
|
||||
return env
|
||||
|
||||
|
||||
class TestNamespaceInference(unittest.TestCase):
|
||||
def test_reviewer_and_author_namespaces(self):
|
||||
self.assertEqual(
|
||||
role_namespace_gate.infer_mcp_namespace("prgs-reviewer"),
|
||||
"gitea-reviewer",
|
||||
)
|
||||
self.assertEqual(
|
||||
role_namespace_gate.infer_mcp_namespace("prgs-author"),
|
||||
"gitea-author",
|
||||
)
|
||||
|
||||
|
||||
class TestReviewerAuthorMutationBlocks(NamespaceGateBase):
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_reviewer_namespace_create_issue_blocked(
|
||||
self, _auth, mock_api, _get_all, _role
|
||||
):
|
||||
with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True):
|
||||
result = mcp_server.gitea_create_issue(
|
||||
title="should not post", remote="prgs")
|
||||
self.assertFalse(result["success"])
|
||||
self.assertFalse(result["performed"])
|
||||
self.assertTrue(result.get("namespace_block"))
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_reviewer_namespace_create_pr_blocked(
|
||||
self, _auth, mock_api, _get_all, _role
|
||||
):
|
||||
with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True):
|
||||
result = mcp_server.gitea_create_pr(
|
||||
title="feat: x Closes #1",
|
||||
head="feat/issue-1-x",
|
||||
body="Closes #1",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(result.get("success", True))
|
||||
self.assertFalse(result.get("performed", True))
|
||||
self.assertTrue(result.get("namespace_block"))
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.api_request", return_value={"number": 12})
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_reviewer_with_explicit_issue_create_allowed(
|
||||
self, _auth, mock_api, _get_all, _role
|
||||
):
|
||||
with patch.dict(os.environ, self._env("prgs-reviewer-issue-writer"),
|
||||
clear=True):
|
||||
result = mcp_server.gitea_create_issue(
|
||||
title="escalation issue", remote="prgs")
|
||||
self.assertEqual(result.get("number"), 12)
|
||||
mock_api.assert_called()
|
||||
|
||||
|
||||
class TestHandoffNamespaceParity(unittest.TestCase):
|
||||
def test_author_handoff_with_reviewer_namespace_invalid(self):
|
||||
result = assess_mutation_namespace_handoff(
|
||||
declared_role="author",
|
||||
namespaces_used=["gitea-reviewer"],
|
||||
)
|
||||
self.assertFalse(result["valid"])
|
||||
self.assertTrue(result["blocked"])
|
||||
|
||||
def test_capability_author_profile_with_reviewer_namespace_blocked(self):
|
||||
result = assess_mutation_namespace_handoff(
|
||||
declared_role="author",
|
||||
namespaces_used=["gitea-reviewer"],
|
||||
capability_profile="prgs-author",
|
||||
)
|
||||
self.assertFalse(result["valid"])
|
||||
blob = " ".join(result["reasons"])
|
||||
self.assertIn("capability proof profile", blob)
|
||||
|
||||
|
||||
class TestMutationAuditFields(NamespaceGateBase):
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.api_request", return_value={"number": 3})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_successful_create_issue_audit_includes_namespace_fields(
|
||||
self, _auth, mock_api, _get_all, _role
|
||||
):
|
||||
env = {**self._env("prgs-author", audit=True), **ISSUE_WRITE_ENV}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
mcp_server.gitea_create_issue(title="audited", remote="prgs")
|
||||
with open(self.audit_path, encoding="utf-8") as fh:
|
||||
record = json.loads(fh.readline())
|
||||
self.assertEqual(record["mcp_namespace"], "gitea-author")
|
||||
self.assertEqual(record["task_role"], "author")
|
||||
self.assertEqual(record["operation"], "create_issue")
|
||||
self.assertEqual(record["profile_name"], "prgs-author")
|
||||
self.assertEqual(record["remote"], "prgs")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,112 +0,0 @@
|
||||
"""Checks for the Gitea Wiki sync helper (issue #224).
|
||||
|
||||
The repo-tracked ``docs/wiki/`` stays the source of truth; the helper
|
||||
mirrors it into the Gitea native wiki only when explicitly invoked.
|
||||
"""
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
SCRIPT = REPO_ROOT / "scripts" / "sync-gitea-wiki.sh"
|
||||
|
||||
REQUIRED_PAGES = [
|
||||
"Home.md",
|
||||
"Operator-Guide.md",
|
||||
"Repositories.md",
|
||||
"Identity-and-Profiles.md",
|
||||
"Workflow.md",
|
||||
"Safety-and-Gates.md",
|
||||
"MCP-Tools.md",
|
||||
"Runbooks.md",
|
||||
"Open-Decisions.md",
|
||||
"History.md",
|
||||
]
|
||||
|
||||
|
||||
def _run(*args, env_extra=None):
|
||||
env = dict(os.environ)
|
||||
env.pop("GITEA_WIKI_SYNC_CONFIRM", None)
|
||||
if env_extra:
|
||||
env.update(env_extra)
|
||||
return subprocess.run(
|
||||
[str(SCRIPT), *args], capture_output=True, text=True, env=env,
|
||||
cwd=str(REPO_ROOT), timeout=30)
|
||||
|
||||
|
||||
def test_script_exists_and_is_executable():
|
||||
assert SCRIPT.is_file(), "missing scripts/sync-gitea-wiki.sh"
|
||||
assert os.access(SCRIPT, os.X_OK), "sync script is not executable"
|
||||
|
||||
|
||||
def test_dry_run_is_default_and_lists_every_page():
|
||||
result = _run()
|
||||
assert result.returncode == 0, result.stderr
|
||||
out = result.stdout
|
||||
assert "dry-run" in out.lower()
|
||||
for page in REQUIRED_PAGES:
|
||||
assert page in out, f"dry-run plan does not list {page}"
|
||||
|
||||
|
||||
def test_dry_run_does_not_push_or_clone():
|
||||
result = _run()
|
||||
blob = (result.stdout + result.stderr).lower()
|
||||
assert "pushing" not in blob
|
||||
assert "cloning" not in blob
|
||||
|
||||
|
||||
def test_push_refused_without_exact_confirmation():
|
||||
result = _run("--push")
|
||||
assert result.returncode != 0
|
||||
blob = result.stdout + result.stderr
|
||||
assert "GITEA_WIKI_SYNC_CONFIRM" in blob
|
||||
assert "cloning" not in blob.lower()
|
||||
|
||||
|
||||
def test_push_refused_with_wrong_confirmation():
|
||||
result = _run("--push", env_extra={"GITEA_WIKI_SYNC_CONFIRM": "yes please"})
|
||||
assert result.returncode != 0
|
||||
assert "cloning" not in (result.stdout + result.stderr).lower()
|
||||
|
||||
|
||||
def test_script_embeds_no_service_urls_or_secrets():
|
||||
text = SCRIPT.read_text(encoding="utf-8")
|
||||
for marker in ("http://", "https://", "token ", "Authorization", "ghp_", "keychain"):
|
||||
assert marker not in text, f"script embeds {marker!r}"
|
||||
|
||||
|
||||
def test_runbook_documents_wiki_sync():
|
||||
runbooks = (REPO_ROOT / "docs" / "wiki" / "Runbooks.md").read_text(encoding="utf-8")
|
||||
assert "sync-gitea-wiki" in runbooks
|
||||
home = (REPO_ROOT / "docs" / "wiki" / "Home.md").read_text(encoding="utf-8")
|
||||
assert "source of truth" in home
|
||||
template = (REPO_ROOT / ".gitea" / "pull_request_template.md").read_text(encoding="utf-8")
|
||||
assert "sync-gitea-wiki" in template
|
||||
|
||||
|
||||
def test_wiki_pages_have_no_local_file_links():
|
||||
for path in (REPO_ROOT / "docs" / "wiki").glob("*.md"):
|
||||
text = path.read_text(encoding="utf-8")
|
||||
assert "file://" not in text, f"{path.name} contains a file:// link"
|
||||
|
||||
|
||||
def test_pr_template_carries_wiki_publication_gate():
|
||||
template = (REPO_ROOT / ".gitea" / "pull_request_template.md").read_text(encoding="utf-8")
|
||||
assert "Wiki tab" in template
|
||||
assert "#224" in template
|
||||
assert "not sufficient to close a wiki issue" in template
|
||||
|
||||
|
||||
def test_runbooks_document_publication_gate_and_closure_rule():
|
||||
runbooks = (REPO_ROOT / "docs" / "wiki" / "Runbooks.md").read_text(encoding="utf-8")
|
||||
assert "Wiki Publication Readiness Gate (#224)" in runbooks or "Wiki publication readiness gate (#224)" in runbooks.lower()
|
||||
assert "Closure prevention" in runbooks
|
||||
assert "GITEA_WIKI_SYNC_CONFIRM" in runbooks
|
||||
|
||||
|
||||
def test_repositories_map_tracks_wiki_publication_status():
|
||||
repositories = (REPO_ROOT / "docs" / "wiki" / "Repositories.md").read_text(encoding="utf-8")
|
||||
assert "Wiki publication status" in repositories or "Wiki Publication Status" in repositories
|
||||
for repo in ("mcp-control-plane", "Gitea-Tools"):
|
||||
assert repo in repositories
|
||||
assert "Gitea Wiki published" in repositories
|
||||
Reference in New Issue
Block a user