diff --git a/tests/test_persimmons.py b/tests/test_persimmons.py new file mode 100644 index 0000000..0d1d58a --- /dev/null +++ b/tests/test_persimmons.py @@ -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()