Compare commits
12 Commits
4e43347b2d
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 2d5cb4bb29 | |||
| 1441591e74 | |||
| 848a4294ba | |||
| 496e796cdd | |||
| 642adf4705 | |||
| e842b60ad8 | |||
| 3a246ab553 | |||
| dbfa0fe188 | |||
| b3728c54ce | |||
| 4afada098c | |||
| 6089ec724a | |||
| c6c6e75af6 |
@@ -382,3 +382,35 @@ python3 -m pytest tests/ -v
|
||||
| `test_mirror_refs.py` | Flags, safety defaults, local integration tests |
|
||||
|
||||
All tests mock network and keychain access — no real API calls are made.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### macOS: `com.apple.provenance` blocks Python execution (#3)
|
||||
|
||||
On macOS Sequoia and later, files written by an agent/IDE terminal receive the
|
||||
`com.apple.provenance` extended attribute, and macOS blocks `Python.app` from
|
||||
**executing** such files. Symptoms: newly created/restored `.py` files fail to
|
||||
run (e.g. `create_issue.py` "vanishing" or refusing to execute), while shell
|
||||
scripts and files created before the session are unaffected. This is a macOS
|
||||
security feature, not a bug in this project's code.
|
||||
|
||||
Workarounds (run from a terminal with **Full Disk Access**, e.g. `Terminal.app`
|
||||
— not the IDE terminal, or the removal itself may be blocked):
|
||||
|
||||
```bash
|
||||
# Preferred: strip only com.apple.provenance under the repo (dry-run first)
|
||||
./scripts/clear-provenance --dry-run
|
||||
./scripts/clear-provenance
|
||||
|
||||
# Or a single file
|
||||
./scripts/clear-provenance /path/to/file.py
|
||||
|
||||
# Manual equivalents
|
||||
xattr -r -d com.apple.provenance /Users/jasonwalker/Development/Gitea-Tools/
|
||||
xattr -cr /Users/jasonwalker/Development/Gitea-Tools/ # clears ALL xattrs
|
||||
```
|
||||
|
||||
Alternatively, grant Full Disk Access to the terminal app in
|
||||
**System Settings → Privacy & Security**. `scripts/clear-provenance` removes only
|
||||
`com.apple.provenance` (leaving other extended attributes intact) and supports
|
||||
`--dry-run`.
|
||||
|
||||
+78
-17
@@ -4,9 +4,14 @@
|
||||
Auth follows the project convention: credentials are pulled from the macOS
|
||||
keychain via `git credential fill` (HTTPS), then sent as Basic auth.
|
||||
|
||||
Usage:
|
||||
./manage_labels.py # create labels, then apply the mapping below
|
||||
./manage_labels.py --dry # print actions without writing
|
||||
Modes (default = create labels then apply the one-off MAPPING, preserving the
|
||||
original behavior):
|
||||
|
||||
./manage_labels.py # create labels + apply MAPPING
|
||||
./manage_labels.py --create-labels # idempotent label creation only
|
||||
./manage_labels.py --apply-mapping # one-off MAPPING labeling only
|
||||
./manage_labels.py --add-label 42 chore # add one label to one issue
|
||||
./manage_labels.py --dry ... # print actions without writing
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
@@ -34,7 +39,7 @@ LABELS = [
|
||||
"description": "Issue is being worked on"},
|
||||
]
|
||||
|
||||
# issue number -> label names to apply
|
||||
# issue number -> label names to apply (one-off backfill)
|
||||
MAPPING = {
|
||||
23: ["chore"],
|
||||
22: ["chore"],
|
||||
@@ -56,6 +61,11 @@ MAPPING = {
|
||||
|
||||
BASE_URL = repo_api_url(HOST, ORG, REPO)
|
||||
|
||||
USAGE = (
|
||||
"usage: manage_labels.py [--dry] "
|
||||
"[--create-labels | --apply-mapping | --add-label <issue> <label>]"
|
||||
)
|
||||
|
||||
|
||||
def api(method, path, auth, payload=None):
|
||||
"""Thin wrapper around auth.api_request that prepends BASE_URL and
|
||||
@@ -68,19 +78,15 @@ def api(method, path, auth, payload=None):
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
dry = "--dry" in sys.argv
|
||||
auth = get_auth_header(HOST)
|
||||
if auth is None:
|
||||
print("Could not get credentials from git credential fill",
|
||||
file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# 1. Existing labels -> name:id
|
||||
def _labels_by_name(auth):
|
||||
"""Return {label name: id} for the repo's existing labels."""
|
||||
existing = api("GET", "/labels?limit=100", auth) or []
|
||||
by_name = {l["name"]: l["id"] for l in existing}
|
||||
return {lb["name"]: lb["id"] for lb in existing}
|
||||
|
||||
# 2. Create missing labels
|
||||
|
||||
def create_labels(auth, dry=False):
|
||||
"""Idempotently create the LABELS set; return the resulting name->id map."""
|
||||
by_name = _labels_by_name(auth)
|
||||
for spec in LABELS:
|
||||
if spec["name"] in by_name:
|
||||
print(f"label exists: {spec['name']}")
|
||||
@@ -92,8 +98,13 @@ def main():
|
||||
if created:
|
||||
by_name[created["name"]] = created["id"]
|
||||
print(f"created label: {created['name']} (id {created['id']})")
|
||||
return by_name
|
||||
|
||||
# 3. Apply mapping
|
||||
|
||||
def apply_mapping(auth, by_name=None, dry=False):
|
||||
"""Apply the one-off MAPPING (PUT replaces each issue's label set)."""
|
||||
if by_name is None:
|
||||
by_name = _labels_by_name(auth)
|
||||
for issue, names in sorted(MAPPING.items(), reverse=True):
|
||||
ids = [by_name[n] for n in names if n in by_name]
|
||||
missing = [n for n in names if n not in by_name]
|
||||
@@ -105,9 +116,59 @@ def main():
|
||||
# PUT replaces the issue's labels with exactly this set (idempotent).
|
||||
res = api("PUT", f"/issues/{issue}/labels", auth, {"labels": ids})
|
||||
if res is not None:
|
||||
applied = [l["name"] for l in res]
|
||||
applied = [lb["name"] for lb in res]
|
||||
print(f"#{issue} labeled: {applied}")
|
||||
|
||||
|
||||
def add_label(auth, issue, label, dry=False):
|
||||
"""Ad-hoc: ADD a single existing label to one issue (append, not replace)."""
|
||||
by_name = _labels_by_name(auth)
|
||||
if label not in by_name:
|
||||
print(f" unknown label '{label}'; create it first (--create-labels)",
|
||||
file=sys.stderr)
|
||||
return False
|
||||
if dry:
|
||||
print(f"[dry] #{issue} += {label}")
|
||||
return True
|
||||
# POST appends to the issue's existing labels (does not replace).
|
||||
res = api("POST", f"/issues/{issue}/labels", auth, {"labels": [by_name[label]]})
|
||||
if res is not None:
|
||||
print(f"#{issue} += {label}")
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
argv = list(sys.argv[1:] if argv is None else argv)
|
||||
dry = "--dry" in argv or "--dry-run" in argv
|
||||
|
||||
auth = get_auth_header(HOST)
|
||||
if auth is None:
|
||||
print("Could not get credentials from git credential fill",
|
||||
file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if "--create-labels" in argv:
|
||||
create_labels(auth, dry=dry)
|
||||
elif "--apply-mapping" in argv:
|
||||
apply_mapping(auth, dry=dry)
|
||||
elif "--add-label" in argv:
|
||||
i = argv.index("--add-label")
|
||||
if i + 2 >= len(argv):
|
||||
print(USAGE, file=sys.stderr)
|
||||
sys.exit(2)
|
||||
try:
|
||||
issue = int(argv[i + 1])
|
||||
except ValueError:
|
||||
print(f"--add-label: issue must be a number, got '{argv[i + 1]}'",
|
||||
file=sys.stderr)
|
||||
sys.exit(2)
|
||||
add_label(auth, issue, argv[i + 2], dry=dry)
|
||||
else:
|
||||
# Default (backward compatible): create labels, then apply the mapping.
|
||||
by_name = create_labels(auth, dry=dry)
|
||||
apply_mapping(auth, by_name, dry=dry)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
+111
-2
@@ -14,6 +14,7 @@ Configuration (mcp_config.json):
|
||||
}
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import functools
|
||||
import contextlib
|
||||
@@ -48,6 +49,71 @@ mcp = FastMCP("gitea-tools", instructions=(
|
||||
))
|
||||
|
||||
|
||||
def extract_linked_issue_numbers(text: str | None, branch_name: str | None = None) -> list[int]:
|
||||
issues = set()
|
||||
if text:
|
||||
pattern = re.compile(r'(?i)(?:close[sd]?|fix(?:e[sd])?|resolve[sd]?|ref[s]?)\s+#(\d+)')
|
||||
issues.update(int(m) for m in pattern.findall(text))
|
||||
if branch_name:
|
||||
pattern = re.compile(r'(?i)issue-(\d+)')
|
||||
issues.update(int(m) for m in pattern.findall(branch_name))
|
||||
return sorted(list(issues))
|
||||
|
||||
def release_in_progress_label(issue_numbers: list[int], remote: str, host: str | None, org: str | None, repo: str | None) -> dict:
|
||||
if not issue_numbers:
|
||||
return {}
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
|
||||
try:
|
||||
labels = api_request("GET", f"{base}/labels?limit=100", auth)
|
||||
label_id = None
|
||||
for lb in labels:
|
||||
if lb["name"] == "status:in-progress":
|
||||
label_id = lb["id"]
|
||||
break
|
||||
except Exception as exc:
|
||||
return {num: f"error fetching repo labels: {_redact(str(exc))}" for num in issue_numbers}
|
||||
|
||||
results = {}
|
||||
if label_id is None:
|
||||
for num in issue_numbers:
|
||||
results[num] = "not present"
|
||||
return results
|
||||
|
||||
for num in issue_numbers:
|
||||
try:
|
||||
url = f"{base}/issues/{num}"
|
||||
issue_data = api_request("GET", url, auth)
|
||||
issue_labels = [lb["name"] for lb in issue_data.get("labels", [])]
|
||||
|
||||
if "status:in-progress" in issue_labels:
|
||||
with _audited("release_in_progress_label", host=h, remote=remote, org=o, repo=r, issue_number=num, request_metadata={"action": "remove status:in-progress"}):
|
||||
api_request("DELETE", f"{url}/labels/{label_id}", auth)
|
||||
results[num] = "released"
|
||||
else:
|
||||
results[num] = "not present"
|
||||
except Exception as exc:
|
||||
results[num] = f"error: {_redact(str(exc))}"
|
||||
|
||||
return results
|
||||
|
||||
def cleanup_in_progress_for_pr(pr_payload: dict, remote: str, host: str | None, org: str | None, repo: str | None) -> dict:
|
||||
body = pr_payload.get("body") or ""
|
||||
title = pr_payload.get("title") or ""
|
||||
branch = pr_payload.get("head", {}).get("ref") or ""
|
||||
|
||||
text = f"{title}\n{body}"
|
||||
issues = extract_linked_issue_numbers(text, branch)
|
||||
|
||||
if not issues:
|
||||
return {"cleanup_status": "no linked issue found"}
|
||||
|
||||
results = release_in_progress_label(issues, remote, host, org, repo)
|
||||
return {"cleanup_status": results}
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
|
||||
@@ -743,6 +809,20 @@ def gitea_edit_pr(
|
||||
with _audited("edit_pr", host=h, remote=remote, org=o, repo=r,
|
||||
pr_number=pr_number, request_metadata={"fields": sorted(payload)}):
|
||||
data = api_request("PATCH", url, auth, payload)
|
||||
|
||||
cleanup_status = None
|
||||
if state == "closed":
|
||||
cleanup = cleanup_in_progress_for_pr(data, remote, host, org, repo)
|
||||
cleanup_status = cleanup.get("cleanup_status")
|
||||
if isinstance(cleanup_status, dict):
|
||||
for issue_num, st in cleanup_status.items():
|
||||
if st == "released":
|
||||
try:
|
||||
comment_url = f"{repo_api_url(h, o, r)}/issues/{issue_num}/comments"
|
||||
api_request("POST", comment_url, auth, {"body": f"Tracker cleanup: removed `status:in-progress` from this issue because linked PR #{pr_number} was closed."})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"number": data["number"],
|
||||
@@ -750,6 +830,7 @@ def gitea_edit_pr(
|
||||
"body": data.get("body", ""),
|
||||
"state": data["state"],
|
||||
"url": data["html_url"],
|
||||
"cleanup_status": cleanup_status,
|
||||
}
|
||||
|
||||
|
||||
@@ -820,7 +901,7 @@ def gitea_commit_files(
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
url = f"{repo_api_url(h, o, r)}/contents"
|
||||
|
||||
|
||||
payload = {
|
||||
"files": files,
|
||||
"message": message,
|
||||
@@ -1021,6 +1102,9 @@ def gitea_merge_pr(
|
||||
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth
|
||||
)
|
||||
result["merge_commit"] = (merged or {}).get("merged_commit_sha")
|
||||
|
||||
cleanup = cleanup_in_progress_for_pr(merged or {}, remote, host, org, repo)
|
||||
result["cleanup_status"] = cleanup.get("cleanup_status")
|
||||
except Exception:
|
||||
result["merge_commit"] = None
|
||||
except Exception as exc: # noqa: BLE001 — redact before surfacing
|
||||
@@ -1157,7 +1241,14 @@ def gitea_close_issue(
|
||||
with _audited("close_issue", host=h, remote=remote, org=o, repo=r,
|
||||
issue_number=issue_number, request_metadata={"state": "closed"}):
|
||||
api_request("PATCH", url, auth, {"state": "closed"})
|
||||
return {"success": True, "message": f"Issue #{issue_number} closed."}
|
||||
|
||||
cleanup_result = release_in_progress_label([issue_number], remote, host, org, repo)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Issue #{issue_number} closed.",
|
||||
"cleanup_status": cleanup_result
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@@ -1293,6 +1384,24 @@ def gitea_whoami(
|
||||
}
|
||||
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_get_authenticated_user(
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
) -> dict:
|
||||
"""Alias for gitea_whoami. Look up the authenticated Gitea account."""
|
||||
return gitea_whoami(remote=remote, host=host)
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_get_current_user(
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
) -> dict:
|
||||
"""Alias for gitea_whoami. Look up the authenticated Gitea account."""
|
||||
return gitea_whoami(remote=remote, host=host)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_get_profile(
|
||||
remote: str = "dadeschools",
|
||||
|
||||
Executable
+61
@@ -0,0 +1,61 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
# clear-provenance — strip the macOS com.apple.provenance extended attribute so
|
||||
# Python.app can execute .py files created by agent/IDE terminals (issue #3).
|
||||
#
|
||||
# macOS Sequoia+ blocks Python.app from executing files carrying
|
||||
# com.apple.provenance. Files written by the agent terminal get it; shell
|
||||
# scripts are unaffected. This is a macOS security feature, not a bug in our
|
||||
# code — see the Troubleshooting section of the README.
|
||||
#
|
||||
# Run from a terminal with Full Disk Access (e.g. Terminal.app), not the IDE
|
||||
# terminal, or the removal itself may be blocked.
|
||||
|
||||
usage() {
|
||||
cat <<'EOF'
|
||||
usage: scripts/clear-provenance [--dry-run] [path]
|
||||
|
||||
Recursively remove the com.apple.provenance extended attribute under <path>
|
||||
(default: the repository root). macOS only. Only that attribute is removed;
|
||||
other extended attributes are left intact.
|
||||
|
||||
Examples:
|
||||
scripts/clear-provenance --dry-run
|
||||
scripts/clear-provenance
|
||||
scripts/clear-provenance /path/to/file.py
|
||||
EOF
|
||||
}
|
||||
|
||||
dry_run=0
|
||||
while [[ "${1:-}" == --* ]]; do
|
||||
case "$1" in
|
||||
--dry-run) dry_run=1 ;;
|
||||
--help) usage; exit 0 ;;
|
||||
*) usage >&2; exit 2 ;;
|
||||
esac
|
||||
shift
|
||||
done
|
||||
|
||||
if [[ $# -gt 1 ]]; then
|
||||
usage >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
repo_root="$(cd "$script_dir/.." && pwd)"
|
||||
target="${1:-$repo_root}"
|
||||
|
||||
if [[ ! -e "$target" ]]; then
|
||||
printf 'clear-provenance: no such path: %s\n' "$target" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Remove only com.apple.provenance; tolerate files that do not carry it.
|
||||
if [[ "$dry_run" -eq 1 ]]; then
|
||||
printf 'clear-provenance: [dry-run] would run: xattr -r -d com.apple.provenance %q\n' "$target"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
xattr -r -d com.apple.provenance "$target" 2>/dev/null || true
|
||||
printf 'clear-provenance: removed com.apple.provenance recursively under: %s\n' "$target"
|
||||
@@ -1,31 +0,0 @@
|
||||
# Recover Dirty Worktree Prompt
|
||||
|
||||
You are recovering repository state in `<repo-name>`.
|
||||
|
||||
Rules:
|
||||
|
||||
- Do not reset, delete, clean, or overwrite work unless explicitly instructed.
|
||||
- Do not edit another issue's worktree unless assigned to that issue.
|
||||
- Preserve ambiguous work before any destructive operation.
|
||||
|
||||
Workflow:
|
||||
|
||||
1. Run `git status --short --branch`.
|
||||
2. Identify whether dirty files belong to the current issue, another issue, or
|
||||
unknown work.
|
||||
3. If dirty work belongs to another issue, leave it alone and use a separate
|
||||
worktree for the current task.
|
||||
4. If an unauthorized untracked file was created, stop and report its exact path.
|
||||
5. Remove unauthorized files only when explicitly instructed.
|
||||
6. If local `<default-branch>` is ahead of `<remote>/<default-branch>`, stop and
|
||||
report both commit hashes.
|
||||
7. If cleanup is requested, verify the branch is merged or explicitly abandoned
|
||||
before deleting any branch or worktree.
|
||||
|
||||
Report:
|
||||
|
||||
- current branch
|
||||
- dirty files
|
||||
- ownership assessment
|
||||
- actions taken
|
||||
- remaining blockers
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Tests for scripts/clear-provenance (#3).
|
||||
|
||||
Exercises argument handling and the inert --dry-run path only — no real xattr
|
||||
mutation, no network. (Actually removing com.apple.provenance is macOS-only and
|
||||
has real side effects, so it is not exercised here.)
|
||||
"""
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
SCRIPT = REPO / "scripts" / "clear-provenance"
|
||||
|
||||
|
||||
def run(*args):
|
||||
proc = subprocess.run(["bash", str(SCRIPT), *args],
|
||||
capture_output=True, text=True, cwd=str(REPO))
|
||||
return proc.returncode, proc.stdout, proc.stderr
|
||||
|
||||
|
||||
class TestClearProvenance(unittest.TestCase):
|
||||
|
||||
def test_dry_run_defaults_to_repo_root(self):
|
||||
rc, out, _ = run("--dry-run")
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertIn("would run: xattr -r -d com.apple.provenance", out)
|
||||
self.assertIn(str(REPO), out)
|
||||
|
||||
def test_dry_run_explicit_path(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
f = Path(d) / "x.py"
|
||||
f.write_text("print('hi')\n")
|
||||
rc, out, _ = run("--dry-run", str(f))
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertIn(str(f), out)
|
||||
|
||||
def test_missing_path_errors(self):
|
||||
rc, _, err = run("--dry-run", "/no/such/path-xyz")
|
||||
self.assertEqual(rc, 1)
|
||||
self.assertIn("no such path", err)
|
||||
|
||||
def test_bad_flag_exit_2(self):
|
||||
rc, _, _ = run("--bogus")
|
||||
self.assertEqual(rc, 2)
|
||||
|
||||
def test_too_many_args_exit_2(self):
|
||||
rc, _, _ = run("a", "b")
|
||||
self.assertEqual(rc, 2)
|
||||
|
||||
def test_only_targets_provenance_attribute(self):
|
||||
# The command removes only com.apple.provenance, not all xattrs.
|
||||
rc, out, _ = run("--dry-run")
|
||||
self.assertIn("com.apple.provenance", out)
|
||||
self.assertNotIn("xattr -rc", out) # not a blanket "clear all"
|
||||
self.assertNotIn("-c ", out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -137,5 +137,89 @@ class TestConstants(unittest.TestCase):
|
||||
f"Label '{label['name']}' has invalid color")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Modes: --create-labels / --apply-mapping / --add-label (#6)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestModes(unittest.TestCase):
|
||||
|
||||
def _methods(self, mock_api):
|
||||
return [(c[0][0], c[0][1]) for c in mock_api.call_args_list]
|
||||
|
||||
@patch("manage_labels.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("manage_labels.api")
|
||||
def test_create_labels_only_no_mapping(self, mock_api, _auth):
|
||||
def se(method, path, auth, payload=None):
|
||||
if method == "GET":
|
||||
return [] # no existing labels
|
||||
if method == "POST" and path == "/labels":
|
||||
return {"id": 1, "name": payload["name"]}
|
||||
return None
|
||||
mock_api.side_effect = se
|
||||
manage_labels.main(["--create-labels"])
|
||||
methods = self._methods(mock_api)
|
||||
self.assertTrue(any(m == ("POST", "/labels") for m in methods))
|
||||
self.assertFalse(any(m[0] == "PUT" for m in methods)) # no mapping applied
|
||||
|
||||
@patch("manage_labels.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("manage_labels.api")
|
||||
def test_apply_mapping_only_no_label_creation(self, mock_api, _auth):
|
||||
existing = [_make_label(l["name"], i + 1)
|
||||
for i, l in enumerate(manage_labels.LABELS)]
|
||||
|
||||
def se(method, path, auth, payload=None):
|
||||
if method == "GET":
|
||||
return existing
|
||||
if method == "PUT":
|
||||
return [{"name": "applied"}]
|
||||
return None
|
||||
mock_api.side_effect = se
|
||||
manage_labels.main(["--apply-mapping"])
|
||||
methods = self._methods(mock_api)
|
||||
self.assertFalse(any(m == ("POST", "/labels") for m in methods))
|
||||
put_calls = [m for m in methods if m[0] == "PUT"]
|
||||
self.assertEqual(len(put_calls), len(manage_labels.MAPPING))
|
||||
|
||||
@patch("manage_labels.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("manage_labels.api")
|
||||
def test_add_label_appends_to_issue(self, mock_api, _auth):
|
||||
existing = [_make_label("chore", 5)]
|
||||
|
||||
def se(method, path, auth, payload=None):
|
||||
if method == "GET":
|
||||
return existing
|
||||
if method == "POST":
|
||||
return [{"name": "chore"}]
|
||||
return None
|
||||
mock_api.side_effect = se
|
||||
manage_labels.main(["--add-label", "42", "chore"])
|
||||
posts = [c for c in mock_api.call_args_list
|
||||
if c[0][0] == "POST" and c[0][1] == "/issues/42/labels"]
|
||||
self.assertEqual(len(posts), 1)
|
||||
self.assertEqual(posts[0][0][3], {"labels": [5]}) # append, id 5
|
||||
# POST appends; no PUT (which would replace the whole set).
|
||||
self.assertFalse(any(c[0][0] == "PUT" for c in mock_api.call_args_list))
|
||||
|
||||
@patch("manage_labels.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("manage_labels.api")
|
||||
def test_add_label_unknown_makes_no_write(self, mock_api, _auth):
|
||||
mock_api.side_effect = lambda *a, **k: [] if a[0] == "GET" else None
|
||||
manage_labels.main(["--add-label", "42", "ghost"])
|
||||
# Only the GET label lookup; no POST/PUT for an undefined label.
|
||||
self.assertTrue(all(c[0][0] == "GET" for c in mock_api.call_args_list))
|
||||
|
||||
@patch("manage_labels.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("manage_labels.api")
|
||||
def test_add_label_dry_makes_no_write(self, mock_api, _auth):
|
||||
mock_api.side_effect = lambda *a, **k: [_make_label("chore", 5)] if a[0] == "GET" else None
|
||||
manage_labels.main(["--dry", "--add-label", "42", "chore"])
|
||||
self.assertTrue(all(c[0][0] == "GET" for c in mock_api.call_args_list))
|
||||
|
||||
@patch("manage_labels.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("manage_labels.api")
|
||||
def test_add_label_non_numeric_issue_exits(self, mock_api, _auth):
|
||||
with self.assertRaises(SystemExit):
|
||||
manage_labels.main(["--add-label", "notanum", "chore"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
+249
-1
@@ -93,7 +93,8 @@ class TestCloseIssue(unittest.TestCase):
|
||||
result = gitea_close_issue(issue_number=42)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertIn("42", result["message"])
|
||||
payload = mock_api.call_args[0][3]
|
||||
patch_call = next(call for call in mock_api.call_args_list if call[0][0] == "PATCH")
|
||||
payload = patch_call[0][3]
|
||||
self.assertEqual(payload["state"], "closed")
|
||||
|
||||
|
||||
@@ -861,6 +862,34 @@ class TestWhoami(unittest.TestCase):
|
||||
# ---------------------------------------------------------------------------
|
||||
# Runtime profile (env-configured profile metadata) — issue #19
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server._auth")
|
||||
def test_gitea_get_authenticated_user_alias(self, _auth, mock_api):
|
||||
mock_api.return_value = {
|
||||
"login": "alias_user",
|
||||
"full_name": "Alias User",
|
||||
"id": 999,
|
||||
"email": "alias@example.com"
|
||||
}
|
||||
from mcp_server import gitea_get_authenticated_user
|
||||
result = gitea_get_authenticated_user(remote="prgs")
|
||||
self.assertEqual(result["username"], "alias_user")
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server._auth")
|
||||
def test_gitea_get_current_user_alias(self, _auth, mock_api):
|
||||
mock_api.return_value = {
|
||||
"login": "alias_user",
|
||||
"full_name": "Alias User",
|
||||
"id": 999,
|
||||
"email": "alias@example.com"
|
||||
}
|
||||
from mcp_server import gitea_get_current_user
|
||||
result = gitea_get_current_user(remote="prgs")
|
||||
self.assertEqual(result["username"], "alias_user")
|
||||
|
||||
|
||||
class TestRuntimeProfile(unittest.TestCase):
|
||||
|
||||
def test_defaults_when_unset(self):
|
||||
@@ -1352,3 +1381,222 @@ class TestSubmitPrReview(unittest.TestCase):
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tracker Hygiene Cleanup Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestTrackerHygieneCleanup(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.mock_api = patch("mcp_server.api_request").start()
|
||||
self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
|
||||
patch("gitea_audit.audit_enabled", return_value=True).start()
|
||||
self.mock_audit = patch("gitea_audit.write_event").start()
|
||||
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close"], "audit_label": "test", "forbidden_operations": []}).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
|
||||
def test_close_issue_removes_in_progress(self):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "PATCH" and "issues/1" in url:
|
||||
return {"state": "closed"}
|
||||
if method == "GET" and "labels" in url and "issues" not in url:
|
||||
return [{"name": "status:in-progress", "id": 1}, {"name": "bug", "id": 2}]
|
||||
if method == "GET" and "issues/1" in url:
|
||||
return {"labels": [{"name": "status:in-progress"}, {"name": "bug"}]}
|
||||
if method == "DELETE" and url.endswith("/issues/1/labels/1"):
|
||||
return {}
|
||||
if method == "PUT" and "labels" in url:
|
||||
self.fail("Should not replace the issue label set")
|
||||
return {}
|
||||
self.mock_api.side_effect = api_side_effect
|
||||
|
||||
res = gitea_close_issue(issue_number=1)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["cleanup_status"].get(1), "released")
|
||||
self.mock_audit.assert_called()
|
||||
|
||||
def test_close_issue_no_label_is_noop(self):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "PATCH" and "issues/1" in url:
|
||||
return {"state": "closed"}
|
||||
if method == "GET" and "labels" in url and "issues" not in url:
|
||||
return [{"name": "status:in-progress", "id": 1}, {"name": "bug", "id": 2}]
|
||||
if method == "GET" and "issues/1" in url:
|
||||
return {"labels": [{"name": "bug"}]}
|
||||
if method == "DELETE" and "labels" in url:
|
||||
self.fail("Should not DELETE labels")
|
||||
if method == "PUT" and "labels" in url:
|
||||
self.fail("Should not replace the issue label set")
|
||||
return {}
|
||||
self.mock_api.side_effect = api_side_effect
|
||||
|
||||
res = gitea_close_issue(issue_number=1)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["cleanup_status"].get(1), "not present")
|
||||
|
||||
def test_merge_pr_with_closes_removes_label(self):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "GET" and "/user" in url:
|
||||
return {"login": "merger"}
|
||||
if method == "GET" and "pulls/1" in url and "/files" not in url:
|
||||
return {
|
||||
"user": {"login": "author"},
|
||||
"state": "open",
|
||||
"head": {"sha": "sha123", "ref": "feat/my-branch"},
|
||||
"base": {"ref": "main"},
|
||||
"mergeable": True,
|
||||
"merged_commit_sha": "merge123",
|
||||
"title": "My PR",
|
||||
"body": "Closes #123"
|
||||
}
|
||||
if method == "POST" and "merge" in url:
|
||||
return {}
|
||||
if method == "GET" and "labels" in url and "issues" not in url:
|
||||
return [{"name": "status:in-progress", "id": 1}, {"name": "bug", "id": 2}]
|
||||
if method == "GET" and "issues/123" in url:
|
||||
return {"labels": [{"name": "status:in-progress"}, {"name": "bug"}]}
|
||||
if method == "DELETE" and url.endswith("/issues/123/labels/1"):
|
||||
return {}
|
||||
if method == "PUT" and "labels" in url:
|
||||
self.fail("Should not replace the issue label set")
|
||||
return {}
|
||||
self.mock_api.side_effect = api_side_effect
|
||||
|
||||
res = gitea_merge_pr(pr_number=1, confirmation="MERGE PR 1", do="merge")
|
||||
self.assertTrue(res["performed"])
|
||||
self.assertEqual(res["cleanup_status"].get(123), "released")
|
||||
|
||||
def test_merge_pr_with_branch_name_removes_label(self):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "GET" and "/user" in url:
|
||||
return {"login": "merger"}
|
||||
if method == "GET" and "pulls/1" in url and "/files" not in url:
|
||||
return {
|
||||
"user": {"login": "author"},
|
||||
"state": "open",
|
||||
"head": {"sha": "sha123", "ref": "fix/issue-123-slug"},
|
||||
"base": {"ref": "main"},
|
||||
"mergeable": True,
|
||||
"merged_commit_sha": "merge123",
|
||||
"title": "My PR",
|
||||
"body": "Fixing things"
|
||||
}
|
||||
if method == "POST" and "merge" in url:
|
||||
return {}
|
||||
if method == "GET" and "labels" in url and "issues" not in url:
|
||||
return [{"name": "status:in-progress", "id": 1}, {"name": "bug", "id": 2}]
|
||||
if method == "GET" and "issues/123" in url:
|
||||
return {"labels": [{"name": "status:in-progress"}, {"name": "bug"}]}
|
||||
if method == "DELETE" and url.endswith("/issues/123/labels/1"):
|
||||
return {}
|
||||
if method == "PUT" and "labels" in url:
|
||||
self.fail("Should not replace the issue label set")
|
||||
return {}
|
||||
self.mock_api.side_effect = api_side_effect
|
||||
|
||||
res = gitea_merge_pr(pr_number=1, confirmation="MERGE PR 1", do="merge")
|
||||
self.assertTrue(res["performed"])
|
||||
self.assertEqual(res["cleanup_status"].get(123), "released")
|
||||
|
||||
def test_close_pr_removes_label_but_does_not_close_issue(self):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "PATCH" and "pulls/1" in url:
|
||||
return {
|
||||
"number": 1,
|
||||
"title": "My PR",
|
||||
"state": "closed",
|
||||
"html_url": "url",
|
||||
"body": "Closes #123",
|
||||
"head": {"ref": "feat/my-branch"}
|
||||
}
|
||||
if method == "GET" and "labels" in url and "issues" not in url:
|
||||
return [{"name": "status:in-progress", "id": 1}]
|
||||
if method == "GET" and "issues/123" in url:
|
||||
return {"labels": [{"name": "status:in-progress"}]}
|
||||
if method == "DELETE" and url.endswith("/issues/123/labels/1"):
|
||||
return {}
|
||||
if method == "PUT" and "labels" in url:
|
||||
self.fail("Should not replace the issue label set")
|
||||
if method == "POST" and "comments" in url:
|
||||
return {}
|
||||
return {}
|
||||
self.mock_api.side_effect = api_side_effect
|
||||
|
||||
res = gitea_edit_pr(pr_number=1, state="closed")
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["cleanup_status"].get(123), "released")
|
||||
|
||||
def test_multiple_linked_issues(self):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "PATCH" and "pulls/1" in url:
|
||||
return {
|
||||
"number": 1,
|
||||
"title": "My PR",
|
||||
"state": "closed",
|
||||
"html_url": "url",
|
||||
"body": "Closes #123\nFixes #124",
|
||||
"head": {"ref": "issue-125"}
|
||||
}
|
||||
if method == "GET" and "labels" in url and "issues" not in url:
|
||||
return [{"name": "status:in-progress", "id": 1}]
|
||||
if method == "GET" and "issues/123" in url:
|
||||
return {"labels": [{"name": "status:in-progress"}]}
|
||||
if method == "GET" and "issues/124" in url:
|
||||
return {"labels": [{"name": "status:in-progress"}]}
|
||||
if method == "GET" and "issues/125" in url:
|
||||
return {"labels": []}
|
||||
if method == "DELETE" and url.endswith("/issues/123/labels/1"):
|
||||
return {}
|
||||
if method == "DELETE" and url.endswith("/issues/124/labels/1"):
|
||||
return {}
|
||||
if method == "PUT" and "labels" in url:
|
||||
self.fail("Should not replace the issue label set")
|
||||
if method == "POST" and "comments" in url:
|
||||
return {}
|
||||
return {}
|
||||
self.mock_api.side_effect = api_side_effect
|
||||
|
||||
res = gitea_edit_pr(pr_number=1, state="closed")
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["cleanup_status"].get(123), "released")
|
||||
self.assertEqual(res["cleanup_status"].get(124), "released")
|
||||
self.assertEqual(res["cleanup_status"].get(125), "not present")
|
||||
|
||||
def test_no_linked_issue_found(self):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "PATCH" and "pulls/1" in url:
|
||||
return {
|
||||
"number": 1,
|
||||
"title": "My PR",
|
||||
"state": "closed",
|
||||
"html_url": "url",
|
||||
"body": "No issue link",
|
||||
"head": {"ref": "main"}
|
||||
}
|
||||
return {}
|
||||
self.mock_api.side_effect = api_side_effect
|
||||
|
||||
res = gitea_edit_pr(pr_number=1, state="closed")
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["cleanup_status"], "no linked issue found")
|
||||
|
||||
def test_label_removal_failure_reported(self):
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "PATCH" and "issues/1" in url:
|
||||
return {"state": "closed"}
|
||||
if method == "GET" and "labels" in url and "issues" not in url:
|
||||
return [{"name": "status:in-progress", "id": 1}]
|
||||
if method == "GET" and "issues/1" in url:
|
||||
return {"labels": [{"name": "status:in-progress"}]}
|
||||
if method == "DELETE" and url.endswith("/issues/1/labels/1"):
|
||||
raise RuntimeError("API failure")
|
||||
if method == "PUT" and "labels" in url:
|
||||
self.fail("Should not replace the issue label set")
|
||||
return {}
|
||||
self.mock_api.side_effect = api_side_effect
|
||||
|
||||
res = gitea_close_issue(issue_number=1)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertIn("error:", res["cleanup_status"].get(1))
|
||||
|
||||
Reference in New Issue
Block a user