Consolidate Persimmons shared backend and tests
This commit is contained in:
matcha committed
1 parent
92151538e7
commit
53c4bb50f4
1 file changed
+296
@@ -0,0 +1,296 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unittest
|
||||
|
||||
from tea_mcp.persimmons import list_connections
|
||||
from tea_mcp.persimmons_web import PersimmonsWebRequester, _repo_results, parse_secret_page, parse_variable_page
|
||||
|
||||
|
||||
class PersimmonsWebSessionParserTests(unittest.TestCase):
|
||||
def test_gitea_127_secret_markup_is_safe_and_identifiable(self) -> None:
|
||||
page = """
|
||||
<div class="item">
|
||||
<button data-modal-secret-name.value="CLOUDFLARE_TOKEN"></button>
|
||||
<button data-url="/user/settings/actions/secrets/delete?id=41"></button>
|
||||
</div>
|
||||
<div class="item">
|
||||
<button data-modal-secret-name.value="DEPLOY_PAT"></button>
|
||||
<button data-url="/user/settings/actions/secrets/delete?id=42"></button>
|
||||
</div>
|
||||
"""
|
||||
items = parse_secret_page(page)
|
||||
self.assertEqual(
|
||||
[(item["name"], item["id"]) for item in items],
|
||||
[("CLOUDFLARE_TOKEN", 41), ("DEPLOY_PAT", 42)],
|
||||
)
|
||||
self.assertNotIn("data", items[0])
|
||||
|
||||
def test_gitea_127_variable_markup_yields_non_secret_metadata(self) -> None:
|
||||
page = """
|
||||
<button
|
||||
data-modal-form.url="/user/settings/actions/variables/9/edit"
|
||||
data-modal-dialog-variable-name="PERSIM_META_ABCDEF012345"
|
||||
data-modal-dialog-variable-data="{"provider":"Gitea"}"
|
||||
data-modal-dialog-variable-description="Persimmons metadata"
|
||||
></button>
|
||||
"""
|
||||
items = parse_variable_page(page)
|
||||
self.assertEqual(len(items), 1)
|
||||
self.assertEqual(items[0]["id"], 9)
|
||||
self.assertEqual(items[0]["name"], "PERSIM_META_ABCDEF012345")
|
||||
self.assertEqual(items[0]["data"], '{"provider":"Gitea"}')
|
||||
self.assertEqual(items[0]["description"], "Persimmons metadata")
|
||||
|
||||
|
||||
def test_gitea_127_web_repo_search_unwraps_nested_repository(self) -> None:
|
||||
payload = {
|
||||
"ok": True,
|
||||
"data": [
|
||||
{
|
||||
"repository": {
|
||||
"full_name": "acme/example",
|
||||
"private": True,
|
||||
},
|
||||
"latest_commit_status": None,
|
||||
}
|
||||
],
|
||||
}
|
||||
self.assertEqual(
|
||||
_repo_results(payload),
|
||||
[{"full_name": "acme/example", "private": True}],
|
||||
)
|
||||
|
||||
def test_gitea_web_repo_search_rejects_unknown_shape(self) -> None:
|
||||
with self.assertRaises(ValueError):
|
||||
_repo_results({"ok": True})
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
def __init__(self, status_code: int, payload=None) -> None:
|
||||
self.status_code = status_code
|
||||
self._payload = payload
|
||||
|
||||
def json(self):
|
||||
return self._payload
|
||||
|
||||
|
||||
class _RepoDiscoveryRequester(PersimmonsWebRequester):
|
||||
async def _get(self, path: str, *, params=None, accept_json: bool = False):
|
||||
if path == "/repo/search":
|
||||
return _FakeResponse(
|
||||
200,
|
||||
{
|
||||
"ok": True,
|
||||
"data": [
|
||||
{"repository": {"full_name": "matcha/owned", "private": True}},
|
||||
{"repository": {"full_name": "shared/collab", "private": True}},
|
||||
{"repository": {"full_name": "other/hidden", "private": True}},
|
||||
],
|
||||
},
|
||||
)
|
||||
if path in {
|
||||
"/matcha/owned/settings/actions/secrets",
|
||||
"/shared/collab/settings/actions/secrets",
|
||||
}:
|
||||
return _FakeResponse(200)
|
||||
if path == "/other/hidden/settings/actions/secrets":
|
||||
return _FakeResponse(404)
|
||||
raise AssertionError(path)
|
||||
|
||||
|
||||
class PersimmonsRepositoryDiscoveryTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_discovery_keeps_writable_collaborator_repo_and_filters_inaccessible(self) -> None:
|
||||
requester = _RepoDiscoveryRequester("https://example.invalid", "cookie", "actor")
|
||||
repos = await requester._repositories({})
|
||||
self.assertEqual(
|
||||
[repo["full_name"] for repo in repos],
|
||||
["matcha/owned", "shared/collab"],
|
||||
)
|
||||
|
||||
|
||||
class _ActorScopedRequester(PersimmonsWebRequester):
|
||||
ALL_REPOS = [
|
||||
"teahub-admin/bootstrap-persistence-check",
|
||||
"matcha/greentea",
|
||||
"matcha/persimmons",
|
||||
"chai/pointtest",
|
||||
"matcha/teahub",
|
||||
]
|
||||
|
||||
def __init__(self, actor_name: str, allowed_repos: set[str], user_secret: str) -> None:
|
||||
super().__init__("https://example.invalid", f"cookie-{actor_name}", actor_name)
|
||||
self.allowed_repos = allowed_repos
|
||||
self.user_secret = user_secret
|
||||
self.posts: list[tuple[str, dict[str, str]]] = []
|
||||
|
||||
async def _get(self, path: str, *, params=None, accept_json: bool = False):
|
||||
if path == "/repo/search":
|
||||
return _FakeResponse(
|
||||
200,
|
||||
{
|
||||
"ok": True,
|
||||
"data": [
|
||||
{"repository": {"full_name": name, "private": True}}
|
||||
for name in self.ALL_REPOS
|
||||
],
|
||||
},
|
||||
)
|
||||
if path == "/user/settings/actions/secrets":
|
||||
return _HtmlResponse(
|
||||
200,
|
||||
f'<button data-modal-secret-name.value="{self.user_secret}"></button>'
|
||||
'<button data-url="/user/settings/actions/secrets/delete?id=1"></button>',
|
||||
)
|
||||
if path == "/user/settings/actions/variables":
|
||||
return _HtmlResponse(200, "")
|
||||
match = re.fullmatch(r"/([^/]+/[^/]+)/settings/actions/(secrets|variables)", path)
|
||||
if match:
|
||||
repository, kind = match.groups()
|
||||
if repository not in self.allowed_repos:
|
||||
return _HtmlResponse(404, "")
|
||||
if kind == "variables":
|
||||
return _HtmlResponse(200, "")
|
||||
secret = "POINTTEST_TOKEN" if repository == "chai/pointtest" else "REPO_TOKEN"
|
||||
return _HtmlResponse(
|
||||
200,
|
||||
f'<button data-modal-secret-name.value="{secret}"></button>'
|
||||
f'<button data-url="/{repository}/settings/actions/secrets/delete?id=2"></button>',
|
||||
)
|
||||
raise AssertionError(path)
|
||||
|
||||
async def _post(self, path: str, data: dict[str, str], *, params=None):
|
||||
self.posts.append((path, data))
|
||||
return _HtmlResponse(302, "")
|
||||
|
||||
|
||||
class _MutationRequester(PersimmonsWebRequester):
|
||||
def __init__(self) -> None:
|
||||
super().__init__("https://example.invalid", "session-cookie", "actor")
|
||||
self.posts: list[tuple[str, dict[str, str], object]] = []
|
||||
|
||||
async def _get(self, path: str, *, params=None, accept_json: bool = False):
|
||||
if path.endswith("/secrets"):
|
||||
return _HtmlResponse(
|
||||
200,
|
||||
'<button data-modal-secret-name.value="TEST_FIXTURE"></button>'
|
||||
'<button data-url="/user/settings/actions/secrets/delete?id=7"></button>',
|
||||
)
|
||||
if path.endswith("/variables"):
|
||||
return _HtmlResponse(200, "")
|
||||
raise AssertionError(path)
|
||||
|
||||
async def _post(self, path: str, data: dict[str, str], *, params=None):
|
||||
self.posts.append((path, data, params))
|
||||
return _HtmlResponse(302, "")
|
||||
|
||||
|
||||
class PersimmonsMutationContractTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_secret_write_does_not_require_legacy_csrf_field(self) -> None:
|
||||
requester = _MutationRequester()
|
||||
await requester._secret_put(
|
||||
"/user/settings/actions",
|
||||
"TEST_FIXTURE",
|
||||
"fixture-value",
|
||||
"fixture",
|
||||
)
|
||||
path, data, params = requester.posts[-1]
|
||||
self.assertEqual(path, "/user/settings/actions/secrets")
|
||||
self.assertEqual(data["name"], "TEST_FIXTURE")
|
||||
self.assertEqual(data["data"], "fixture-value")
|
||||
self.assertNotIn("_csrf", data)
|
||||
self.assertIsNone(params)
|
||||
|
||||
async def test_secret_delete_does_not_require_legacy_csrf_field(self) -> None:
|
||||
requester = _MutationRequester()
|
||||
await requester._secret_delete("/user/settings/actions", "TEST_FIXTURE")
|
||||
path, data, params = requester.posts[-1]
|
||||
self.assertEqual(path, "/user/settings/actions/secrets/delete")
|
||||
self.assertEqual(data, {"id": "7"})
|
||||
self.assertEqual(params, {"id": 7})
|
||||
|
||||
|
||||
class _HtmlResponse(_FakeResponse):
|
||||
def __init__(self, status_code: int, text: str) -> None:
|
||||
super().__init__(status_code)
|
||||
self.text = text
|
||||
|
||||
|
||||
class PersimmonsMultiUserIsolationTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_distinct_callers_get_distinct_repository_sets(self) -> None:
|
||||
matcha = _ActorScopedRequester(
|
||||
"matcha",
|
||||
set(_ActorScopedRequester.ALL_REPOS),
|
||||
"MATCHA_USER_TOKEN",
|
||||
)
|
||||
chai = _ActorScopedRequester(
|
||||
"chai",
|
||||
{"chai/pointtest"},
|
||||
"CHAI_USER_TOKEN",
|
||||
)
|
||||
|
||||
matcha_repos = await matcha._repositories({})
|
||||
chai_repos = await chai._repositories({})
|
||||
|
||||
self.assertEqual(
|
||||
[repo["full_name"] for repo in matcha_repos],
|
||||
_ActorScopedRequester.ALL_REPOS,
|
||||
)
|
||||
self.assertEqual(
|
||||
[repo["full_name"] for repo in chai_repos],
|
||||
["chai/pointtest"],
|
||||
)
|
||||
|
||||
async def test_user_vault_listing_is_bound_to_session(self) -> None:
|
||||
matcha = _ActorScopedRequester("matcha", set(), "MATCHA_USER_TOKEN")
|
||||
chai = _ActorScopedRequester("chai", set(), "CHAI_USER_TOKEN")
|
||||
|
||||
matcha_vault = await list_connections(matcha, include_legacy=False)
|
||||
chai_vault = await list_connections(chai, include_legacy=False)
|
||||
|
||||
self.assertEqual(
|
||||
[item["name"] for item in matcha_vault["connections"]],
|
||||
["MATCHA_USER_TOKEN"],
|
||||
)
|
||||
self.assertEqual(
|
||||
[item["name"] for item in chai_vault["connections"]],
|
||||
["CHAI_USER_TOKEN"],
|
||||
)
|
||||
|
||||
async def test_guessed_repository_cannot_bypass_caller_authorization(self) -> None:
|
||||
chai = _ActorScopedRequester("chai", {"chai/pointtest"}, "CHAI_USER_TOKEN")
|
||||
|
||||
with self.assertRaises(RuntimeError):
|
||||
await list_connections(
|
||||
chai,
|
||||
repository="matcha/teahub",
|
||||
include_user=False,
|
||||
include_legacy=False,
|
||||
)
|
||||
|
||||
with self.assertRaises(RuntimeError):
|
||||
await chai(
|
||||
"PUT",
|
||||
"repos/matcha/teahub/actions/secrets/TEST_FIXTURE",
|
||||
json={"data": "fixture-only"},
|
||||
)
|
||||
|
||||
self.assertEqual(chai.posts, [])
|
||||
|
||||
async def test_authorized_repository_projects_safe_secret_metadata_only(self) -> None:
|
||||
chai = _ActorScopedRequester("chai", {"chai/pointtest"}, "CHAI_USER_TOKEN")
|
||||
listed = await list_connections(
|
||||
chai,
|
||||
repository="chai/pointtest",
|
||||
include_user=False,
|
||||
include_legacy=False,
|
||||
)
|
||||
item = listed["connections"][0]
|
||||
self.assertEqual(item["name"], "POINTTEST_TOKEN")
|
||||
self.assertEqual(item["repository"], "chai/pointtest")
|
||||
self.assertTrue(item["available"])
|
||||
self.assertNotIn("credential", item)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user