"""The Teahub MCP against a fake Gitea. From mcp/: python -m unittest discover -s tests""" from __future__ import annotations import base64 import json import os import sys import tempfile import unittest from unittest import mock sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) os.environ.update({"GITEA_URL": "https://gitea.test", "PUBLIC_URL": "https://mcp.test", "GITEA_OAUTH_CLIENT_ID": "id", "GITEA_OAUTH_CLIENT_SECRET": "not-a-secret", "JWT_SIGNING_KEY": "k" * 32, "FASTMCP_HOME": tempfile.mkdtemp()}) import httpx # noqa: E402 from fastmcp import Client # noqa: E402 from fastmcp.exceptions import ToolError # noqa: E402 from tea_mcp import server # noqa: E402 R = "repos/o/r" READS = {"tea_whoami", "tea_issues", "tea_issue_get", "tea_issue_comments", "tea_labels", "tea_pull_get", "tea_repo", "tea_tree", "tea_file_get", "tea_actions", "tea_action_logs", "tea_domains"} WRITES = {"tea_issue_create", "tea_issue_edit", "tea_issue_comment", "tea_label", "tea_pull_create", "tea_pull_review", "tea_pull_merge", "tea_repo_edit", "tea_repo_create", "tea_org_repo_create", "tea_org_create", "tea_org_add_member", "tea_hook_create", "tea_file_change", "tea_file_edit", "tea_commit", "tea_branch_create", "tea_workflow_dispatch", "tea_secret_set", "tea_variable_set"} AGAIN = {"tea_issue_edit", "tea_label", "tea_repo_edit", "tea_org_add_member", "tea_hook_create", "tea_secret_set", "tea_variable_set"} def b64(data: bytes) -> str: return base64.b64encode(data).decode() def answer(data=None, status=200, total=None) -> httpx.Response: return httpx.Response(status, json=data, headers={"x-total-count": str(total)} if total is not None else None) class FakeGitea: """Just enough of Gitea's API. Every call is kept as (method, path, query, json body).""" def __init__(self): self.calls: list[tuple[str, str, dict, object]] = [] self.raw_paths: list[bytes] = [] self.repo_count = 230 self.files = {"README.md": b"hello\n", "img/logo.png": b"\x89PNG\r\n\x1a\n\xff", "docs/a b.md": b"spaced\n"} self.labels = [{"id": 1, "name": "bug", "color": "ee0701", "description": "broken"}, {"id": 2, "name": "docs", "color": "0075ca", "description": ""}] self.on_issue = [self.labels[0]] self.comments = [{"id": i, "user": {"login": "tea"}, "body": f"c{i}"} for i in range(1, 46)] self.variables = {"EXISTING": "1"} self.hooks = [{"id": 7, "config": {"url": "https://pages.test/hook"}, "events": ["push"]}] self.merged = False self.diff = "diff --git a/x b/x\n" self.log = "\n".join(f"line {i}" for i in range(1, 1001)) def __call__(self, request: httpx.Request) -> httpx.Response: m, p, q = request.method, request.url.path.removeprefix("/api/v1/"), dict(request.url.params) body = json.loads(request.content) if request.content else None self.calls.append((m, p, q, body)) self.raw_paths.append(request.url.raw_path) if request.headers.get("authorization") != "Bearer tok": return answer({"message": "token is required"}, 401) page, limit = int(q.get("page", 1)), int(q.get("limit", 50)) window = slice((page - 1) * limit, page * limit) if (m, p) == ("GET", "user"): return answer({"login": "tea", "is_admin": False}) if (m, p) == ("GET", "user/repos"): return answer([{"full_name": f"tea/r{i}"} for i in range(self.repo_count)][window], total=self.repo_count) if (m, p) in (("POST", "user/repos"), ("POST", "orgs/acme/repos")): owner = "tea" if p == "user/repos" else "acme" return answer({"full_name": f"{owner}/{body['name']}", "private": body["private"]}, 201) if (m, p) == ("POST", "orgs"): return answer({"username": body["username"]}, 201) if (m, p) == ("GET", "orgs/acme/teams"): return answer([{"id": 3, "name": "Owners"}, {"id": 4, "name": "Writers"}], total=2) if m == "PUT" and p.startswith("teams/"): return answer(status=204) if (m, p) == ("GET", R): return answer({"full_name": "o/r", "default_branch": "trunk", "private": True, "archived": False}) if (m, p) == ("PATCH", R): return answer({"full_name": "o/r", **body}) if (m, p) == ("GET", "repos/o/forbidden"): return answer({"message": "user should have a permission"}, 403) if (m, p) == ("GET", "repos/o/broken"): return answer({"message": "boom: secret internals"}, 500) if (m, p) == ("GET", f"{R}/labels"): return answer(self.labels, total=len(self.labels)) if (m, p) == ("GET", f"{R}/issues"): return answer([{"number": n, "title": f"t{n}", "state": "open"} for n in range(25, 0, -1)][window], total=25) if (m, p) == ("POST", f"{R}/issues"): return answer({"number": 26, "title": body["title"], "labels": [l for l in self.labels if l["id"] in body.get("labels", [])]}, 201) if (m, p) == ("GET", f"{R}/issues/1"): return answer({"number": 1, "title": "t1", "body": "x" * (server.MAX_BYTES + 5), "labels": self.on_issue}) if (m, p) == ("PATCH", f"{R}/issues/1"): return answer({"number": 1, "title": body.get("title", "t1"), "state": body.get("state", "open")}, 201) if (m, p) == ("GET", f"{R}/issues/1/comments"): return answer(self.comments, total=len(self.comments)) if (m, p) == ("POST", f"{R}/issues/1/comments"): return answer({"id": 99, "html_url": "https://gitea.test/o/r/issues/1#issuecomment-99"}, 201) if (m, p) == ("POST", f"{R}/issues/1/labels"): self.on_issue = self.on_issue + [l for l in self.labels if l["id"] in body["labels"]] return answer(self.on_issue) if m == "DELETE" and p.startswith(f"{R}/issues/1/labels/"): self.on_issue = [l for l in self.on_issue if str(l["id"]) != p.rsplit("/", 1)[1]] return answer(status=204) if (m, p) == ("GET", f"{R}/pulls/2"): return answer({"number": 2, "title": "p", "body": "why", "head": {"ref": "feature", "sha": "abcdef1234567890"}, "base": {"ref": "trunk"}, "mergeable": True, "merged": self.merged, "merge_commit_sha": "fedcba9876543210" if self.merged else None}) if (m, p) == ("GET", f"{R}/pulls/2.diff"): return httpx.Response(200, text=self.diff) if (m, p) == ("GET", f"{R}/commits/abcdef1234567890/status"): return answer({"state": "success", "statuses": [{"context": "ci / test", "status": "success", "description": "ok", "target_url": "https://ci.test/1"}]}) if (m, p) == ("POST", f"{R}/pulls"): return answer({"number": 3, "title": body["title"], "head": {"ref": body["head"]}}, 201) if (m, p) == ("POST", f"{R}/pulls/2/reviews"): return answer({"id": 11, "state": body["event"]}) if (m, p) == ("POST", f"{R}/pulls/2/merge"): self.merged = True return answer() if (m, p) == ("GET", f"{R}/contents"): return answer([{"name": n, "type": "file" if "." in n else "dir", "size": 1} for n in ("README.md", "img", "docs")]) if m == "GET" and p.startswith(f"{R}/contents/"): path = p.removeprefix(f"{R}/contents/") if path in ("img", "docs"): return answer([{"name": k.split("/")[1], "type": "file", "size": len(v)} for k, v in self.files.items() if k.startswith(path + "/")]) if path not in self.files: return answer({"message": "object does not exist"}, 404) data = self.files[path] return answer({"type": "file", "path": path, "sha": f"sha-{path}", "size": len(data), "content": b64(data)}) if (m, p) == ("POST", f"{R}/contents"): return answer({"commit": {"sha": "c0ffee" * 7, "html_url": "https://gitea.test/o/r/commit/c0ffee"}}, 201) if (m, p) == ("POST", f"{R}/branches"): return answer({"name": body["new_branch_name"], "commit": {"id": "b" * 40}}, 201) if (m, p) == ("GET", f"{R}/hooks"): return answer(self.hooks, total=len(self.hooks)) if (m, p) == ("POST", f"{R}/hooks"): return answer({"id": 8, "events": body["events"]}, 201) if (m, p) == ("GET", f"{R}/actions/runs"): runs = [{"id": 20 - i, "run_number": 15 - i, "display_title": "deploy", "path": "deploy.yml@main", "event": "push", "status": "completed", "conclusion": "success", "head_branch": None, "head_sha": "1234567890abcdef", "html_url": f"https://gitea.test/o/r/actions/runs/{20 - i}"} for i in range(15)] return answer({"total_count": len(runs), "workflow_runs": runs[window]}) if (m, p) == ("GET", f"{R}/actions/runs/5/jobs"): return answer({"total_count": 2, "jobs": [ {"id": 31, "name": "test", "status": "completed", "conclusion": "success", "steps": [{"name": "Set up job", "conclusion": "success"}]}, {"id": 32, "name": "lint", "status": "completed", "conclusion": "failure", "steps": []}]}) if (m, p) == ("GET", f"{R}/actions/jobs/31/logs"): return httpx.Response(200, text=self.log) if (m, p) == ("POST", f"{R}/actions/workflows/ci.yml/dispatches"): return answer(status=204) if m == "PUT" and p.startswith(f"{R}/actions/secrets/"): return answer(status=201) if p.startswith(f"{R}/actions/variables/"): name = p.rsplit("/", 1)[1] if m == "GET": return answer({"name": name, "data": self.variables[name]}) if name in self.variables else \ answer({"message": "not found"}, 404) self.variables[name] = body["value"] return answer(status=201 if m == "POST" else 204) return answer({"message": f"no route {m} {p}"}, 404) class TeaMCP(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.gitea = FakeGitea() server._transport = httpx.MockTransport(self.gitea) self.real_token, server.actor_token = server.actor_token, lambda: "tok" self.client = Client(server.mcp) await self.client.__aenter__() async def asyncTearDown(self): await self.client.__aexit__(None, None, None) server.actor_token, server._transport = self.real_token, None async def call(self, tool: str, **args) -> dict: return (await self.client.call_tool(tool, args)).structured_content async def refused(self, tool: str, **args) -> str: with self.assertRaises(ToolError) as caught: await self.client.call_tool(tool, args) return str(caught.exception) def sent(self, method: str, path: str | None = None) -> list[tuple]: return [c for c in self.gitea.calls if c[0] == method and (path is None or c[1] == path)] async def test_tools_and_their_hints(self): tools = {t.name: t for t in await self.client.list_tools()} self.assertEqual(set(tools), READS | WRITES) for name, tool in tools.items(): with self.subTest(tool=name): hints = tool.annotations self.assertEqual(bool(hints.read_only_hint), name in READS) self.assertIs(hints.destructive_hint, None if name in READS else False) self.assertEqual(bool(hints.idempotent_hint), name in AGAIN) self.assertEqual(tools["tea_issues"].input_schema["properties"]["limit"]["maximum"], 50) async def test_bad_arguments_never_reach_gitea(self): for tool, args in [("tea_issues", {"repo": "no-slash"}), ("tea_issues", {"repo": "o/r", "limit": 51}), ("tea_issues", {"repo": "o/r", "state": "merged"}), ("tea_repo", {"repo": "../x"}), ("tea_repo", {"repo": "o/.."}), ("tea_file_get", {"repo": "o/r", "path": "a\\b"}), ("tea_secret_set", {"repo": "o/r", "name": "1BAD", "value": "v"}), ("tea_commit", {"repo": "o/r", "message": "m", "files": []}), ("tea_pull_review", {"repo": "o/r", "number": 2, "event": "LGTM"}), ("tea_workflow_dispatch", {"repo": "o/r", "workflow": "../x.yml"}), ("tea_org_add_member", {"org": "acme", "user": "kim"})]: with self.subTest(tool=tool, args=args): await self.refused(tool, **args) self.assertEqual(self.gitea.calls, []) async def test_whoami_says_truthfully_whether_it_stopped_short(self): me = await self.call("tea_whoami") self.assertEqual((me["login"], len(me["repos"]), me["truncated"]), ("tea", 200, True)) self.gitea.repo_count = 200 me = await self.call("tea_whoami") self.assertEqual((len(me["repos"]), me["truncated"]), (200, False)) async def test_issues_pages(self): first = await self.call("tea_issues", repo="o/r", limit=20) last = await self.call("tea_issues", repo="o/r", page=2, limit=20) self.assertEqual((len(first["issues"]), first["more"], len(last["issues"]), last["more"]), (20, True, 5, False)) self.assertEqual(self.sent("GET", f"{R}/issues")[0][2], {"state": "open", "type": "issues", "page": "1", "limit": "20"}) await self.call("tea_issues", repo="o/r", kind="pulls", labels=["bug", "docs"]) self.assertEqual(self.sent("GET", f"{R}/issues")[-1][2]["labels"], "bug,docs") await self.call("tea_issues", repo="o/r", labels="bug, docs") self.assertEqual(self.sent("GET", f"{R}/issues")[-1][2]["labels"], "bug,docs") async def test_unknown_labels_are_refused(self): message = await self.refused("tea_issue_create", repo="o/r", title="x", labels=["bug", "nope"]) self.assertIn("unknown labels: nope", message) self.assertIn("bug, docs", message) self.assertEqual(self.sent("POST"), []) made = await self.call("tea_issue_create", repo="o/r", title=" x ", labels=["docs"]) self.assertEqual(self.sent("POST", f"{R}/issues")[0][3], {"title": "x", "body": "", "labels": [2]}) self.assertEqual((made["number"], made["labels"]), (26, ["docs"])) async def test_label_adds_and_removes(self): out = await self.call("tea_label", repo="o/r", number=1, add=["docs"], remove=["bug"]) self.assertEqual(self.sent("POST", f"{R}/issues/1/labels")[0][3], {"labels": [2]}) self.assertEqual([c[1] for c in self.sent("DELETE")], [f"{R}/issues/1/labels/1"]) self.assertEqual(out["labels"], ["docs"]) self.assertIn("nothing to change", await self.refused("tea_label", repo="o/r", number=1)) async def test_long_bodies_are_clipped_and_comments_paged(self): issue = await self.call("tea_issue_get", repo="o/r", number=1) self.assertEqual((len(issue["body"]), issue["body_truncated"]), (server.MAX_BYTES, True)) page2 = await self.call("tea_issue_comments", repo="o/r", number=1, page=2, limit=20) page3 = await self.call("tea_issue_comments", repo="o/r", number=1, page=3, limit=20) self.assertEqual([c["id"] for c in page2["comments"]], list(range(21, 41))) self.assertEqual((page2["more"], len(page3["comments"]), page3["more"]), (True, 5, False)) self.assertNotIn("body_truncated", page2["comments"][0]) async def test_issue_edit_and_comment(self): out = await self.call("tea_issue_edit", repo="o/r", number=1, state="closed", assignees=["tea"]) self.assertEqual((self.sent("PATCH")[0][3], out["state"]), ({"state": "closed", "assignees": ["tea"]}, "closed")) self.assertIn("nothing to change", await self.refused("tea_issue_edit", repo="o/r", number=1)) comment = await self.call("tea_issue_comment", repo="o/r", number=1, body=" hi ") self.assertEqual((self.sent("POST")[0][3], comment["comment_id"]), ({"body": "hi"}, 99)) self.assertIn("body is empty", await self.refused("tea_issue_comment", repo="o/r", number=1, body=" ")) async def test_pull_read_review_merge(self): self.gitea.diff = "+" * (server.MAX_BYTES + 10) pr = await self.call("tea_pull_get", repo="o/r", number=2) self.assertEqual((pr["head"], pr["base"], pr["head_sha"], pr["is_pull"]), ("feature", "trunk", "abcdef123456", True)) self.assertEqual((pr["checks"]["state"], pr["checks"]["statuses"][0]["context"]), ("success", "ci / test")) self.assertEqual((len(pr["diff"]), pr["diff_truncated"]), (server.MAX_BYTES, True)) self.assertNotIn("diff", await self.call("tea_pull_get", repo="o/r", number=2, diff=False)) review = await self.call("tea_pull_review", repo="o/r", number=2, event="APPROVED") self.assertEqual(review["state"], "APPROVED") merged = await self.call("tea_pull_merge", repo="o/r", number=2) self.assertEqual(self.sent("POST", f"{R}/pulls/2/merge")[0][3], {"Do": "squash", "delete_branch_after_merge": False}) self.assertEqual((merged["merged"], merged["merge_commit"]), (True, "fedcba987654")) async def test_pull_create_targets_the_default_branch(self): made = await self.call("tea_pull_create", repo="o/r", head="feature", title="p") self.assertEqual(self.sent("POST", f"{R}/pulls")[0][3], {"head": "feature", "base": "trunk", "title": "p", "body": ""}) self.assertEqual((made["number"], made["is_pull"]), (3, True)) async def test_file_reads(self): text = await self.call("tea_file_get", repo="o/r", path="/README.md") binary = await self.call("tea_file_get", repo="o/r", path="img/logo.png") spaced = await self.call("tea_file_get", repo="o/r", path="docs/a b.md") self.assertEqual((text["content"], text["sha"]), ("hello\n", "sha-README.md")) self.assertEqual((binary.get("binary"), "content" in binary), (True, False)) self.assertEqual(spaced["content"], "spaced\n") self.assertIn(b"/api/v1/repos/o/r/contents/docs/a%20b.md", self.gitea.raw_paths) self.gitea.files["big.txt"] = b"x" * (server.MAX_BYTES + 1) self.assertTrue((await self.call("tea_file_get", repo="o/r", path="big.txt"))["too_large"]) self.assertIn("not a file", await self.refused("tea_file_get", repo="o/r", path="docs")) self.assertIn("plain repository path", await self.refused("tea_file_get", repo="o/r", path="docs/../x")) async def test_tree(self): root = await self.call("tea_tree", repo="o/r") self.assertEqual(([e["name"] for e in root["entries"]], root["truncated"]), (["README.md", "img", "docs"], False)) self.assertEqual(self.gitea.calls[-1][1], f"{R}/contents") docs = await self.call("tea_tree", repo="o/r", path="docs/", ref="trunk") self.assertEqual(([e["name"] for e in docs["entries"]], docs["path"]), (["a b.md"], "docs")) self.assertEqual(self.gitea.calls[-1][2], {"ref": "trunk"}) async def test_commit_creates_updates_and_carries_binaries(self): png = b"\x89PNG\r\n\x1a\n\x00\x01" out = await self.call("tea_commit", repo="o/r", message=" add logo ", new_branch="art", files=[ {"path": "README.md", "content": "hello again\n"}, {"path": "img/new.png", "content_base64": b64(png)}]) payload = self.sent("POST", f"{R}/contents")[0][3] self.assertEqual((payload["message"], payload["branch"], payload["new_branch"]), ("add logo", "trunk", "art")) readme, image = payload["files"] self.assertEqual((readme["operation"], readme["sha"], base64.b64decode(readme["content"])), ("update", "sha-README.md", b"hello again\n")) self.assertEqual((image["operation"], base64.b64decode(image["content"]), "sha" in image), ("create", png, False)) self.assertEqual(out["files"], {"README.md": "update", "img/new.png": "create"}) self.assertEqual((out["branch"], out["commit"]), ("art", "c0ffeec0ffee")) self.assertTrue(all(c[2] == {"ref": "trunk"} for c in self.gitea.calls if c[1].startswith(f"{R}/contents/"))) async def test_commit_refusals(self): for files, expected in [([{"path": "a.txt"}], "exactly one"), ([{"path": "a.txt", "content": "x", "content_base64": b64(b"x")}], "exactly one"), ([{"path": "a.bin", "content_base64": "not base64!"}], "not valid base64"), ([{"path": "../a.txt", "content": "x"}], "plain repository path"), ([{"path": "a", "content": "x"}, {"path": "/a", "content": "y"}], "appears twice")]: with self.subTest(expected=expected): self.assertIn(expected, await self.refused("tea_commit", repo="o/r", message="m", files=files)) cap, server.MAX_COMMIT_BYTES = server.MAX_COMMIT_BYTES, 10 try: self.assertIn("more than 10 bytes", await self.refused("tea_commit", repo="o/r", message="m", files=[{"path": "a", "content": "x" * 11}])) finally: server.MAX_COMMIT_BYTES = cap self.assertEqual(self.gitea.calls, []) # all of that was refused before asking Gitea anything self.assertIn("is not a file", await self.refused("tea_commit", repo="o/r", message="m", files=[{"path": "docs", "content": "x"}])) self.assertEqual(self.sent("POST"), []) async def test_file_change_creates_or_updates_without_guessing(self): await self.call("tea_file_change", repo="o/r", path="new.md", content="n", message="m") await self.call("tea_file_change", repo="o/r", path="README.md", content="u", message="m", branch="dev") created, updated = [c[3] for c in self.sent("POST", f"{R}/contents")] self.assertEqual((created["files"][0]["operation"], created["branch"]), ("create", "trunk")) self.assertEqual((updated["files"][0]["operation"], updated["branch"]), ("update", "dev")) async def test_file_edit_replaces_exact_text_in_order_and_says_what_it_made(self): import hashlib self.gitea.files["notes.md"] = "one\ntwo — deux\nthree\n".encode("utf-8") out = await self.call("tea_file_edit", repo="o/r", path="/notes.md", message=" edit ", edits=[ {"old": "two — deux", "new": "two"}, {"old": "two\nthree", "new": "two\nthree\nfour"}, {"old": "one\n"}]) payload = self.sent("POST", f"{R}/contents")[0][3] (op,) = payload["files"] self.assertEqual((payload["message"], payload["branch"], "new_branch" in payload), ("edit", "trunk", False)) self.assertEqual((op["operation"], op["path"], op["sha"]), ("update", "notes.md", "sha-notes.md")) made = base64.b64decode(op["content"]) self.assertEqual(made, b"two\nthree\nfour\n") self.assertEqual((out["path"], out["edits"], out["size"], out["sha256"], out["branch"], out["commit"]), ("notes.md", 3, len(made), hashlib.sha256(made).hexdigest(), "trunk", "c0ffeec0ffee")) self.assertEqual(self.sent("GET", f"{R}/contents/notes.md")[0][2], {"ref": "trunk"}, "read from the branch it edits") await self.call("tea_file_edit", repo="o/r", path="README.md", message="m", branch="dev", edits=[{"old": "hello", "new": "hi"}]) self.assertEqual((self.sent("GET", f"{R}/contents/README.md")[-1][2], self.sent("POST", f"{R}/contents")[-1][3]["branch"]), ({"ref": "dev"}, "dev")) async def test_file_edit_refuses_what_it_cannot_do_exactly(self): self.gitea.files["notes.md"] = b"one two one\n" for edits, expected in [([{"old": "three"}], "edit 1: its old text occurs 0 times in notes.md, not once"), ([{"old": "one"}], "edit 1: its old text occurs 2 times in notes.md, not once"), ([{"old": "two", "new": "one"}, {"old": "one"}], "edit 2: its old text occurs 3 times")]: with self.subTest(expected=expected): self.assertIn(expected, await self.refused("tea_file_edit", repo="o/r", path="notes.md", message="m", edits=edits)) self.assertIn("is not a file", await self.refused("tea_file_edit", repo="o/r", path="docs", message="m", edits=[{"old": "x"}])) self.assertIn("is not UTF-8 text", await self.refused("tea_file_edit", repo="o/r", path="img/logo.png", message="m", edits=[{"old": "x"}])) self.assertEqual(self.sent("POST"), [], "nothing was committed") for args in [{"edits": []}, {"edits": [{"old": ""}]}, {"edits": [{"new": "x"}]}, {"edits": [{"old": "x"}], "path": "../x"}]: with self.subTest(args=args): await self.refused("tea_file_edit", **{"repo": "o/r", "path": "notes.md", "message": "m", **args}) cap, server.MAX_COMMIT_BYTES = server.MAX_COMMIT_BYTES, 10 try: self.assertIn("more than 10 bytes", await self.refused("tea_file_edit", repo="o/r", path="notes.md", message="m", edits=[{"old": "two", "new": "x" * 20}])) finally: server.MAX_COMMIT_BYTES = cap self.assertEqual(self.sent("POST"), []) async def test_branch_create(self): branch = await self.call("tea_branch_create", repo="o/r", name="feature/x") self.assertEqual(self.sent("POST", f"{R}/branches")[0][3], {"new_branch_name": "feature/x", "old_ref_name": "trunk"}) self.assertEqual(branch["commit"], "b" * 12) async def test_actions_lists_runs_newest_first(self): first = await self.call("tea_actions", repo="o/r") last = await self.call("tea_actions", repo="o/r", page=2) run = first["runs"][0] self.assertEqual((run["id"], run["number"], run["workflow"], run["branch"], run["sha"]), (20, 15, "deploy.yml", "main", "1234567890ab")) self.assertEqual((len(first["runs"]), first["more"], len(last["runs"]), last["more"]), (10, True, 5, False)) async def test_action_logs_tail_each_job(self): test, lint = (await self.call("tea_action_logs", repo="o/r", run=5, tail=50))["jobs"] self.assertEqual(test["log_tail"].splitlines(), [f"line {i}" for i in range(951, 1001)]) self.assertEqual((test["steps"][0]["name"], lint["conclusion"], lint["log_tail"]), ("Set up job", "failure", "")) async def test_a_long_log_keeps_its_end_and_whole_lines(self): self.gitea.log = "\n".join("x" * 100 for _ in range(2000)) + "\nTHE END" tail = (await self.call("tea_action_logs", repo="o/r", run=5, tail=400))["jobs"][0]["log_tail"] self.assertEqual(tail.splitlines()[-1], "THE END") self.assertTrue(all(line == "x" * 100 for line in tail.splitlines()[:-1])) self.assertLessEqual(len(tail), server.LOG_BYTES // 2) async def test_workflow_dispatch_defaults_to_the_default_branch(self): out = await self.call("tea_workflow_dispatch", repo="o/r", workflow="ci.yml", inputs={"why": "test"}) self.assertEqual((self.sent("POST")[0][3], out["dispatched"]), ({"ref": "trunk", "inputs": {"why": "test"}}, True)) async def test_a_secret_is_never_echoed(self): result = await self.client.call_tool("tea_secret_set", {"repo": "o/r", "name": "deploy_key", "value": "s3cr3t-v"}) shown = json.dumps(result.structured_content) + "".join(getattr(c, "text", "") for c in result.content) self.assertNotIn("s3cr3t-v", shown) self.assertEqual(self.sent("PUT")[0][1:], (f"{R}/actions/secrets/DEPLOY_KEY", {}, {"data": "s3cr3t-v"})) async def test_variable_set_creates_or_updates(self): fresh = await self.call("tea_variable_set", repo="o/r", name="fresh", value="1") existing = await self.call("tea_variable_set", repo="o/r", name="existing", value="2") self.assertEqual((fresh["created"], existing["created"]), (True, False)) self.assertEqual([(c[0], c[1].rsplit("/", 1)[1]) for c in self.gitea.calls if c[0] in ("POST", "PUT")], [("POST", "FRESH"), ("PUT", "EXISTING")]) async def test_hook_create_is_idempotent(self): same = await self.call("tea_hook_create", repo="o/r", url="https://pages.test/hook") self.assertEqual((same["created"], same["hook_id"], self.sent("POST")), (False, 7, [])) new = await self.call("tea_hook_create", repo="o/r", url="https://ci.test/hook", events=["push", "release"], secret="shh") posted = self.sent("POST", f"{R}/hooks")[0][3] self.assertEqual((posted["events"], posted["config"]), (["push", "release"], {"url": "https://ci.test/hook", "content_type": "json", "secret": "shh"})) self.assertNotIn("shh", json.dumps(new)) async def test_repositories_are_private_unless_asked(self): mine = await self.call("tea_repo_create", name="lab") org = await self.call("tea_org_repo_create", org="acme", name="site", private=False) self.assertEqual([c[3]["private"] for c in self.sent("POST")], [True, False]) self.assertEqual((mine["full_name"], org["full_name"]), ("tea/lab", "acme/site")) async def test_org_add_member_needs_a_real_team(self): self.assertIn("Owners, Writers", await self.refused("tea_org_add_member", org="acme", user="kim", team="Admins")) added = await self.call("tea_org_add_member", org="acme", user="kim", team="writers") self.assertEqual((self.sent("PUT")[0][1], added["team"]), ("teams/4/members/kim", "Writers")) async def test_repo_edit_can_archive(self): out = await self.call("tea_repo_edit", repo="o/r", archived=True, description="old") self.assertEqual(self.sent("PATCH", R)[0][3], {"description": "old", "archived": True}) self.assertEqual((out["archived"], out["description"]), (True, "old")) self.assertIn("nothing to change", await self.refused("tea_repo_edit", repo="o/r")) async def test_domains_are_looked_up_with_cloudflare(self): from tea_mcp import domains def cloudflare(request: httpx.Request) -> httpx.Response: if request.url.path.endswith("/domain-search"): return httpx.Response(200, json={"success": True, "result": {"domains": [ {"name": "teahub.uk", "registrable": True, "tier": "standard", "pricing": {"currency": "USD", "registration_cost": "5.30", "renewal_cost": "5.30"}}]}}) return httpx.Response(200, json={"success": True, "result": {"domains": [ {"name": name, "registrable": False, "reason": "domain_unavailable"} for name in json.loads(request.content)["domains"]]}}) domains.transport = httpx.MockTransport(cloudflare) self.addCleanup(setattr, domains, "transport", None) for kept in (domains._kept, domains._by_client, domains._by_all): kept.clear() self.assertIn("not set up", await self.refused("tea_domains", search="teahub")) with mock.patch.dict(os.environ, {"CLOUDFLARE_API_TOKEN": "t", "CLOUDFLARE_ACCOUNT_ID": "a"}): found = await self.call("tea_domains", search="teahub") self.assertEqual((len(found["exact_domains"]), found["domains"][0]["name"]), (10, "teahub.uk")) async def test_an_unreachable_gitea_is_named(self): def flaky(request): if request.url.path.endswith((".diff", "/contents")): raise httpx.ConnectError("refused", request=request) return self.gitea(request) server._transport = httpx.MockTransport(flaky) self.assertIn("Gitea did not answer (ConnectError)", await self.refused("tea_tree", repo="o/r")) self.assertIn("Gitea did not answer (ConnectError)", await self.refused("tea_pull_get", repo="o/r", number=2)) async def test_gitea_refusals_pass_through_but_its_internals_do_not(self): self.assertIn("Gitea refused (403): user should have a permission", await self.refused("tea_repo", repo="o/forbidden")) message = await self.refused("tea_repo", repo="o/broken") self.assertIn("Gitea is unavailable", message) self.assertNotIn("secret internals", message) class App(unittest.IsolatedAsyncioTestCase): async def test_health_and_sign_in_are_wired(self): async with httpx.AsyncClient(transport=httpx.ASGITransport(app=server.app), base_url="http://localhost") as web: health = await web.get("/healthz") anonymous = await web.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"}, headers={"Accept": "application/json, text/event-stream"}) metadata = await web.get("/.well-known/oauth-authorization-server") domains = await web.get("/domains", params={"q": "teahub"}, headers={"origin": "https://gitea.test"}) self.assertEqual((health.json()["version"], anonymous.status_code), (server.VERSION, 401)) self.assertEqual((domains.status_code, domains.headers["access-control-allow-origin"]), (503, "https://gitea.test")) self.assertEqual(metadata.json()["issuer"].rstrip("/"), "https://mcp.test") if __name__ == "__main__": unittest.main()