diff --git a/tests/test_persimmons_web.py b/tests/test_persimmons_web.py new file mode 100644 index 0000000..1517655 --- /dev/null +++ b/tests/test_persimmons_web.py @@ -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 = """ +
+ + +
+
+ + +
+ """ + 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 = """ + + """ + 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'' + '', + ) + 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'' + f'', + ) + 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, + '' + '', + ) + 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()