root/homelab-mcp-gitea
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initial homelab-mcp-gitea v0.1

Browse Source 
MCP server exposing self-hosted Gitea (gitlab.impresion3d.pro) via stdio
JSON-RPC 2.0. Six tools: list_repos, get_repo, get_file, create_issue,
list_open_prs, merge_pr. The merge_pr tool pins the Gitea 'do'-field trap
(see skill gitea pitfall #13) — sends lowercase 'do', treats HTTP 200
with non-empty body as failure.

Stack: FastMCP + httpx + pydantic + uv, with respx-mocked pytest suite
(15 tests, 0.69s) and ruff + mypy strict green.

Verification: scripts/verify_stdio.py boots the server as a subprocess,
exchanges real MCP messages over stdio, and calls list_repos against the
live Gitea instance — all green.
This commit is contained in:
homelab-mcp
2026-07-22 12:47:59 +00:00
commit 0212913831
12 changed files with 2218 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+3
View File
@@ -0,0 +1,3 @@
# Copy to .env and fill in. The server reads these on every tool call.
GITEA_URL=https://gitlab.impresion3d.pro
GITEA_TOKEN=replace-with-your-40-char-personal-access-token
.gitignore
+27
View File
@@ -0,0 +1,27 @@
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
# Editor / OS
.vscode/
.idea/
.DS_Store
# Local secrets — never commit. Copy .env.example to .env.
.env
# Coverage / test artifacts
.coverage
htmlcov/
.pytest_cache/
.mypy_cache/
.ruff_cache/
# uv lock is committed; the build artifacts are not.
.python-version
+1
View File
@@ -0,0 +1 @@
3.12
README.md
+131
View File
@@ -0,0 +1,131 @@
# homelab-mcp-gitea
A **Model Context Protocol (MCP) server** that exposes a self-hosted **Gitea**
instance as a set of tools, so any MCP-aware client (Claude Desktop, Hermes
Agent, mcp-cli, the MCP Inspector) can list repos, fetch files, open issues,
list PRs, and merge PRs through natural language — without hand-rolling
`curl` invocations every time.
This is the **first** package of a planned `homelab-mcp-*` family:
`homelab-mcp-gitea` → `homelab-mcp-portainer` → `homelab-mcp-nas` … each one
a thin MCP wrapper over an existing self-hosted service.
## What is MCP, in one paragraph
MCP (Model Context Protocol) is the open standard (Anthropic, late 2024) that
lets an LLM client talk to "tool servers" the same way regardless of which
host it runs on. A server exposes:
- **tools** (named functions with typed arguments and return values)
- **resources** (named blobs the client can read)
- **prompts** (templated prompts the client can render)
…over **stdio** or **HTTP+SSE**, using **JSON-RPC 2.0** as the wire format.
The protocol is deliberately minimal so any language can implement it; the
Python and TypeScript SDKs are the most mature.
In this project, each `GiteaClient` method becomes an MCP `tool` decorated
with `@mcp.tool()`. The server runs over stdio (the default and simplest
transport); the client launches it as a subprocess and exchanges JSON-RPC
messages on stdin/stdout.
## Tools exposed
| Tool | What it does |
|-----------------|---------------------------------------------------------|
| `list_repos` | List repos visible to the authenticated user |
| `get_repo` | Get one repo's metadata |
| `get_file` | Read a file's text content (base64-decoded) |
| `create_issue` | Open a new issue on a repo |
| `list_open_prs` | List open pull requests for a repo |
| `merge_pr` | Merge a PR by index (with the Gitea `do` field trap fix)|
The `merge_pr` tool is non-trivial on purpose: Gitea's merge endpoint is
the famous `[Do]: Required` footgun (see the `gitea` skill, pitfall #13).
The client sends `{"do": "merge"}` (lowercase, two chars) and treats a
non-empty response body at HTTP 200 as failure. The tests pin both halves
of this contract so it can't regress.
## Setup
```bash
cd ~/homelab-mcp-gitea
uv sync --all-groups
cp .env.example .env
# Edit .env: set GITEA_URL and GITEA_TOKEN
uv run pytest # 14 tests, no network required
uv run ruff check src tests
uv run mypy src
```
## Running the server
```bash
# Foreground stdio transport (the default — what MCP clients expect)
uv run homelab-mcp-gitea
```
The server reads `GITEA_URL` and `GITEA_TOKEN` from the environment on every
tool call (no global state), so the same server process can be reused across
requests without leaking credentials between calls.
## Connecting from a client
### Claude Desktop (`~/.config/claude_desktop_config.json` on Linux)
```json
{
"mcpServers": {
"homelab-gitea": {
"command": "uv",
"args": ["--directory", "/home/caleidos/homelab-mcp-gitea", "run", "homelab-mcp-gitea"],
"env": {
"GITEA_URL": "https://gitlab.impresion3d.pro",
"GITEA_TOKEN": "<your-token>"
}
}
}
}
```
Restart Claude Desktop. The `homelab-gitea` server appears with its 6 tools
under the "tools" menu.
### MCP Inspector (browser-based debugger)
```bash
npx @modelcontextprotocol/inspector uv --directory ~/homelab-mcp-gitea run homelab-mcp-gitea
```
Opens a web UI where you can call each tool by hand and see the raw JSON-RPC.
## Architecture
```
src/homelab_mcp_gitea/
client.py # httpx-based Gitea REST client, no MCP knowledge
server.py # FastMCP server; thin wrapper that maps tools -> client calls
tests/
test_client.py # respx-mocked unit tests for every HTTP call
test_server.py # tool registration + routing smoke tests
```
The split is deliberate: `client.py` knows nothing about MCP, so it can be
reused from a plain Python REPL or a CLI, and `server.py` is small enough
to read in one sitting (~120 lines including docstrings).
## What's NOT here (and why)
- **No FastAPI / Celery / Alembic / Docker**. Those belong to the
`python-project-template-internal` stack, which targets deployable web
apps. An MCP server is a long-lived subprocess, not an HTTP service.
- **No CI-to-NAS deploy**. Local-only for v0.1; we can add Gitea Actions +
SSH deploy when there's a reason to keep the server running unattended.
- **No auth scopes, no rate limiting, no audit log**. The Gitea token
scopes are enforced server-side; this MCP just forwards.
## Next steps in the `homelab-mcp-*` family
- `homelab-mcp-portainer`: list containers, restart, tail logs, redeploy
stack. Requires the Portainer token + endpoint ID.
- `homelab-mcp-nas`: SSH-based filesystem + Container Station ops. The
trickier one (askpass / key auth, see `qnap-nas` skill pitfall J).
pyproject.toml
+50
View File
@@ -0,0 +1,50 @@
[project]
name = "homelab-mcp-gitea"
version = "0.1.0"
description = "MCP server that exposes a self-hosted Gitea instance to MCP-aware clients"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"httpx>=0.27",
"mcp[cli]>=1.2",
"pydantic>=2.7",
]
[project.scripts]
homelab-mcp-gitea = "homelab_mcp_gitea.server:main"
[dependency-groups]
dev = [
"pytest>=8.3",
"pytest-asyncio>=0.24",
"pytest-cov>=6",
"ruff>=0.7",
"mypy>=1.13",
"respx>=0.21",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/homelab_mcp_gitea"]
[tool.ruff]
line-length = 100
target-version = "py311"
[tool.ruff.lint]
select = ["E", "F", "W", "I", "N", "UP", "B", "SIM", "RUF"]
ignore = ["E501"]
[tool.mypy]
python_version = "3.11"
strict = true
warn_unused_ignores = true
disallow_untyped_defs = true
[tool.pytest.ini_options]
addopts = "-ra --strict-markers"
asyncio_mode = "auto"
testpaths = ["tests"]
scripts/verify_stdio.py
+122
View File
@@ -0,0 +1,122 @@
"""End-to-end stdio verification.
Boots the MCP server as a subprocess, exchanges real JSON-RPC 2.0 messages
over stdio (the same transport Claude Desktop / mcp-cli use), and verifies:
1. The initialize handshake completes and reports our 6 tools.
2. tools/list returns the same 6 tools with proper JSON Schema.
3. tools/call (list_repos) hits the real Gitea instance and returns data.
This is the closest you can get to "what does a real MCP client see?"
without installing one. If this script passes, the server speaks the
protocol correctly.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any
def run_e2e() -> None:
env = os.environ.copy()
if "GITEA_URL" not in env or "GITEA_TOKEN" not in env:
print("ERROR: GITEA_URL and GITEA_TOKEN must be in env.", file=sys.stderr)
sys.exit(2)
cmd = ["uv", "run", "homelab-mcp-gitea"]
proc = subprocess.Popen(
cmd,
cwd=Path(__file__).parent.parent,
env=env,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
def send(msg: dict[str, Any]) -> dict[str, Any]:
line = json.dumps(msg) + "\n"
assert proc.stdin is not None
proc.stdin.write(line)
proc.stdin.flush()
assert proc.stdout is not None
response_line = proc.stdout.readline()
return json.loads(response_line)
try:
# 1) initialize
init_resp = send(
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "stdio-verifier", "version": "0.1.0"},
},
}
)
assert "result" in init_resp, f"initialize failed: {init_resp}"
server_info = init_resp["result"]["serverInfo"]
print(f"server: {server_info['name']} v{server_info['version']}")
# MCP requires the initialized notification before tool calls
assert proc.stdin is not None
proc.stdin.write(
json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}) + "\n"
)
proc.stdin.flush()
# 2) tools/list
tools_resp = send({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}})
assert "result" in tools_resp, f"tools/list failed: {tools_resp}"
tools = tools_resp["result"]["tools"]
tool_names = sorted(t["name"] for t in tools)
print(f"tools registered ({len(tools)}): {tool_names}")
expected = {
"create_issue",
"get_file",
"get_repo",
"list_open_prs",
"list_repos",
"merge_pr",
}
assert set(tool_names) == expected, f"unexpected tools: {tool_names}"
# 3) tools/call list_repos
call_resp = send(
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {"name": "list_repos", "arguments": {"limit": 3}},
}
)
assert "result" in call_resp, f"tools/call failed: {call_resp}"
content = call_resp["result"]["content"]
assert content and content[0]["type"] == "text"
text = content[0]["text"]
print("list_repos returned:")
for line in text.splitlines()[:5]:
print(f" {line}")
print("\nALL CHECKS PASSED — MCP server speaks the protocol correctly.")
finally:
proc.terminate()
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
proc.kill()
stderr = proc.stderr.read() if proc.stderr else ""
if stderr.strip():
print("\n--- server stderr ---")
print(stderr)
if __name__ == "__main__":
run_e2e()
src/homelab_mcp_gitea/__init__.py
+2
View File
@@ -0,0 +1,2 @@
def main() -> None:
print("Hello from homelab-mcp-gitea!")
src/homelab_mcp_gitea/client.py
+150
View File
@@ -0,0 +1,150 @@
"""HTTP client for the Gitea REST API.
All operations live here so the MCP server layer stays thin and the client
can be unit-tested with respx (httpx MockTransport) without touching the MCP
machinery.
Auth: read GITEA_URL and GITEA_TOKEN from environment. The token is sent as
'Authorization: token <value>' per the Gitea API convention.
"""
from __future__ import annotations
import base64
import os
from typing import Any, cast
import httpx
class GiteaError(RuntimeError):
"""Raised when the Gitea API returns a non-2xx response or a known trap."""
def __init__(self, status: int, message: str, body: Any = None) -> None:
super().__init__(f"HTTP {status}: {message}")
self.status = status
self.body = body
class GiteaClient:
"""Thin wrapper over httpx for the Gitea /api/v1/ endpoints we use."""
def __init__(self, base_url: str | None = None, token: str | None = None) -> None:
self.base_url = (base_url or os.environ.get("GITEA_URL", "")).rstrip("/")
token = token or os.environ.get("GITEA_TOKEN", "")
if not self.base_url:
raise ValueError("GITEA_URL is required (env var or constructor arg)")
if not token:
raise ValueError("GITEA_TOKEN is required (env var or constructor arg)")
self._client = httpx.AsyncClient(
base_url=self.base_url,
headers={"Authorization": f"token {token}"},
timeout=30.0,
)
async def aclose(self) -> None:
await self._client.aclose()
async def __aenter__(self) -> GiteaClient:
return self
async def __aexit__(self, *exc: object) -> None:
await self.aclose()
# ----- low-level -----
async def _request_json(
self,
method: str,
path: str,
*,
params: dict[str, Any] | None = None,
json_body: dict[str, Any] | None = None,
) -> Any:
response = await self._client.request(
method,
path,
params=params,
json=json_body,
)
if response.status_code >= 400:
# Gitea returns either JSON {"message": "..."} or plain text
try:
body = response.json()
except Exception:
body = response.text
raise GiteaError(response.status_code, str(body), body)
if not response.content:
return None
return response.json()
# ----- repositories -----
async def list_repos(self, limit: int = 30) -> list[dict[str, Any]]:
"""List repositories visible to the authenticated user."""
data = await self._request_json("GET", "/api/v1/user/repos", params={"limit": limit})
return data or []
async def get_repo(self, owner: str, repo: str) -> dict[str, Any]:
"""Fetch a single repo by owner/name."""
return cast(
dict[str, Any], await self._request_json("GET", f"/api/v1/repos/{owner}/{repo}")
)
async def get_file(self, owner: str, repo: str, path: str, ref: str = "main") -> str:
"""Fetch a file's raw text content. Gitea returns base64; we decode."""
data = await self._request_json(
"GET",
f"/api/v1/repos/{owner}/{repo}/contents/{path}",
params={"ref": ref},
)
if not data or "content" not in data:
raise GiteaError(404, f"file not found: {path}")
return base64.b64decode(data["content"]).decode("utf-8")
# ----- issues -----
async def create_issue(
self, owner: str, repo: str, title: str, body: str = ""
) -> dict[str, Any]:
"""Open a new issue. Returns the issue object (includes 'number')."""
return cast(
dict[str, Any],
await self._request_json(
"POST",
f"/api/v1/repos/{owner}/{repo}/issues",
json_body={"title": title, "body": body},
),
)
# ----- pull requests -----
async def list_open_prs(self, owner: str, repo: str) -> list[dict[str, Any]]:
"""List pull requests in 'open' state for the given repo."""
data = await self._request_json(
"GET", f"/api/v1/repos/{owner}/{repo}/pulls", params={"state": "open"}
)
return data or []
async def merge_pr(self, owner: str, repo: str, index: int) -> bool:
"""Merge a PR. Returns True on success.
Trap (from skill gitea): the field is the lowercase key `do`, NOT `Do`
or `Merge_Method`. Gitea returns HTTP 200 with a misleading body when
the field name is wrong. Success body is empty.
"""
response = await self._client.post(
f"/api/v1/repos/{owner}/{repo}/pulls/{index}/merge",
json={"do": "merge"},
)
if response.status_code >= 400:
raise GiteaError(response.status_code, response.text)
# Success = HTTP 200 with EMPTY body. Non-empty = server-side error
# even when status is 200 (the classic Gitea trap).
if response.content:
raise GiteaError(
response.status_code,
f"merge returned non-empty body (likely the 'do' field trap): {response.text!r}",
response.json() if response.text else None,
)
return True
src/homelab_mcp_gitea/server.py
+167
View File
@@ -0,0 +1,167 @@
"""MCP server exposing Gitea operations as tools.
Run with:
uv run homelab-mcp-gitea
The server uses stdio transport (the default for MCP). Connect from any MCP
client (Claude Desktop, mcp-cli, inspector) by pointing it at this command.
"""
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
from .client import GiteaClient, GiteaError
mcp = FastMCP("homelab-gitea")
def _client() -> GiteaClient:
"""Fresh client per call — MCP tools are stateless and short-lived."""
return GiteaClient()
def _format_error(e: GiteaError) -> str:
return f"Gitea API error: {e}"
# ----- tools -----
@mcp.tool()
async def list_repos(limit: int = 30) -> str:
"""List repositories the authenticated user can see.
Args:
limit: Max number of repos to return (default 30).
"""
try:
async with _client() as c:
repos = await c.list_repos(limit=limit)
except GiteaError as e:
return _format_error(e)
except ValueError as e:
return f"Configuration error: {e}"
if not repos:
return "(no repositories found)"
lines = [f"{r['full_name']} [private={r['private']}]" for r in repos]
return "\n".join(lines)
@mcp.tool()
async def get_repo(owner: str, repo: str) -> str:
"""Fetch a single repository's metadata.
Args:
owner: Gitea login (e.g. 'root').
repo: Repository name.
"""
try:
async with _client() as c:
data = await c.get_repo(owner, repo)
except GiteaError as e:
return _format_error(e)
except ValueError as e:
return f"Configuration error: {e}"
return (
f"{data['full_name']}\n"
f" description : {data.get('description') or '(none)'}\n"
f" private : {data['private']}\n"
f" default : {data['default_branch']}\n"
f" stars : {data['stars_count']}\n"
f" open issues : {data['open_issues_count']}"
)
@mcp.tool()
async def get_file(owner: str, repo: str, path: str, ref: str = "main") -> str:
"""Fetch a file's text content from a repository.
Args:
owner: Gitea login.
repo: Repository name.
path: Path inside the repo (e.g. 'README.md').
ref: Branch or tag (default 'main').
"""
try:
async with _client() as c:
content = await c.get_file(owner, repo, path, ref=ref)
except GiteaError as e:
return _format_error(e)
except ValueError as e:
return f"Configuration error: {e}"
return content
@mcp.tool()
async def create_issue(owner: str, repo: str, title: str, body: str = "") -> str:
"""Open a new issue on a repository.
Args:
owner: Gitea login.
repo: Repository name.
title: Issue title.
body: Optional issue body (markdown).
"""
try:
async with _client() as c:
issue = await c.create_issue(owner, repo, title, body)
except GiteaError as e:
return _format_error(e)
except ValueError as e:
return f"Configuration error: {e}"
return f"Created issue #{issue['number']}: {issue['title']}\n url: {issue['html_url']}"
@mcp.tool()
async def list_open_prs(owner: str, repo: str) -> str:
"""List open pull requests for a repository.
Args:
owner: Gitea login.
repo: Repository name.
"""
try:
async with _client() as c:
prs = await c.list_open_prs(owner, repo)
except GiteaError as e:
return _format_error(e)
except ValueError as e:
return f"Configuration error: {e}"
if not prs:
return f"(no open PRs in {owner}/{repo})"
lines = [
f"#{pr['number']} {pr['title']} ({pr['user']['login']} -> {pr['base']['ref']})"
for pr in prs
]
return "\n".join(lines)
@mcp.tool()
async def merge_pr(owner: str, repo: str, index: int) -> str:
"""Merge a pull request.
Args:
owner: Gitea login.
repo: Repository name.
index: PR number (the URL-visible number, NOT the internal id).
"""
try:
async with _client() as c:
ok = await c.merge_pr(owner, repo, index)
except GiteaError as e:
return _format_error(e)
except ValueError as e:
return f"Configuration error: {e}"
if ok:
return f"PR #{index} merged successfully."
return f"PR #{index} merge returned ok=False (unexpected)."
def main() -> None:
"""Entry point for the 'homelab-mcp-gitea' script."""
mcp.run()
if __name__ == "__main__":
main()
tests/test_client.py
+171
View File
@@ -0,0 +1,171 @@
"""Unit tests for the Gitea HTTP client. Uses respx to mock httpx."""
from __future__ import annotations
import base64
import httpx
import pytest
import respx
from homelab_mcp_gitea.client import GiteaClient, GiteaError
BASE = "https://gitea.test"
TOKEN = "test-token-xyz"
def _client() -> GiteaClient:
return GiteaClient(base_url=BASE, token=TOKEN)
@pytest.mark.asyncio
@respx.mock
async def test_list_repos_returns_parsed_json() -> None:
respx.get(f"{BASE}/api/v1/user/repos").mock(
return_value=httpx.Response(
200,
json=[
{"full_name": "root/alpha", "private": False},
{"full_name": "root/beta", "private": True},
],
)
)
async with _client() as c:
repos = await c.list_repos(limit=10)
assert len(repos) == 2
assert repos[0]["full_name"] == "root/alpha"
@pytest.mark.asyncio
@respx.mock
async def test_get_repo_returns_metadata() -> None:
respx.get(f"{BASE}/api/v1/repos/root/alpha").mock(
return_value=httpx.Response(
200,
json={
"full_name": "root/alpha",
"description": "test repo",
"private": False,
"default_branch": "main",
"stars_count": 7,
"open_issues_count": 2,
},
)
)
async with _client() as c:
data = await c.get_repo("root", "alpha")
assert data["stars_count"] == 7
@pytest.mark.asyncio
@respx.mock
async def test_get_file_decodes_base64_content() -> None:
raw = "hello world\n"
encoded = base64.b64encode(raw.encode()).decode()
respx.get(f"{BASE}/api/v1/repos/root/alpha/contents/README.md").mock(
return_value=httpx.Response(200, json={"content": encoded})
)
async with _client() as c:
text = await c.get_file("root", "alpha", "README.md")
assert text == raw
@pytest.mark.asyncio
@respx.mock
async def test_create_issue_posts_payload_and_returns_issue() -> None:
respx.post(f"{BASE}/api/v1/repos/root/alpha/issues").mock(
return_value=httpx.Response(
201,
json={
"number": 42,
"title": "bug",
"html_url": "https://gitea.test/root/alpha/issues/42",
},
)
)
async with _client() as c:
issue = await c.create_issue("root", "alpha", "bug", "details")
assert issue["number"] == 42
sent = respx.calls.last.request
import json
payload = json.loads(sent.content)
assert payload == {"title": "bug", "body": "details"}
@pytest.mark.asyncio
@respx.mock
async def test_list_open_prs_filters_by_state() -> None:
respx.get(f"{BASE}/api/v1/repos/root/alpha/pulls").mock(
return_value=httpx.Response(
200,
json=[
{
"number": 5,
"title": "feat: x",
"user": {"login": "alice"},
"base": {"ref": "main"},
}
],
)
)
async with _client() as c:
prs = await c.list_open_prs("root", "alpha")
assert prs[0]["number"] == 5
sent = respx.calls.last.request
assert sent.url.params["state"] == "open"
@pytest.mark.asyncio
@respx.mock
async def test_merge_pr_uses_lowercase_do_field() -> None:
"""Trap regression: Gitea needs the JSON key `do`, not `Do` or `Merge_Method`."""
respx.post(f"{BASE}/api/v1/repos/root/alpha/pulls/5/merge").mock(
return_value=httpx.Response(200, content=b"")
)
async with _client() as c:
ok = await c.merge_pr("root", "alpha", 5)
assert ok is True
import json
sent_payload = json.loads(respx.calls.last.request.content)
assert sent_payload == {"do": "merge"}
@pytest.mark.asyncio
@respx.mock
async def test_merge_pr_raises_on_nonempty_body_even_at_200() -> None:
"""The other half of the trap: status 200 with a body means failure."""
respx.post(f"{BASE}/api/v1/repos/root/alpha/pulls/5/merge").mock(
return_value=httpx.Response(
200,
json={"message": "[Do]: Required"},
)
)
async with _client() as c:
with pytest.raises(GiteaError) as exc_info:
await c.merge_pr("root", "alpha", 5)
assert "do" in str(exc_info.value).lower()
@pytest.mark.asyncio
@respx.mock
async def test_http_error_raises_gitea_error() -> None:
respx.get(f"{BASE}/api/v1/repos/root/missing").mock(
return_value=httpx.Response(404, json={"message": "not found"})
)
async with _client() as c:
with pytest.raises(GiteaError) as exc_info:
await c.get_repo("root", "missing")
assert exc_info.value.status == 404
def test_constructor_requires_url_and_token() -> None:
import os
os.environ.pop("GITEA_URL", None)
os.environ.pop("GITEA_TOKEN", None)
with pytest.raises(ValueError, match="GITEA_URL"):
GiteaClient(token="t")
with pytest.raises(ValueError, match="GITEA_TOKEN"):
GiteaClient(base_url=BASE)
tests/test_server.py
+115
View File
@@ -0,0 +1,115 @@
"""Smoke tests for the MCP server layer.
Uses FastMCP's public API:
- mcp.list_tools() to enumerate registered tools
- mcp.call_tool(name, args) to invoke one
This avoids reaching into private attributes and reflects exactly what an
MCP client sees over the wire.
"""
from __future__ import annotations
import base64
import httpx
import pytest
import respx
from homelab_mcp_gitea import server
BASE = "https://gitea.test"
@pytest.fixture(autouse=True)
def _env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", BASE)
monkeypatch.setenv("GITEA_TOKEN", "test-token")
def _text(call_result: object) -> str:
"""Extract the first text block from a FastMCP call_tool result."""
content_blocks, _meta = call_result # type: ignore[misc]
return content_blocks[0].text # type: ignore[union-attr]
@pytest.mark.asyncio
async def test_all_six_tools_registered() -> None:
tools = await server.mcp.list_tools()
names = {t.name for t in tools}
expected = {
"list_repos",
"get_repo",
"get_file",
"create_issue",
"list_open_prs",
"merge_pr",
}
assert expected <= names, f"missing tools: {expected - names}"
@pytest.mark.asyncio
@respx.mock
async def test_list_repos_tool_routes_through_client() -> None:
respx.get(f"{BASE}/api/v1/user/repos").mock(
return_value=httpx.Response(200, json=[{"full_name": "root/alpha", "private": False}])
)
result = await server.mcp.call_tool("list_repos", {"limit": 10})
assert "root/alpha" in _text(result)
@pytest.mark.asyncio
@respx.mock
async def test_get_file_tool_returns_decoded_text() -> None:
raw = "from fastapi import FastAPI\n"
encoded = base64.b64encode(raw.encode()).decode()
respx.get(f"{BASE}/api/v1/repos/root/alpha/contents/main.py").mock(
return_value=httpx.Response(200, json={"content": encoded})
)
result = await server.mcp.call_tool(
"get_file", {"owner": "root", "repo": "alpha", "path": "main.py", "ref": "main"}
)
assert _text(result) == raw
@pytest.mark.asyncio
@respx.mock
async def test_create_issue_tool_returns_number_and_url() -> None:
respx.post(f"{BASE}/api/v1/repos/root/alpha/issues").mock(
return_value=httpx.Response(
201,
json={
"number": 7,
"title": "x",
"html_url": "https://gitea.test/root/alpha/issues/7",
},
)
)
result = await server.mcp.call_tool(
"create_issue", {"owner": "root", "repo": "alpha", "title": "x", "body": ""}
)
text = _text(result)
assert "#7" in text
assert "issues/7" in text
@pytest.mark.asyncio
@respx.mock
async def test_merge_pr_tool_success_message() -> None:
respx.post(f"{BASE}/api/v1/repos/root/alpha/pulls/5/merge").mock(
return_value=httpx.Response(200, content=b"")
)
result = await server.mcp.call_tool("merge_pr", {"owner": "root", "repo": "alpha", "index": 5})
assert "merged successfully" in _text(result)
@pytest.mark.asyncio
@respx.mock
async def test_error_from_gitea_surfaces_as_human_string() -> None:
respx.get(f"{BASE}/api/v1/user/repos").mock(
return_value=httpx.Response(401, json={"message": "unauthorized"})
)
result = await server.mcp.call_tool("list_repos", {})
text = _text(result)
assert "Gitea API error" in text
assert "401" in text
uv.lock
Generated
+1279
View File
File diff suppressed because it is too large Load Diff