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()