From 82fcd5a4bc352ae504c856af41a9b07a31d25d92 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 24 Jun 2026 00:14:47 -0400 Subject: [PATCH] feat: expand MCP server tools for PR and label management, add helper CLI scripts Closes #7 --- close_issue.py | 7 + create_issue.py | 7 + create_pr.py | 15 ++- delete_branch.py | 48 +++++++ list_all_remotes_prs.py | 37 ++++++ list_eagenda_prs.py | 39 ++++++ list_gitea_tools_prs.py | 36 ++++++ list_issues.py | 59 +++++++++ list_prs.py | 53 ++++++++ manage_labels.py | 7 + mark_issue.py | 7 + mcp_server.py | 267 ++++++++++++++++++++++++++++++++++++++- merge_pr.py | 58 +++++++++ tests/test_mcp_server.py | 82 ++++++++++++ tests/test_merge_pr.py | 64 ++++++++++ tests/test_prs.py | 68 ++++++++++ view_pr.py | 52 ++++++++ 17 files changed, 901 insertions(+), 5 deletions(-) create mode 100644 delete_branch.py create mode 100755 list_all_remotes_prs.py create mode 100755 list_eagenda_prs.py create mode 100755 list_gitea_tools_prs.py create mode 100644 list_issues.py create mode 100755 list_prs.py create mode 100755 merge_pr.py create mode 100644 tests/test_merge_pr.py create mode 100644 tests/test_prs.py create mode 100755 view_pr.py diff --git a/close_issue.py b/close_issue.py index e0ebb6b..02b4fc5 100755 --- a/close_issue.py +++ b/close_issue.py @@ -5,9 +5,16 @@ Usage: close_issue.py close_issue.py --remote prgs 12 """ +import os import sys import argparse +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + from gitea_auth import ( get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, diff --git a/create_issue.py b/create_issue.py index 19e51ab..6ec1b4b 100644 --- a/create_issue.py +++ b/create_issue.py @@ -15,9 +15,16 @@ Examples: create_issue.py --remote prgs --title "Add tests" --body-file desc.md create_issue.py --host gitea.example.com --org Foo --repo Bar --title "..." """ +import os import sys import argparse +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + from gitea_auth import ( get_credentials, resolve_remote, add_remote_args, api_request, repo_api_url, diff --git a/create_pr.py b/create_pr.py index 559b358..0a1b487 100755 --- a/create_pr.py +++ b/create_pr.py @@ -11,17 +11,18 @@ Auth is pulled from the macOS keychain via `git credential fill` for the chosen host -- no tokens on the command line. Examples: - create_pr.py --remote dadeschools \\ - --title "Open generated PDF after creation (#7)" \\ - --head feat/7-open-pdf-2234ddf5 --base feat/4-5-24-validations \\ + create_pr.py --remote dadeschools \ + --title "Open generated PDF after creation (#7)" \ + --head feat/7-open-pdf-2234ddf5 --base feat/4-5-24-validations \ --body "Closes #7" create_pr.py --remote prgs --title "Fix X" --head fix/x --body-file body.md # override any field of a known remote, or point at an arbitrary repo: - create_pr.py --host gitea.example.com --org Foo --repo Bar \\ + create_pr.py --host gitea.example.com --org Foo --repo Bar \ --title "..." --head topic """ +import os import sys import json import base64 @@ -29,6 +30,12 @@ import argparse import urllib.request import urllib.error +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + from gitea_auth import get_credentials, resolve_remote, add_remote_args diff --git a/delete_branch.py b/delete_branch.py new file mode 100644 index 0000000..12b11dd --- /dev/null +++ b/delete_branch.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +"""Delete a Gitea remote branch. + +Usage: + delete_branch.py + delete_branch.py --remote prgs feat/my-feature +""" +import os +import sys +import argparse + +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +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="Delete a Gitea remote branch.") + add_remote_args(parser) + parser.add_argument("branch", help="Remote branch name to delete.") + 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 + + # Encode the branch name in case it contains slashes, e.g. feat/my-branch + import urllib.parse + encoded_branch = urllib.parse.quote(args.branch, safe="") + url = f"{repo_api_url(host, org, repo)}/branches/{encoded_branch}" + + try: + api_request("DELETE", url, auth) + print(f"Successfully deleted remote branch '{args.branch}'") + return 0 + except Exception as e: + print(f"Error deleting branch '{args.branch}': {e}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/list_all_remotes_prs.py b/list_all_remotes_prs.py new file mode 100755 index 0000000..365a9bd --- /dev/null +++ b/list_all_remotes_prs.py @@ -0,0 +1,37 @@ +#!/usr/bin/env python3 +import os +import sys + +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +from gitea_auth import get_auth_header, api_request, REMOTES, repo_api_url + +def main(): + for name, remote in REMOTES.items(): + host = remote['host'] + org = remote['org'] + repo = remote['repo'] + print(f"Checking remote: {name} ({host}/{org}/{repo})...") + try: + auth = get_auth_header(host) + if not auth: + print(f" No auth header found for {host}") + continue + url = f"{repo_api_url(host, org, repo)}/pulls?state=open" + prs = api_request("GET", url, auth) + if not prs: + print(" No open pull requests.") + for pr in prs: + print(f" PR #{pr['number']}: {pr['title']}") + print(f" Branch: {pr['head']['ref']} -> {pr['base']['ref']}") + print(f" URL: {pr['html_url']}") + print(f" Mergeable: {pr.get('mergeable')}") + except Exception as e: + print(f" Error: {e}") + +if __name__ == '__main__': + main() diff --git a/list_eagenda_prs.py b/list_eagenda_prs.py new file mode 100755 index 0000000..ba6415c --- /dev/null +++ b/list_eagenda_prs.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +import os +import sys + +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +from gitea_auth import get_auth_header, api_request + +def try_repo(host, org, repo): + print(f"Checking repo: {host}/{org}/{repo}...") + try: + auth = get_auth_header(host) + if not auth: + print(f" No auth header found for {host}") + return + url = f"https://{host}/api/v1/repos/{org}/{repo}/pulls?state=open" + prs = api_request("GET", url, auth) + if not prs: + print(" No open pull requests.") + for pr in prs: + print(f" PR #{pr['number']}: {pr['title']}") + print(f" Branch: {pr['head']['ref']} -> {pr['base']['ref']}") + print(f" URL: {pr['html_url']}") + print(f" Mergeable: {pr.get('mergeable')}") + except Exception as e: + print(f" Error: {e}") + +def main(): + try_repo('gitea.dadeschools.net', '913443', 'eAgenda') + try_repo('gitea.dadeschools.net', 'Contractor', 'eAgenda') + try_repo('gitea.prgs.cc', '913443', 'eAgenda') + try_repo('gitea.prgs.cc', 'Scaled-Tech-Consulting', 'eAgenda') + +if __name__ == '__main__': + main() diff --git a/list_gitea_tools_prs.py b/list_gitea_tools_prs.py new file mode 100755 index 0000000..32f9aff --- /dev/null +++ b/list_gitea_tools_prs.py @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 +import os +import sys + +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +from gitea_auth import get_auth_header, api_request + +def main(): + host = 'gitea.prgs.cc' + org = 'Scaled-Tech-Consulting' + repo = 'Gitea-Tools' + print(f"Checking repo: {host}/{org}/{repo}...") + try: + auth = get_auth_header(host) + if not auth: + print(f" No auth header found for {host}") + return + url = f"https://{host}/api/v1/repos/{org}/{repo}/pulls?state=open" + prs = api_request("GET", url, auth) + if not prs: + print(" No open pull requests.") + for pr in prs: + print(f" PR #{pr['number']}: {pr['title']}") + print(f" Branch: {pr['head']['ref']} -> {pr['base']['ref']}") + print(f" URL: {pr['html_url']}") + print(f" Mergeable: {pr.get('mergeable')}") + except Exception as e: + print(f" Error: {e}") + +if __name__ == '__main__': + main() diff --git a/list_issues.py b/list_issues.py new file mode 100644 index 0000000..43ee497 --- /dev/null +++ b/list_issues.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""List Gitea issues. + +Usage: + list_issues.py --remote dadeschools +""" +import os +import sys +import argparse + +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +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="List Gitea issues.") + add_remote_args(parser) + parser.add_argument("--state", choices=["open", "closed", "all"], default="open", + help="Filter by issue state (default: open).") + parser.add_argument("--label", help="Filter by label name.") + 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 + + params = f"state={args.state}&type=issues" + if args.label: + params += f"&labels={args.label}" + url = f"{repo_api_url(host, org, repo)}/issues?{params}" + + try: + issues = api_request("GET", url, auth) + if not issues: + print("No issues found.") + return 0 + + for i in issues: + labels_str = ", ".join(lb["name"] for lb in i.get("labels", [])) + labels_part = f" [{labels_str}]" if labels_str else "" + assignee = (i.get("assignee") or {}).get("login", "unassigned") + print(f"Issue #{i['number']}: {i['title']}{labels_part}") + print(f" State: {i['state']} | Assignee: {assignee}") + print(f" URL: {i['html_url']}") + return 0 + except Exception as e: + print(f"Error listing issues: {e}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/list_prs.py b/list_prs.py new file mode 100755 index 0000000..e56d233 --- /dev/null +++ b/list_prs.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""List Gitea pull requests. + +Usage: + list_prs.py --remote dadeschools +""" +import os +import sys +import argparse + +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +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="List Gitea pull requests.") + add_remote_args(parser) + parser.add_argument("--state", choices=["open", "closed", "all"], default="open", + help="Filter by PR state (default: open).") + 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)}/pulls?state={args.state}" + + try: + prs = api_request("GET", url, auth) + if not prs: + print("No pull requests found.") + return 0 + + for pr in prs: + merge_status = "Mergeable" if pr.get("mergeable") else "Conflicted/Not Mergeable" + print(f"PR #{pr['number']}: {pr['title']}") + print(f" Branch: {pr['head']['ref']} -> {pr['base']['ref']} ({merge_status})") + print(f" URL: {pr['html_url']}") + return 0 + except Exception as e: + print(f"Error listing PRs: {e}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/manage_labels.py b/manage_labels.py index 6517c94..056b08b 100755 --- a/manage_labels.py +++ b/manage_labels.py @@ -8,8 +8,15 @@ Usage: ./manage_labels.py # create labels, then apply the mapping below ./manage_labels.py --dry # print actions without writing """ +import os import sys +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + from gitea_auth import get_auth_header, api_request, repo_api_url HOST = "gitea.dadeschools.net" diff --git a/mark_issue.py b/mark_issue.py index 1ef00d5..b3a2268 100755 --- a/mark_issue.py +++ b/mark_issue.py @@ -5,9 +5,16 @@ Usage: mark_issue.py [start|done] mark_issue.py --remote prgs 12 done """ +import os import sys import argparse +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + from gitea_auth import ( get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, diff --git a/mcp_server.py b/mcp_server.py index 24d7eda..be59bc2 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -17,8 +17,13 @@ import os import sys import subprocess -# Ensure the project root is on the path so gitea_auth.py can be imported. +# Auto-execute using the project's local virtual environment Python PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +# Ensure the project root is on the path so gitea_auth.py can be imported. if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) @@ -128,6 +133,153 @@ def gitea_create_pr( return {"number": data["number"], "url": data["html_url"]} +@mcp.tool() +def gitea_list_prs( + state: str = "open", + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> list[dict]: + """List pull requests on a Gitea repository. + + Args: + state: State filter — 'open', 'closed', or 'all' (default: 'open'). + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + List of dicts with 'number', 'title', 'state', 'head', 'base', 'url', 'mergeable'. + """ + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + url = f"{repo_api_url(h, o, r)}/pulls?state={state}" + prs = api_request("GET", url, auth) or [] + return [ + { + "number": pr["number"], + "title": pr["title"], + "state": pr["state"], + "head": pr["head"]["ref"], + "base": pr["base"]["ref"], + "url": pr["html_url"], + "mergeable": pr.get("mergeable"), + } + for pr in prs + ] + + +@mcp.tool() +def gitea_view_pr( + pr_number: int, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Get details of a single pull request. + + Args: + pr_number: The pull request index/number. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict with PR details. + """ + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}" + pr = api_request("GET", url, auth) + return { + "number": pr["number"], + "title": pr["title"], + "body": pr.get("body", ""), + "state": pr["state"], + "head": pr["head"]["ref"], + "base": pr["base"]["ref"], + "url": pr["html_url"], + "mergeable": pr.get("mergeable"), + "user": pr.get("user", {}).get("login", ""), + } + + +@mcp.tool() +def gitea_merge_pr( + pr_number: int, + do: str = "merge", + title: str | None = None, + message: str | None = None, + force: bool = False, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Merge a Gitea pull request. + + Args: + pr_number: The PR number to merge. + do: Merge style — 'merge', 'squash', or 'rebase'. + title: Optional merge title. + message: Optional merge message. + force: Force merge, ignoring status checks. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict with 'success' and 'message'. + """ + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/merge" + payload = { + "Do": do, + "force_merge": force, + } + if title: + payload["MergeTitleField"] = title + if message: + payload["MergeMessageField"] = message + api_request("POST", url, auth, payload) + return {"success": True, "message": f"PR #{pr_number} merged via '{do}'."} + + +@mcp.tool() +def gitea_delete_branch( + branch: str, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Delete a remote branch from a Gitea repository. + + Args: + branch: The remote branch name (e.g. 'feat/branch-name'). + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict with 'success' and 'message'. + """ + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + import urllib.parse + encoded_branch = urllib.parse.quote(branch, safe="") + url = f"{repo_api_url(h, o, r)}/branches/{encoded_branch}" + api_request("DELETE", url, auth) + return {"success": True, "message": f"Remote branch '{branch}' deleted."} + + @mcp.tool() def gitea_close_issue( issue_number: int, @@ -288,6 +440,119 @@ def gitea_mark_issue( return {"success": True, "message": f"Issue #{issue_number} released."} +@mcp.tool() +def gitea_list_labels( + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> list: + """List all available labels in a Gitea repository. + + Args: + remote: Known Gitea instance ('dadeschools' or 'prgs'). + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + list of labels. + """ + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + return api_request("GET", f"{base}/labels?limit=100", auth) + + +@mcp.tool() +def gitea_create_label( + name: str, + color: str, + description: str = "", + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Create a new label on a Gitea repository. + + Args: + name: Name of the label (e.g. 'bug', 'epic'). + color: HTML color code (hex, e.g. 'fbca04' or '#fbca04'). + description: Description of the label. + remote: Known Gitea instance ('dadeschools' or 'prgs'). + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict containing the created label details. + """ + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + + if color.startswith("#"): + color = color[1:] + + payload = { + "name": name, + "color": color, + "description": description, + } + return api_request("POST", f"{base}/labels", auth, payload) + + +@mcp.tool() +def gitea_set_issue_labels( + issue_number: int, + labels: list[str], + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> list: + """Replace all labels on a Gitea issue with a new list of label names. + + Args: + issue_number: The issue number. + labels: List of label names to apply. + remote: Known Gitea instance ('dadeschools' or 'prgs'). + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + list of all labels currently applied to the issue. + """ + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + + # 1. Fetch existing labels on the repo to resolve names -> IDs + existing = api_request("GET", f"{base}/labels?limit=100", auth) + name_to_id = {lb["name"]: lb["id"] for lb in existing} + + # 2. Check if any requested labels do not exist, and raise error + label_ids = [] + missing_labels = [] + for name in labels: + if name in name_to_id: + label_ids.append(name_to_id[name]) + else: + missing_labels.append(name) + + if missing_labels: + raise RuntimeError( + f"The following labels do not exist on the repository: {missing_labels}. " + "Please create them first using gitea_create_label." + ) + + # 3. PUT the labels to the issue + res = api_request("PUT", f"{base}/issues/{issue_number}/labels", auth, {"labels": label_ids}) + return res + + @mcp.tool() def gitea_mirror_refs( apply: bool = False, diff --git a/merge_pr.py b/merge_pr.py new file mode 100755 index 0000000..c34c8d4 --- /dev/null +++ b/merge_pr.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""Merge a Gitea pull request. + +Usage: + merge_pr.py --pr-number 84 --do squash --remote dadeschools +""" +import os +import sys +import json +import argparse + +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +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="Merge a Gitea pull request.") + add_remote_args(parser) + parser.add_argument("--pr-number", type=int, required=True, help="PR number/index to merge.") + parser.add_argument("--do", choices=["merge", "squash", "rebase"], default="merge", + help="Merge method/style (default: merge).") + parser.add_argument("--title", help="Optional merge title.") + parser.add_argument("--message", help="Optional merge message.") + parser.add_argument("--force", action="store_true", help="Force merge, ignoring status checks.") + 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 for {host}.", file=sys.stderr) + return 1 + + url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}/merge" + payload = { + "Do": args.do, + "force_merge": args.force, + } + if args.title: + payload["MergeTitleField"] = args.title + if args.message: + payload["MergeMessageField"] = args.message + + try: + api_request("POST", url, auth, payload) + print(f"Successfully merged PR #{args.pr_number} using '{args.do}' method.") + return 0 + except Exception as e: + print(f"Error merging PR #{args.pr_number}: {e}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 4642f08..aa42b8b 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -18,6 +18,10 @@ from mcp_server import ( # noqa: E402 gitea_view_issue, gitea_mark_issue, gitea_mirror_refs, + gitea_list_prs, + gitea_view_pr, + gitea_merge_pr, + gitea_delete_branch, ) FAKE_AUTH = "Basic dGVzdDp0ZXN0" @@ -240,5 +244,83 @@ class TestMirrorRefs(unittest.TestCase): self.assertIn("--force", args) +# --------------------------------------------------------------------------- +# List PRs +# --------------------------------------------------------------------------- +class TestListPRs(unittest.TestCase): + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_prs(self, _auth, mock_api): + mock_api.return_value = [ + { + "number": 1, "title": "PR 1", "state": "open", + "head": {"ref": "branch1"}, "base": {"ref": "main"}, + "html_url": "http://url1", "mergeable": True + } + ] + result = gitea_list_prs() + self.assertEqual(len(result), 1) + self.assertEqual(result[0]["number"], 1) + self.assertEqual(result[0]["head"], "branch1") + + +# --------------------------------------------------------------------------- +# View PR +# --------------------------------------------------------------------------- +class TestViewPR(unittest.TestCase): + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_view_pr(self, _auth, mock_api): + mock_api.return_value = { + "number": 1, "title": "PR 1", "state": "open", + "head": {"ref": "branch1"}, "base": {"ref": "main"}, + "html_url": "http://url1", "mergeable": True, "body": "description", + "user": {"login": "user1"} + } + result = gitea_view_pr(pr_number=1) + self.assertEqual(result["number"], 1) + self.assertEqual(result["body"], "description") + self.assertEqual(result["user"], "user1") + + +# --------------------------------------------------------------------------- +# Merge PR +# --------------------------------------------------------------------------- +class TestMergePR(unittest.TestCase): + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_merge_pr(self, _auth, mock_api): + mock_api.return_value = {} + result = gitea_merge_pr(pr_number=1, do="squash", title="T", message="M", force=True) + self.assertTrue(result["success"]) + self.assertIn("merged", result["message"]) + # Check payload + payload = mock_api.call_args[0][3] + self.assertEqual(payload["Do"], "squash") + self.assertEqual(payload["MergeTitleField"], "T") + self.assertEqual(payload["MergeMessageField"], "M") + self.assertEqual(payload["force_merge"], True) + + +# --------------------------------------------------------------------------- +# Delete Branch +# --------------------------------------------------------------------------- +class TestDeleteBranch(unittest.TestCase): + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_delete_branch(self, _auth, mock_api): + mock_api.return_value = {} + result = gitea_delete_branch(branch="feat/branch") + self.assertTrue(result["success"]) + self.assertIn("deleted", result["message"]) + # Check url encoding of branch name + url = mock_api.call_args[0][1] + self.assertIn("feat%2Fbranch", url) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_merge_pr.py b/tests/test_merge_pr.py new file mode 100644 index 0000000..66a7ca9 --- /dev/null +++ b/tests/test_merge_pr.py @@ -0,0 +1,64 @@ +"""Tests for merge_pr.py. + +Mocks api_request and credentials. +""" +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) +import merge_pr # noqa: E402 + + +FAKE_CREDS = "Basic dGVzdHVzZXI6dGVzdHBhc3M=" + + +class TestArgParsing(unittest.TestCase): + + @patch("merge_pr.api_request") + @patch("merge_pr.get_auth_header", return_value=FAKE_CREDS) + def test_minimal_required_args(self, _auth, mock_api): + rc = merge_pr.main(["--pr-number", "81"]) + self.assertEqual(rc, 0) + mock_api.assert_called_once() + + def test_missing_pr_number_exits(self): + with self.assertRaises(SystemExit): + merge_pr.main([]) + + +class TestAPIPayload(unittest.TestCase): + + @patch("merge_pr.api_request") + @patch("merge_pr.get_auth_header", return_value=FAKE_CREDS) + def test_payload_fields(self, _auth, mock_api): + rc = merge_pr.main([ + "--pr-number", "81", + "--do", "squash", + "--title", "Squash title", + "--message", "Squash message", + "--force", + ]) + self.assertEqual(rc, 0) + mock_api.assert_called_once() + method, url, auth, payload = mock_api.call_args[0] + self.assertEqual(method, "POST") + self.assertEqual(auth, FAKE_CREDS) + self.assertEqual(payload["Do"], "squash") + self.assertEqual(payload["MergeTitleField"], "Squash title") + self.assertEqual(payload["MergeMessageField"], "Squash message") + self.assertEqual(payload["force_merge"], True) + + @patch("merge_pr.api_request") + @patch("merge_pr.get_auth_header", return_value=FAKE_CREDS) + def test_url_construction(self, _auth, mock_api): + merge_pr.main(["--pr-number", "81", "--remote", "prgs"]) + url = mock_api.call_args[0][1] + self.assertEqual( + url, + "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Timesheet/pulls/81/merge" + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_prs.py b/tests/test_prs.py new file mode 100644 index 0000000..4d1fdc1 --- /dev/null +++ b/tests/test_prs.py @@ -0,0 +1,68 @@ +"""Tests for list_prs.py, view_pr.py, and delete_branch.py. + +Mocks api_request and credentials. +""" +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) +import list_prs # noqa: E402 +import view_pr # noqa: E402 +import delete_branch # noqa: E402 + + +FAKE_CREDS = "Basic dGVzdHVzZXI6dGVzdHBhc3M=" + + +class TestListPRs(unittest.TestCase): + + @patch("list_prs.api_request") + @patch("list_prs.get_auth_header", return_value=FAKE_CREDS) + def test_list_prs_success(self, _auth, mock_api): + mock_api.return_value = [ + {"number": 1, "title": "PR 1", "head": {"ref": "branch1"}, "base": {"ref": "main"}, "html_url": "http://url1", "mergeable": True} + ] + rc = list_prs.main([]) + self.assertEqual(rc, 0) + mock_api.assert_called_once() + + @patch("list_prs.api_request", return_value=[]) + @patch("list_prs.get_auth_header", return_value=FAKE_CREDS) + def test_list_prs_empty(self, _auth, mock_api): + rc = list_prs.main(["--state", "closed"]) + self.assertEqual(rc, 0) + mock_api.assert_called_once() + + +class TestViewPR(unittest.TestCase): + + @patch("view_pr.api_request") + @patch("view_pr.get_auth_header", return_value=FAKE_CREDS) + def test_view_pr_success(self, _auth, mock_api): + mock_api.return_value = {"number": 1, "title": "PR 1", "state": "open", "user": {"login": "user1"}, "head": {"ref": "branch1"}, "base": {"ref": "main"}, "html_url": "http://url1", "mergeable": True, "body": "PR description"} + rc = view_pr.main(["1"]) + self.assertEqual(rc, 0) + mock_api.assert_called_once() + + def test_missing_pr_number_exits(self): + with self.assertRaises(SystemExit): + view_pr.main([]) + + +class TestDeleteBranch(unittest.TestCase): + + @patch("delete_branch.api_request") + @patch("delete_branch.get_auth_header", return_value=FAKE_CREDS) + def test_delete_branch_success(self, _auth, mock_api): + rc = delete_branch.main(["feat/my-branch"]) + self.assertEqual(rc, 0) + mock_api.assert_called_once() + + def test_missing_branch_exits(self): + with self.assertRaises(SystemExit): + delete_branch.main([]) + + +if __name__ == "__main__": + unittest.main() diff --git a/view_pr.py b/view_pr.py new file mode 100755 index 0000000..ecc2925 --- /dev/null +++ b/view_pr.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""View details of a Gitea pull request. + +Usage: + view_pr.py + view_pr.py --remote prgs 12 +""" +import os +import sys +import argparse + +# Auto-execute using the project's local virtual environment Python +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") +if os.path.exists(venv_python) and sys.executable != venv_python: + os.execv(venv_python, [venv_python] + sys.argv) + +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="View details of a Gitea pull request.") + add_remote_args(parser) + parser.add_argument("pr_number", type=int, help="PR number to view.") + 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)}/pulls/{args.pr_number}" + + try: + pr = api_request("GET", url, auth) + print(f"PR #{pr['number']}: {pr['title']}") + print(f"Status: {pr['state']}") + print(f"Created by: {pr.get('user', {}).get('login', 'unknown')}") + print(f"Branches: {pr['head']['ref']} -> {pr['base']['ref']}") + print(f"Mergeable: {pr.get('mergeable')}") + print(f"URL: {pr['html_url']}") + print("\nDescription:") + print(pr.get("body") or "(No description)") + return 0 + except Exception as e: + print(f"Error viewing PR #{args.pr_number}: {e}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main())