ITWorx_MCP_Hub-Public/tests/test_codex_bridge.py

from __future__ import annotations

import base64
from pathlib import Path
from typing import Any

import httpx

from packages.hub_clients import StdioBridge
from scripts.mcp_stdio_bridge import (
    ClientCredentialsTokenProvider,
    HttpTransport,
    _last_sse_json,
)


class _Response:
    def __init__(self, payload: dict[str, Any]) -> None:
        self.payload = payload

    def raise_for_status(self) -> None:
        return None

    def json(self) -> dict[str, Any]:
        return self.payload


class _TokenClient:
    def __init__(self) -> None:
        self.calls: list[dict[str, Any]] = []

    def post(self, _url: str, **kwargs: Any) -> _Response:
        self.calls.append(kwargs)
        return _Response({"access_token": "signed-token", "expires_in": 300})


def test_client_credentials_provider_encodes_authentik_app_password_and_caches(
    tmp_path: Path,
) -> None:
    secret = tmp_path / "credential"
    secret.write_text("app-password\n", encoding="utf-8")
    provider = ClientCredentialsTokenProvider(
        token_endpoint="https://auth.itworx.test/application/o/token/",  # noqa: S106
        client_id="itworx-mcp-hub",
        username="svc-mcp-codex-ludarium",
        secret_file=secret,
    )
    client = _TokenClient()
    provider._client = client  # type: ignore[assignment]  # deterministic network seam

    assert provider() == "signed-token"
    assert provider() == "signed-token"
    assert len(client.calls) == 1
    form = client.calls[0]["data"]
    decoded = base64.b64decode(form["client_secret"]).decode("utf-8")
    assert decoded == "svc-mcp-codex-ludarium:app-password"
    assert form["grant_type"] == "client_credentials"
    assert form["scope"] == "openid profile"


def test_client_credentials_provider_rejects_symlink_secret(tmp_path: Path) -> None:
    target = tmp_path / "target"
    target.write_text("secret", encoding="utf-8")
    link = tmp_path / "link"
    try:
        link.symlink_to(target)
    except OSError:
        return
    provider = ClientCredentialsTokenProvider(
        token_endpoint="https://auth.itworx.test/application/o/token/",  # noqa: S106
        client_id="client",
        username="service",
        secret_file=link,
    )
    try:
        provider()
    except ValueError as exc:
        assert "secret file is invalid" in str(exc)
    else:
        raise AssertionError("symlink credential was accepted")


def test_stdio_bridge_supports_codex_lifecycle_notifications_without_fake_response() -> None:
    class Transport:
        def send(
            self, request: dict[str, Any], *, token: str, endpoint: str
        ) -> dict[str, Any] | None:
            del token, endpoint
            if request["method"] == "notifications/initialized":
                return None
            return {"jsonrpc": "2.0", "id": request.get("id"), "result": {}}

    outputs = tuple(
        StdioBridge(endpoint="https://mcp.itworx.test/mcp/ludarium/prod").forward(
            [
                '{"jsonrpc":"2.0","id":1,"method":"initialize"}',
                '{"jsonrpc":"2.0","method":"notifications/initialized"}',
                '{"jsonrpc":"2.0","id":2,"method":"prompts/list"}',
            ],
            Transport(),
            token_provider=lambda: "token",
        )
    )
    assert len(outputs) == 2
    assert '"id":1' in outputs[0]
    assert '"id":2' in outputs[1]


def test_sse_parser_returns_last_json_rpc_event() -> None:
    body = 'event: message\ndata: {"jsonrpc":"2.0","id":1,"result":{}}\n\n'
    assert _last_sse_json(body)["id"] == 1


def test_http_transport_preserves_streamable_http_session() -> None:
    class Response:
        status_code = 200
        content = b"{}"
        headers = {"content-type": "application/json", "mcp-session-id": "session-1"}

        def raise_for_status(self) -> None:
            return None

        def json(self) -> dict[str, Any]:
            return {"jsonrpc": "2.0", "id": 1, "result": {}}

    class Client:
        def __init__(self) -> None:
            self.headers: list[dict[str, str]] = []

        def post(self, _endpoint: str, **kwargs: Any) -> Response:
            self.headers.append(kwargs["headers"])
            return Response()

    transport = HttpTransport()
    client = Client()
    transport.client = client  # type: ignore[assignment]
    request = {"jsonrpc": "2.0", "id": 1, "method": "initialize"}
    transport.send(
        request,
        token="token",  # noqa: S106
        endpoint="https://mcp.itworx.test/mcp",
    )
    transport.send(
        request,
        token="token",  # noqa: S106
        endpoint="https://mcp.itworx.test/mcp",
    )
    assert "Mcp-Session-Id" not in client.headers[0]
    assert client.headers[1]["Mcp-Session-Id"] == "session-1"


def test_token_endpoint_must_be_https(tmp_path: Path) -> None:
    try:
        ClientCredentialsTokenProvider(
            token_endpoint="http://auth.itworx.test/token",  # noqa: S106
            client_id="client",
            username="service",
            secret_file=tmp_path / "secret",
        )
    except ValueError as exc:
        assert "HTTPS" in str(exc)
    else:
        raise AssertionError("insecure OAuth endpoint was accepted")


def test_httpx_types_are_importable_for_runtime_packaging() -> None:
    assert httpx.Client is not None