ITWorx_MCP_Hub-Public/tests/test_gitea_public_surface_reconcile.py

from __future__ import annotations

from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest

from scripts.reconcile_gitea_public_surface import (
    REQUIRED_RECOVERY_TOOL,
    _load_surface,
    _published_tool_name,
    reconcile_gitea_public_surface,
)
from services.control_api.contextforge.models import GatewayTool

ROOT = Path(__file__).resolve().parents[1]
CATALOG = ROOT / "catalog"


class SurfaceClient:
    def __init__(
        self,
        *,
        missing_before_refresh: set[str] | None = None,
        ignore_server_update: bool = False,
    ) -> None:
        required, prompt_patterns = _load_surface(CATALOG)
        self.required = list(required)
        self.gitea_required = [
            name for name in required if name.startswith("gitea.") or name in {"search", "fetch"}
        ]
        self.current_gitea = [
            name for name in self.gitea_required if name not in (missing_before_refresh or set())
        ]
        self.prompt_patterns = list(prompt_patterns)
        self.ignore_server_update = ignore_server_update
        self.refresh_count = 0
        self.server_update_count = 0
        self.live_ids: tuple[str, ...] = ()
        self.id_to_name = {f"tool-{index:03d}": name for index, name in enumerate(required)}
        self.name_to_id = {name: external_id for external_id, name in self.id_to_name.items()}

    def _find(self, slug: str) -> Any:
        assert slug == "itworx-gitea"
        return SimpleNamespace(external_id="gateway-gitea")

    def list_gateway_tools(self, _gateway: Any) -> tuple[GatewayTool, ...]:
        return tuple(
            GatewayTool(
                external_id=self.name_to_id[name],
                name=name,
                published_name=name,
            )
            for name in self.current_gitea
        )

    def ensure_canonical_tool_names(
        self, tools: tuple[GatewayTool, ...]
    ) -> tuple[GatewayTool, ...]:
        return tools

    def refresh_gateway(self, _gateway: Any) -> None:
        self.refresh_count += 1
        self.current_gitea = list(self.gitea_required)

    def ensure_virtual_server(self, desired: Any) -> Any:
        self.desired = desired
        return SimpleNamespace(external_id="server-gitea-all")

    def _global_tool_payload(self) -> list[dict[str, str]]:
        available = set(self.current_gitea)
        available.update(name for name in self.required if name not in self.gitea_required)
        return [
            {
                "id": self.name_to_id[name],
                # Deliberately make originalName different. customName is the
                # actual client-facing name and must win in the verifier.
                "originalName": f"imported-{name}",
                "customName": name,
            }
            for name in self.required
            if name in available
        ]

    def _request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, Any] | None = None,
        json_body: dict[str, Any] | None = None,
        retryable: bool = False,
    ) -> Any:
        del params, retryable
        if method == "GET" and path == "/tools":
            return self._global_tool_payload()
        if method == "GET" and path == "/prompts":
            return [
                {
                    "id": f"prompt-{index:03d}",
                    "customName": name,
                }
                for index, name in enumerate(self.prompt_patterns)
            ]
        if method == "PUT" and path == "/servers/server-gitea-all":
            self.server_update_count += 1
            assert json_body is not None
            if not self.ignore_server_update:
                self.live_ids = tuple(sorted(str(item) for item in json_body["associated_tools"]))
            return {
                "id": "server-gitea-all",
                "name": "gitea-all-prod-readonly",
                "associatedTools": list(self.live_ids),
                "associatedPrompts": [],
                "visibility": "public",
                "enabled": True,
            }
        if method == "GET" and path == "/servers/server-gitea-all/tools":
            return [
                {
                    "id": external_id,
                    "originalName": f"imported-{self.id_to_name[external_id]}",
                    "customName": self.id_to_name[external_id],
                }
                for external_id in self.live_ids
            ]
        raise AssertionError(f"unexpected request: {method} {path}")


def test_reconciles_complete_live_surface_and_keeps_pr_recovery_available() -> None:
    client = SurfaceClient()

    result = reconcile_gitea_public_surface(client, CATALOG, attempts=1, delay_seconds=0)  # type: ignore[arg-type]

    assert result["status"] == "pass"
    assert result["pullRequestSyncLive"] is True
    assert result["toolCount"] == len(client.required)
    assert client.server_update_count == 1
    assert client.refresh_count == 0
    live_names = {client.id_to_name[item] for item in client.live_ids}
    assert REQUIRED_RECOVERY_TOOL in live_names
    assert "project.status.get" not in live_names


def test_published_custom_name_wins_over_import_time_original_name() -> None:
    assert (
        _published_tool_name(
            {
                "id": "tool-1",
                "originalName": "itworx-gitea-gitea.pullrequest.sync",
                "customName": "gitea.pullrequest.sync",
            }
        )
        == "gitea.pullrequest.sync"
    )


def test_refreshes_gateway_once_when_sync_tool_is_missing_from_cached_inventory() -> None:
    client = SurfaceClient(missing_before_refresh={REQUIRED_RECOVERY_TOOL})

    result = reconcile_gitea_public_surface(client, CATALOG, attempts=1, delay_seconds=0)  # type: ignore[arg-type]

    assert result["status"] == "pass"
    assert client.refresh_count == 1
    assert REQUIRED_RECOVERY_TOOL in {client.id_to_name[item] for item in client.live_ids}


def test_fails_closed_when_virtual_server_update_does_not_change_live_membership() -> None:
    client = SurfaceClient(ignore_server_update=True)

    with pytest.raises(RuntimeError, match="read-back mismatch"):
        reconcile_gitea_public_surface(client, CATALOG, attempts=1, delay_seconds=0)  # type: ignore[arg-type]


def test_long_running_pr_primitives_are_part_of_the_fixed_surface() -> None:
    required, _ = _load_surface(CATALOG)
    expected = {
        "gitea.pullrequest.sync",
        "gitea.ref.compare",
        "gitea.change.review",
        "gitea.change.status",
        "gitea.commitstatus.list",
        "gitea.validation.get",
        "gitea.validation.list",
        "gitea.validation.run",
        "gitea.workspace.run",
    }
    assert expected <= set(required)