app/tests/test_persimmons_web.py

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="{&quot;provider&quot;:&quot;Gitea&quot;}"
          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()