MCP: tests for tea_file_edit (#12)

Replacements apply in order and the answer says what was made; every refusal comes before anything is committed.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SKUMBXauq7XVN4BhTXjCSM

mcp/README.md

@@ -4,11 +4,11 @@ Teahub's MCP (Fly app `green-tea-mcp`, `https://mcp.teahub.xyz/mcp`) against the
 
 Env: `GITEA_URL`, `PUBLIC_URL`, `ALLOWED_HOSTS`; secrets `GITEA_OAUTH_CLIENT_ID`, `GITEA_OAUTH_CLIENT_SECRET` (an OAuth2 application on the Gitea with redirect `PUBLIC_URL/auth/callback`), `JWT_SIGNING_KEY`. State: `/data` (FastMCP's OAuth store).
 
-Tools (31), all acting as you. Their parameters are typed, so a bad argument is refused before Gitea is asked anything.
+Tools (32), all acting as you. Their parameters are typed, so a bad argument is refused before Gitea is asked anything.
 
 - Issues and pull requests: `tea_issues`, `tea_issue_get/create/edit`, `tea_issue_comments/comment`, `tea_labels`, `tea_label` (add or remove by name), `tea_pull_get` (branches, mergeability, checks, diff), `tea_pull_create`, `tea_pull_review`, `tea_pull_merge` (keeps the head branch).
 - Repositories and organisations: `tea_whoami`, `tea_repo`, `tea_repo_edit` (description, website, visibility, default branch, archived), `tea_repo_create`, `tea_org_create`, `tea_org_repo_create`, `tea_org_add_member` (name the team), `tea_hook_create` (one hook per URL).
-- Files: `tea_tree`, `tea_file_get`, `tea_file_change`, `tea_commit` (up to 50 files, text or base64, 3 MB in all), `tea_branch_create`.
+- Files: `tea_tree`, `tea_file_get`, `tea_file_change`, `tea_file_edit` (exact-text replacements in one file, for a file too long to send whole; it answers with the new file's size and sha256), `tea_commit` (up to 50 files, text or base64, 3 MB in all), `tea_branch_create`.
 - Domains: `tea_domains` (Cloudflare Registrar: exact names checked live, and suggestions, with prices).
 - Actions: `tea_actions` (recent runs), `tea_action_logs` (a run's jobs, steps and log tails), `tea_workflow_dispatch`, `tea_secret_set` (never echoed), `tea_variable_set`.
 

mcp/tea_mcp/server.py

@@ -46,7 +46,7 @@ GITEA_OAUTH_CLIENT_SECRET = required_env("GITEA_OAUTH_CLIENT_SECRET")
 JWT_SIGNING_KEY = required_env("JWT_SIGNING_KEY")
 ALLOWED_HOSTS = [h.strip() for h in os.getenv("ALLOWED_HOSTS", "mcp.teahub.xyz,localhost").split(",") if h.strip()]
 
-VERSION = "2.2.0"
+VERSION = "2.3.0"
 OAUTH_SCOPES = ["openid", "profile", "email"]
 CLAUDE_CIMD_CLIENT_ID = "https://claude.ai/oauth/mcp-oauth-client-metadata"
 CLAUDE_REDIRECT_URI = "https://claude.ai/api/mcp/auth_callback"
@@ -143,6 +143,13 @@ class FileChange(BaseModel):
     content_base64: str | None = None
 
 
+class Edit(BaseModel):
+    """One replacement in a text file: `old` occurs exactly once in the file as it stands (after the edits before it),
+    and `new` takes its place (empty removes it)."""
+    old: Annotated[str, Field(min_length=1, max_length=MAX_BYTES)]
+    new: Annotated[str, Field(max_length=MAX_BYTES)] = ""
+
+
 class InputError(ToolError):
     pass
 
@@ -595,6 +602,37 @@ async def commit(repo: str, message: str, files: list[FileChange], branch: str,
             "files": {o["path"]: o["operation"] for o in ops}}
 
 
+@mcp.tool(description="Edit one text file on a branch by exact-text replacements, as a commit by you: for a file too "
+                      "long to send whole (tea_file_change takes 200 KB). The edits apply in order; each `old` must "
+                      "occur exactly once in the file as it stands, and `new` takes its place. The answer gives the new "
+                      "file's size and sha256, to check against your copy. Branch protection applies.",
+          annotations=WRITE)
+async def tea_file_edit(repo: Repo, path: FilePath, edits: Annotated[list[Edit], Field(min_length=1, max_length=50)],
+                        message: Message, branch: Ref = "") -> dict[str, Any]:
+    rp, p = repo_path(repo), file_path(path)
+    branch = branch or await default_branch(rp)
+    x = await gitea("GET", contents(rp, p), params={"ref": branch})
+    if not isinstance(x, dict) or x.get("type") != "file":
+        raise InputError(f"{p} is not a file")
+    try:
+        text = base64.b64decode(x.get("content") or "").decode("utf-8")
+    except (binascii.Error, ValueError):
+        raise InputError(f"{p} is not UTF-8 text") from None
+    for n, e in enumerate(edits, 1):
+        found = text.count(e.old)
+        if found != 1:
+            raise InputError(f"edit {n}: its old text occurs {found} times in {p}, not once")
+        text = text.replace(e.old, e.new, 1)
+    data = text.encode("utf-8")
+    if len(data) > MAX_COMMIT_BYTES:
+        raise InputError(f"{p} would come to more than {MAX_COMMIT_BYTES} bytes")
+    payload = {"files": [{"operation": "update", "path": p, "sha": x.get("sha"),
+                          "content": base64.b64encode(data).decode()}], "message": message.strip(), "branch": branch}
+    c = (await gitea("POST", f"{rp}/contents", json=payload)).get("commit") or {}
+    return {"repo": repo, "branch": branch, "path": p, "edits": len(edits), "size": len(data),
+            "sha256": hashlib.sha256(data).hexdigest(), "commit": (c.get("sha") or "")[:12], "url": c.get("html_url")}
+
+
 @mcp.tool(description="Create a branch from another branch, a tag or a commit (default: the default branch).",
           annotations=WRITE)
 async def tea_branch_create(repo: Repo, name: Branch, from_ref: Ref = "") -> dict[str, Any]:

mcp/tests/test_server.py

@@ -25,8 +25,8 @@ READS = {"tea_whoami", "tea_issues", "tea_issue_get", "tea_issue_comments", "tea
          "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_commit", "tea_branch_create",
-          "tea_workflow_dispatch", "tea_secret_set", "tea_variable_set"}
+          "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"}
 
@@ -346,6 +346,46 @@ class TeaMCP(unittest.IsolatedAsyncioTestCase):
         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"})