From f0b9d468f2ca2381d3b9765f6244d09c1e00387d Mon Sep 17 00:00:00 2001 From: matcha Date: Sun, 4 Oct 2026 02:58:03 +0000 Subject: [PATCH] Consolidate Persimmons shared backend and tests --- tests/test_persimmons_juice.py | 102 +++++++++++++++++++++++++++++++++ 1 file changed, 102 insertions(+) create mode 100644 tests/test_persimmons_juice.py diff --git a/tests/test_persimmons_juice.py b/tests/test_persimmons_juice.py new file mode 100644 index 0000000..05f95cc --- /dev/null +++ b/tests/test_persimmons_juice.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import json +import unittest + +from tea_mcp.persimmons import create_connection, create_juice, list_juice, revoke_connection, revoke_juice +from tests.test_persimmons import FakeGitea + + +class JuiceTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.store = FakeGitea() + self.source = await create_connection(self.store.request, { + "provider": "Example", "credential_type": "Token", "credential": "source-secret-value", + "vault_kind": "user", + }) + self.data = {"source_id": self.source["id"], "subject": "agent:example", "scopes": ["read"]} + self.calls = [] + + async def request(self, method, path, **kwargs): + self.calls.append((method, path)) + # Juice must not read individual secrets or mutate any source secret. + self.assertNotIn("/secrets/", path) + return await self.store.request(method, path, **kwargs) + + async def test_lifecycle_metadata_only_and_source_survives_revoke(self): + created = await create_juice(self.request, self.data) + self.assertEqual(created["kind"], "juice") + self.assertEqual(created["status"], "active") + self.assertEqual(created["source"], "Example") + self.assertEqual(created["vault"], {"kind": "user"}) + self.assertIsNone(created["expires_at"]) + self.assertTrue(created["revocable"]) + self.assertIn("PERSIM_JUICE_" + created["id"], self.store.user_variables) + self.calls.clear() + listed = await list_juice(self.request, source_id=self.source["id"]) + self.assertEqual(listed, {"juice": [created]}) + self.assertTrue(all("/secrets" not in path for _, path in self.calls)) + self.assertEqual(await list_juice(self.request, source_id="AAAAAAAAAAAA"), {"juice": []}) + revoked = await revoke_juice(self.request, created["id"]) + self.assertEqual(revoked["status"], "revoked") + self.assertEqual(await revoke_juice(self.request, created["id"]), revoked) + self.assertEqual(await list_juice(self.request), {"juice": [revoked]}) + self.assertNotIn("source-secret-value", json.dumps([created, listed, revoked, self.store.user_variables])) + self.assertEqual(self.store.user_secrets, {"EXAMPLE": "source-secret-value"}) + + async def test_missing_revoked_and_unavailable_sources_rejected(self): + with self.assertRaisesRegex(ValueError, "not found"): + await create_juice(self.request, {**self.data, "source_id": "AAAAAAAAAAAA"}) + self.store.user_secrets.clear() + with self.assertRaisesRegex(ValueError, "unavailable"): + await create_juice(self.request, self.data) + await revoke_connection(self.store.request, self.source["id"], {"vault_kind": "user"}) + self.store.user_secrets["EXAMPLE"] = "source-secret-value" + with self.assertRaisesRegex(ValueError, "revoked"): + await create_juice(self.request, self.data) + + async def test_repository_isolation_and_mismatched_metadata(self): + repo = {"repository": "matcha/example"} + with self.assertRaisesRegex(ValueError, "not found"): + await create_juice(self.request, {**self.data, **repo}) + key = "PERSIM_META_" + self.source["id"] + self.store.repo_variables[repo["repository"]][key] = self.store.user_variables[key] + with self.assertRaisesRegex(ValueError, "vault does not match"): + await create_juice(self.request, {**self.data, **repo}) + source = await create_connection(self.store.request, { + "provider": "Example", "credential_type": "Token", "credential": "repo-secret", + "vault_kind": "repository", **repo, + }) + juice = await create_juice(self.request, {**self.data, **repo, "source_id": source["id"]}) + self.assertEqual(await list_juice(self.request), {"juice": []}) + self.assertEqual(await list_juice(self.request, **repo), {"juice": [juice]}) + with self.assertRaisesRegex(ValueError, "not found"): + await revoke_juice(self.request, juice["id"]) + copied = "PERSIM_JUICE_" + juice["id"] + self.store.user_variables[copied] = self.store.repo_variables[repo["repository"]][copied] + with self.assertRaisesRegex(ValueError, "vault does not match"): + await revoke_juice(self.request, juice["id"]) + self.assertEqual(await list_juice(self.request), {"juice": []}) + await revoke_juice(self.request, juice["id"], repo) + self.assertEqual(self.store.repo_secrets[repo["repository"]]["EXAMPLE"], "repo-secret") + + async def test_validation_rejects_unbounded_or_ambiguous_policy(self): + for override in ( + {"subject": ""}, {"subject": "x" * 201}, {"scopes": []}, {"scopes": "read"}, + {"scopes": ["x"] * 41}, {"scopes": ["x" * 101]}, {"scopes": [{}]}, + {"expires_at": "tomorrow"}, {"expires_at": "2000-01-01T00:00:00Z"}, + {"expires_at": "2099-01-01T00:00:00"}, {"source_id": "../secret"}, + {"vault_kind": "user", "repository": "matcha/example"}, + ): + with self.subTest(override=override), self.assertRaises(ValueError): + await create_juice(self.request, {**self.data, **override}) + created = await create_juice(self.request, {**self.data, "expires_at": "2099-01-01T00:00:00Z"}) + self.assertEqual(created["expires_at"], "2099-01-01T00:00:00Z") + + async def test_listing_projects_only_juice_fields(self): + created = await create_juice(self.request, self.data) + key = "PERSIM_JUICE_" + created["id"] + metadata = json.loads(self.store.user_variables[key]) + metadata["credential"] = "unexpected-private-field" + self.store.user_variables[key] = json.dumps(metadata) + self.assertNotIn("unexpected-private-field", json.dumps(await list_juice(self.request)))