Consolidate Persimmons shared backend and tests
This commit is contained in:
matcha committed
1 parent
4399e119f5
commit
3bacfbfc0b
1 file changed
+271
@@ -0,0 +1,271 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from tea_mcp.persimmons import (
|
||||
create_connection,
|
||||
delete_connection,
|
||||
list_connections,
|
||||
list_repositories,
|
||||
revoke_connection,
|
||||
rotate_connection,
|
||||
)
|
||||
|
||||
|
||||
class FakeGitea:
|
||||
def __init__(self) -> None:
|
||||
self.user_secrets: dict[str, str] = {}
|
||||
self.user_variables: dict[str, str] = {}
|
||||
self.repo_secrets: dict[str, dict[str, str]] = {}
|
||||
self.repo_variables: dict[str, dict[str, str]] = {}
|
||||
self.repos = {
|
||||
"matcha/example": {
|
||||
"full_name": "matcha/example",
|
||||
"private": True,
|
||||
"owner": {"login": "matcha"},
|
||||
"permissions": {"admin": True, "push": True},
|
||||
},
|
||||
"matcha/read-only": {
|
||||
"full_name": "matcha/read-only",
|
||||
"private": True,
|
||||
"owner": {"login": "matcha"},
|
||||
"permissions": {"admin": False, "push": False},
|
||||
},
|
||||
}
|
||||
|
||||
async def request(self, method: str, path: str, *, params=None, json=None):
|
||||
if path == "user/repos" and method == "GET":
|
||||
return list(self.repos.values())
|
||||
if path.startswith("repos/") and "/actions/" not in path and method == "GET":
|
||||
repo = path[len("repos/"):]
|
||||
if repo not in self.repos and repo != "matcha/persimmons":
|
||||
raise RuntimeError("not found (HTTP 404)")
|
||||
return self.repos.get(repo) or {
|
||||
"full_name": repo,
|
||||
"private": True,
|
||||
"permissions": {"admin": True, "push": True},
|
||||
}
|
||||
|
||||
if path.startswith("user/actions/"):
|
||||
return self._actions(method, path[len("user/actions/"):], self.user_secrets, self.user_variables, json)
|
||||
|
||||
if path.startswith("repos/") and "/actions/" in path:
|
||||
repo, suffix = path[len("repos/"):].split("/actions/", 1)
|
||||
secrets = self.repo_secrets.setdefault(repo, {})
|
||||
variables = self.repo_variables.setdefault(repo, {})
|
||||
return self._actions(method, suffix, secrets, variables, json)
|
||||
|
||||
raise AssertionError(f"unexpected request: {method} {path}")
|
||||
|
||||
def _actions(self, method, suffix, secrets, variables, body):
|
||||
if suffix == "secrets" and method == "GET":
|
||||
return [{"name": name, "created_at": "2026-09-27T00:00:00Z"} for name in secrets]
|
||||
if suffix.startswith("secrets/"):
|
||||
name = suffix.split("/", 1)[1]
|
||||
if method == "PUT":
|
||||
secrets[name] = str((body or {}).get("data") or "")
|
||||
return None
|
||||
if method == "DELETE":
|
||||
if name not in secrets:
|
||||
raise RuntimeError("not found (HTTP 404)")
|
||||
secrets.pop(name)
|
||||
return None
|
||||
|
||||
if suffix == "variables" and method == "GET":
|
||||
return [{"name": name, "data": value} for name, value in variables.items()]
|
||||
if suffix.startswith("variables/"):
|
||||
name = suffix.split("/", 1)[1]
|
||||
if method == "GET":
|
||||
if name not in variables:
|
||||
raise RuntimeError("not found (HTTP 404)")
|
||||
return {"name": name, "data": variables[name]}
|
||||
if method in {"POST", "PUT"}:
|
||||
variables[name] = str((body or {}).get("value") or "")
|
||||
return None
|
||||
if method == "DELETE":
|
||||
if name not in variables:
|
||||
raise RuntimeError("not found (HTTP 404)")
|
||||
variables.pop(name)
|
||||
return None
|
||||
|
||||
raise AssertionError(f"unexpected actions request: {method} {suffix}")
|
||||
|
||||
|
||||
class PersimmonsVaultTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_user_vault_lifecycle_never_reads_written_value(self) -> None:
|
||||
store = FakeGitea()
|
||||
created = await create_connection(
|
||||
store.request,
|
||||
{
|
||||
"provider": "Gitea",
|
||||
"account": "example-user",
|
||||
"credential_type": "PAT",
|
||||
"scopes": ["repo/read", "issues/read"],
|
||||
"credential": "alpha-value",
|
||||
"vault_kind": "user",
|
||||
},
|
||||
)
|
||||
cid = created["id"]
|
||||
self.assertEqual(created["vault_kind"], "user")
|
||||
self.assertTrue(created["available"])
|
||||
|
||||
listed = await list_connections(store.request, include_legacy=False)
|
||||
self.assertEqual(len(listed["connections"]), 1)
|
||||
self.assertEqual(listed["connections"][0]["id"], cid)
|
||||
self.assertNotIn("alpha-value", json.dumps(listed))
|
||||
|
||||
rotated = await rotate_connection(
|
||||
store.request, cid, "beta-value", {"vault_kind": "user"}
|
||||
)
|
||||
self.assertTrue(rotated["available"])
|
||||
self.assertEqual(next(iter(store.user_secrets.values())), "beta-value")
|
||||
self.assertNotIn("beta-value", json.dumps(await list_connections(store.request, include_legacy=False)))
|
||||
|
||||
revoked = await revoke_connection(store.request, cid, {"vault_kind": "user"})
|
||||
self.assertFalse(revoked["available"])
|
||||
self.assertEqual(store.user_secrets, {})
|
||||
|
||||
deleted = await delete_connection(store.request, cid, {"vault_kind": "user"})
|
||||
self.assertTrue(deleted["deleted"])
|
||||
self.assertEqual(await list_connections(store.request, include_legacy=False), {"connections": []})
|
||||
|
||||
async def test_repository_vault_matches_repository_scope(self) -> None:
|
||||
store = FakeGitea()
|
||||
created = await create_connection(
|
||||
store.request,
|
||||
{
|
||||
"provider": "Cloudflare",
|
||||
"credential_type": "API token",
|
||||
"credential": "repo-value",
|
||||
"vault_kind": "repository",
|
||||
"repository": "matcha/example",
|
||||
},
|
||||
)
|
||||
cid = created["id"]
|
||||
self.assertEqual(created["repository"], "matcha/example")
|
||||
|
||||
user_only = await list_connections(store.request, include_legacy=False)
|
||||
self.assertEqual(user_only, {"connections": []})
|
||||
|
||||
repo_only = await list_connections(
|
||||
store.request,
|
||||
repository="matcha/example",
|
||||
include_user=False,
|
||||
include_legacy=False,
|
||||
)
|
||||
self.assertEqual(repo_only["connections"][0]["id"], cid)
|
||||
self.assertTrue(repo_only["connections"][0]["available"])
|
||||
self.assertNotIn("repo-value", json.dumps(repo_only))
|
||||
|
||||
await revoke_connection(
|
||||
store.request,
|
||||
cid,
|
||||
{"vault_kind": "repository", "repository": "matcha/example"},
|
||||
)
|
||||
repo_after = await list_connections(
|
||||
store.request,
|
||||
repository="matcha/example",
|
||||
include_user=False,
|
||||
include_legacy=False,
|
||||
)
|
||||
self.assertFalse(repo_after["connections"][0]["available"])
|
||||
|
||||
async def test_existing_user_secret_projects_without_fabricated_metadata(self) -> None:
|
||||
store = FakeGitea()
|
||||
store.user_secrets["CLOUDFLARE_TOKEN"] = "never-return-this"
|
||||
listed = await list_connections(store.request, include_legacy=False)
|
||||
self.assertEqual(len(listed["connections"]), 1)
|
||||
item = listed["connections"][0]
|
||||
self.assertEqual(item["name"], "CLOUDFLARE_TOKEN")
|
||||
self.assertEqual(item["provider"], "")
|
||||
self.assertEqual(item["account"], "")
|
||||
self.assertEqual(item["scopes"], [])
|
||||
self.assertEqual(item["vault_kind"], "user")
|
||||
self.assertTrue(item["legacy"])
|
||||
self.assertTrue(item["available"])
|
||||
self.assertNotIn("never-return-this", json.dumps(listed))
|
||||
|
||||
async def test_existing_repository_secret_projects_in_repository_scope(self) -> None:
|
||||
store = FakeGitea()
|
||||
store.repo_secrets.setdefault("matcha/example", {})["DEPLOY_TOKEN"] = "repo-hidden"
|
||||
listed = await list_connections(
|
||||
store.request,
|
||||
repository="matcha/example",
|
||||
include_user=False,
|
||||
include_legacy=False,
|
||||
)
|
||||
item = listed["connections"][0]
|
||||
self.assertEqual(item["name"], "DEPLOY_TOKEN")
|
||||
self.assertEqual(item["repository"], "matcha/example")
|
||||
self.assertEqual(item["credential_type"], "Existing Gitea secret")
|
||||
self.assertNotIn("repo-hidden", json.dumps(listed))
|
||||
|
||||
async def test_existing_secret_can_rotate_and_revoke_without_readback(self) -> None:
|
||||
store = FakeGitea()
|
||||
store.repo_secrets.setdefault("matcha/example", {})["DEPLOY_TOKEN"] = "old-value"
|
||||
rotated = await rotate_connection(
|
||||
store.request,
|
||||
"secret:DEPLOY_TOKEN",
|
||||
"new-value",
|
||||
{"vault_kind": "repository", "repository": "matcha/example"},
|
||||
)
|
||||
self.assertTrue(rotated["available"])
|
||||
self.assertEqual(store.repo_secrets["matcha/example"]["DEPLOY_TOKEN"], "new-value")
|
||||
self.assertNotIn("new-value", json.dumps(rotated))
|
||||
|
||||
revoked = await revoke_connection(
|
||||
store.request,
|
||||
"secret:DEPLOY_TOKEN",
|
||||
{"vault_kind": "repository", "repository": "matcha/example"},
|
||||
)
|
||||
self.assertFalse(revoked["available"])
|
||||
self.assertNotIn("DEPLOY_TOKEN", store.repo_secrets["matcha/example"])
|
||||
|
||||
async def test_new_connection_name_matches_gitea_secret_name(self) -> None:
|
||||
store = FakeGitea()
|
||||
created = await create_connection(
|
||||
store.request,
|
||||
{
|
||||
"name": "my useful password",
|
||||
"provider": "Other",
|
||||
"credential_type": "Password",
|
||||
"credential": "hidden-value",
|
||||
"vault_kind": "user",
|
||||
},
|
||||
)
|
||||
self.assertEqual(created["name"], "MY_USEFUL_PASSWORD")
|
||||
self.assertEqual(created["secret_name"], "MY_USEFUL_PASSWORD")
|
||||
self.assertIn("MY_USEFUL_PASSWORD", store.user_secrets)
|
||||
listed = await list_connections(store.request, include_legacy=False)
|
||||
self.assertEqual(listed["connections"][0]["name"], "MY_USEFUL_PASSWORD")
|
||||
self.assertEqual(listed["connections"][0]["secret_name"], "MY_USEFUL_PASSWORD")
|
||||
self.assertNotIn("hidden-value", json.dumps(listed))
|
||||
|
||||
async def test_duplicate_secret_name_does_not_overwrite(self) -> None:
|
||||
store = FakeGitea()
|
||||
store.user_secrets["EXISTING_NAME"] = "original"
|
||||
with self.assertRaisesRegex(ValueError, "already exists"):
|
||||
await create_connection(
|
||||
store.request,
|
||||
{
|
||||
"name": "existing name",
|
||||
"provider": "Other",
|
||||
"credential_type": "Password",
|
||||
"credential": "replacement",
|
||||
"vault_kind": "user",
|
||||
},
|
||||
)
|
||||
self.assertEqual(store.user_secrets["EXISTING_NAME"], "original")
|
||||
|
||||
async def test_repository_picker_only_returns_writable_repositories(self) -> None:
|
||||
store = FakeGitea()
|
||||
result = await list_repositories(store.request)
|
||||
self.assertEqual(
|
||||
[item["full_name"] for item in result["repositories"]],
|
||||
["matcha/example"],
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user