feat: bypass Cloudflare block & convert close/mark issue shell scripts to Python

This commit is contained in:
2026-06-21 22:32:37 -04:00
parent 51296c88a3
commit 80a71f2020
8 changed files with 232 additions and 162 deletions
+8 -8
View File
@@ -132,8 +132,8 @@ The MCP tools can also be used as standalone CLI scripts:
|---------------------|--------------------------------------------------------------------|
| `create_issue.py` | Create an issue (`--remote`, `--title`, `--body`, `--body-file`) |
| `create_pr.py` | Open a Pull Request (`--remote`, `--title`, `--head`, `--base`) |
| `close_issue.sh` | Close a specific issue (dadeschools only) |
| `mark_issue.sh` | Claim/release an issue via `status:in-progress` label |
| `close_issue.py` | Close a specific issue |
| `mark_issue.py` | Claim/release an issue via `status:in-progress` label |
| `manage_labels.py` | Create label set and apply label mappings (`--dry` to preview) |
| `mirror_refs.sh` | Mirror branches + tags between dadeschools ⇄ prgs |
@@ -150,13 +150,13 @@ The MCP tools can also be used as standalone CLI scripts:
./create_pr.py --title "feat: add validation" --head feat/validation --body "Closes #12"
# Close issue #5
./close_issue.sh 5
./close_issue.py 5
# Claim an issue before working on it
./mark_issue.sh 10 start
./mark_issue.py 10 start
# Release when done
./mark_issue.sh 10 done
./mark_issue.py 10 done
# Mirror refs (dry-run by default)
./mirror_refs.sh
@@ -175,8 +175,8 @@ mcp_server.py ← MCP server (FastMCP, stdio transport)
create_issue.py ← CLI: create issues
create_pr.py ← CLI: create PRs
manage_labels.py ← CLI: label management
close_issue.sh ← CLI: close issues
mark_issue.sh ← CLI: claim/release issues
close_issue.py ← CLI: close issues
mark_issue.py ← CLI: claim/release issues
mirror_refs.sh ← CLI: ref mirroring
```
@@ -195,7 +195,7 @@ python3 -m pytest tests/ -v
| `test_create_pr.py` | CLI arg parsing, remote resolution, payload, auth, errors |
| `test_credentials.py` | `get_credentials()`, `get_auth_header()`, `repo_api_url()` |
| `test_manage_labels.py` | Label create/skip, dry run, mapping, constant validation |
| `test_shell_scripts.py` | `close_issue.sh` + `mark_issue.sh` arg validation |
| `test_python_cli.py` | `close_issue.py` + `mark_issue.py` CLI validation |
| `test_mirror_refs.py` | Flags, safety defaults, local integration tests |
All tests mock network and keychain access — no real API calls are made.
Executable
+42
View File
@@ -0,0 +1,42 @@
#!/usr/bin/env python3
"""Close a Gitea issue.
Usage:
close_issue.py <issue_number>
close_issue.py --remote prgs 12
"""
import sys
import argparse
from gitea_auth import (
get_auth_header, resolve_remote, add_remote_args,
api_request, repo_api_url,
)
def main(argv=None):
parser = argparse.ArgumentParser(description="Close a Gitea issue.")
add_remote_args(parser)
parser.add_argument("issue_number", type=int, help="Issue number to close.")
args = parser.parse_args(argv)
host, org, repo = resolve_remote(args)
auth = get_auth_header(host)
if not auth:
print(f"Could not get credentials or token for {host}.", file=sys.stderr)
return 1
url = f"{repo_api_url(host, org, repo)}/issues/{args.issue_number}"
try:
api_request("PATCH", url, auth, {"state": "closed"})
print(f"#{args.issue_number} closed")
return 0
except RuntimeError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
-27
View File
@@ -1,27 +0,0 @@
#!/usr/bin/env bash
# Close a Gitea issue by setting its state to "closed".
#
# Usage: ./close_issue.sh <issue_number>
#
# Auth: macOS keychain via `git credential fill` (same as the other scripts).
set -euo pipefail
HOST="gitea.dadeschools.net"
API="https://$HOST/api/v1"
ORG="Contractor"
REPO="Timesheet"
ISSUE_NUM="${1:?usage: close_issue.sh <issue_number>}"
CREDS=$(printf "host=%s\nprotocol=https\n\n" "$HOST" | git credential fill)
USER=$(printf '%s\n' "$CREDS" | sed -n 's/^username=//p')
PASS=$(printf '%s\n' "$CREDS" | sed -n 's/^password=//p')
curl -sSL -X PATCH \
-u "$USER:$PASS" \
-H "Content-Type: application/json" \
-d '{"state": "closed"}' \
"$API/repos/$ORG/$REPO/issues/$ISSUE_NUM"
echo ""
echo "#$ISSUE_NUM closed"
+1
View File
@@ -157,6 +157,7 @@ def api_request(method, url, auth_header, payload=None):
req = urllib.request.Request(url, data=data, method=method)
req.add_header("Authorization", auth_header)
req.add_header("Content-Type", "application/json")
req.add_header("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
try:
with urllib.request.urlopen(req) as resp:
body = resp.read().decode("utf-8")
Executable
+67
View File
@@ -0,0 +1,67 @@
#!/usr/bin/env python3
"""Claim or release a Gitea issue by toggling status:in-progress label.
Usage:
mark_issue.py <issue_number> [start|done]
mark_issue.py --remote prgs 12 done
"""
import sys
import argparse
from gitea_auth import (
get_auth_header, resolve_remote, add_remote_args,
api_request, repo_api_url,
)
LABEL_NAME = "status:in-progress"
def main(argv=None):
parser = argparse.ArgumentParser(
description="Claim or release a Gitea issue by toggling the status:in-progress label."
)
add_remote_args(parser)
parser.add_argument("issue_number", type=int, help="Issue number.")
parser.add_argument(
"action", nargs="?", choices=["start", "done"], default="start",
help="Action: 'start' to claim (default) or 'done' to release."
)
args = parser.parse_args(argv)
host, org, repo = resolve_remote(args)
auth = get_auth_header(host)
if not auth:
print(f"Could not get credentials or token for {host}.", file=sys.stderr)
return 1
base = repo_api_url(host, org, repo)
try:
# Find the label ID
labels = api_request("GET", f"{base}/labels?limit=100", auth)
label_id = None
for lb in labels:
if lb["name"] == LABEL_NAME:
label_id = lb["id"]
break
if label_id is None:
print(f"Label '{LABEL_NAME}' not found in {org}/{repo} -- run manage_labels.py first.", file=sys.stderr)
return 1
if args.action == "start":
api_request("POST", f"{base}/issues/{args.issue_number}/labels", auth,
{"labels": [label_id]})
print(f"#{args.issue_number} claimed -> {LABEL_NAME}")
else:
api_request("DELETE", f"{base}/issues/{args.issue_number}/labels/{label_id}", auth)
print(f"#{args.issue_number} released -> {LABEL_NAME} removed")
return 0
except RuntimeError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
-50
View File
@@ -1,50 +0,0 @@
#!/usr/bin/env bash
# Claim or release a Gitea issue by toggling the `status:in-progress` label, so
# parallel agents/LLMs don't pick up the same issue.
#
# Usage:
# ./mark_issue.sh <issue_number> # claim (adds status:in-progress)
# ./mark_issue.sh <issue_number> start # claim
# ./mark_issue.sh <issue_number> done # release (removes the label)
#
# Auth: macOS keychain via `git credential fill` (same as the other scripts).
set -euo pipefail
HOST="gitea.dadeschools.net"
API="https://$HOST/api/v1"
ORG="Contractor"
REPO="Timesheet"
LABEL="status:in-progress"
NUM="${1:?usage: mark_issue.sh <issue_number> [start|done]}"
ACTION="${2:-start}"
CREDS=$(printf "host=%s\nprotocol=https\n\n" "$HOST" | git credential fill)
USER=$(printf '%s\n' "$CREDS" | sed -n 's/^username=//p')
PASS=$(printf '%s\n' "$CREDS" | sed -n 's/^password=//p')
AUTH=(-u "$USER:$PASS")
# Resolve the label name -> id (Gitea's issue label endpoints take ids).
LID=$(curl -sSL "${AUTH[@]}" "$API/repos/$ORG/$REPO/labels?limit=100" \
| python3 -c "import sys,json;print(next((l['id'] for l in json.load(sys.stdin) if l['name']=='$LABEL'),''))")
if [ -z "$LID" ]; then
echo "Label '$LABEL' not found in $ORG/$REPO -- run manage_labels.py first." >&2
exit 1
fi
case "$ACTION" in
start)
curl -sSL -X POST "${AUTH[@]}" -H "Content-Type: application/json" \
-d "{\"labels\":[$LID]}" "$API/repos/$ORG/$REPO/issues/$NUM/labels" >/dev/null
echo "#$NUM claimed -> $LABEL"
;;
done)
curl -sSL -X DELETE "${AUTH[@]}" \
"$API/repos/$ORG/$REPO/issues/$NUM/labels/$LID" >/dev/null
echo "#$NUM released -> $LABEL removed"
;;
*)
echo "Unknown action '$ACTION' (expected: start | done)" >&2
exit 1
;;
esac
+114
View File
@@ -0,0 +1,114 @@
"""Tests for the python CLI scripts close_issue.py and mark_issue.py.
All tests mock credentials and API requests so no real network calls are made.
"""
import sys
import unittest
from unittest.mock import patch, MagicMock
# The modules under test live in the repo root
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import close_issue
import mark_issue
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
# ---------------------------------------------------------------------------
# close_issue.py
# ---------------------------------------------------------------------------
class TestCloseIssueCLI(unittest.TestCase):
def test_missing_argument_exits(self):
with self.assertRaises(SystemExit):
close_issue.main([])
def test_invalid_argument_type_exits(self):
with self.assertRaises(SystemExit):
close_issue.main(["not_a_number"])
@patch("close_issue.api_request")
@patch("close_issue.get_auth_header", return_value=FAKE_AUTH)
def test_successful_close(self, _auth, mock_api):
rc = close_issue.main(["42"])
self.assertEqual(rc, 0)
mock_api.assert_called_once()
url = mock_api.call_args[0][1]
self.assertIn("/issues/42", url)
self.assertEqual(mock_api.call_args[0][3], {"state": "closed"})
@patch("close_issue.api_request")
@patch("close_issue.get_auth_header", return_value=FAKE_AUTH)
def test_remote_override(self, _auth, mock_api):
rc = close_issue.main(["--remote", "prgs", "12"])
self.assertEqual(rc, 0)
url = mock_api.call_args[0][1]
self.assertIn("gitea.prgs.cc", url)
self.assertIn("/issues/12", url)
# ---------------------------------------------------------------------------
# mark_issue.py
# ---------------------------------------------------------------------------
class TestMarkIssueCLI(unittest.TestCase):
def test_missing_argument_exits(self):
with self.assertRaises(SystemExit):
mark_issue.main([])
def test_invalid_action_exits(self):
with self.assertRaises(SystemExit):
mark_issue.main(["10", "bogus_action"])
@patch("mark_issue.api_request")
@patch("mark_issue.get_auth_header", return_value=FAKE_AUTH)
def test_successful_start(self, _auth, mock_api):
# First call is GET labels, second is POST label
mock_api.side_effect = [
[{"id": 101, "name": "status:in-progress"}],
[{"name": "status:in-progress"}],
]
rc = mark_issue.main(["15", "start"])
self.assertEqual(rc, 0)
self.assertEqual(mock_api.call_count, 2)
# Verify GET labels call
get_call = mock_api.call_args_list[0]
self.assertEqual(get_call[0][0], "GET")
self.assertIn("/labels?limit=100", get_call[0][1])
# Verify POST labels call
post_call = mock_api.call_args_list[1]
self.assertEqual(post_call[0][0], "POST")
self.assertIn("/issues/15/labels", post_call[0][1])
self.assertEqual(post_call[0][3], {"labels": [101]})
@patch("mark_issue.api_request")
@patch("mark_issue.get_auth_header", return_value=FAKE_AUTH)
def test_successful_done(self, _auth, mock_api):
# First call is GET labels, second is DELETE label
mock_api.side_effect = [
[{"id": 101, "name": "status:in-progress"}],
None,
]
rc = mark_issue.main(["15", "done"])
self.assertEqual(rc, 0)
self.assertEqual(mock_api.call_count, 2)
# Verify DELETE labels call
delete_call = mock_api.call_args_list[1]
self.assertEqual(delete_call[0][0], "DELETE")
self.assertIn("/issues/15/labels/101", delete_call[0][1])
@patch("mark_issue.api_request")
@patch("mark_issue.get_auth_header", return_value=FAKE_AUTH)
def test_label_not_found(self, _auth, mock_api):
# GET labels returns no status:in-progress label
mock_api.return_value = [{"id": 1, "name": "bug"}]
rc = mark_issue.main(["15", "start"])
self.assertEqual(rc, 1)
if __name__ == "__main__":
unittest.main()
-77
View File
@@ -1,77 +0,0 @@
"""Tests for the shell scripts (close_issue.sh, mark_issue.sh).
These are integration-style tests that verify the scripts':
- argument validation and error messages
- exit codes on bad input
- correct curl command construction (via a mock curl wrapper)
We do NOT make real API calls. Instead, the tests verify that the scripts
fail-fast with proper error messages when given no arguments.
"""
import os
import subprocess
import unittest
REPO_ROOT = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
def _run_script(script, args=None, env_override=None):
"""Run a shell script and return (returncode, stdout, stderr)."""
cmd = [os.path.join(REPO_ROOT, script)]
if args:
cmd.extend(args)
env = os.environ.copy()
if env_override:
env.update(env_override)
result = subprocess.run(
cmd, capture_output=True, text=True, env=env, timeout=10,
)
return result.returncode, result.stdout, result.stderr
# ---------------------------------------------------------------------------
# close_issue.sh
# ---------------------------------------------------------------------------
class TestCloseIssue(unittest.TestCase):
def test_no_args_prints_usage_and_fails(self):
rc, stdout, stderr = _run_script("close_issue.sh")
self.assertNotEqual(rc, 0)
self.assertIn("usage:", stderr)
def test_usage_mentions_issue_number(self):
rc, stdout, stderr = _run_script("close_issue.sh")
self.assertIn("issue_number", stderr)
# ---------------------------------------------------------------------------
# mark_issue.sh
# ---------------------------------------------------------------------------
class TestMarkIssue(unittest.TestCase):
def test_no_args_prints_usage_and_fails(self):
rc, stdout, stderr = _run_script("mark_issue.sh")
self.assertNotEqual(rc, 0)
self.assertIn("usage:", stderr)
def test_usage_mentions_start_done(self):
rc, stdout, stderr = _run_script("mark_issue.sh")
self.assertIn("start|done", stderr)
def test_invalid_action_fails(self):
"""Providing an issue number but an invalid action should fail.
Note: this test may fail if `git credential fill` hangs waiting for
input. In CI without credentials, it will error out with a non-zero
exit code, which is what we're testing for — the script should not
succeed with an invalid action.
"""
rc, stdout, stderr = _run_script("mark_issue.sh", ["999", "bogus"])
# The script will either fail at credential lookup (no keychain in CI)
# or at the invalid action case statement. Either way, it should not
# exit 0.
self.assertNotEqual(rc, 0)
if __name__ == "__main__":
unittest.main()