portfolio-optimiser/tests/test_okf_server_smoke.py

"""Smoke test: does a run's own MCP path actually reach the ``okf`` v1.1 server?

The README documents ``examples/okf-server.mcp.json``; ``test_okf_server_example.py`` only proves
the file loads. These tests connect through the same two functions a run uses
(``load_mcp_config`` -> ``build_mcp_tools``) and the same ``async with`` a run enters, against a
tiny invented collection built in ``tmp_path``. No model call, no network.

They are skipped, with the reason named, when no ``okf`` >= 1.1.0 is found on ``PATH`` outside
this environment's own ``bin``. The version is MEASURED from the server's own ``initialize``
answer (``serverInfo.version``), never assumed from the name.
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any

import pytest
from agent_framework.exceptions import ToolException

from portfolio_optimiser.mcp_tools import build_mcp_tools, load_mcp_config

FOUR = ["okf_list", "okf_describe", "okf_ask", "okf_fetch"]
VENV_BIN = str(Path(sys.executable).parent)

_INIT = json.dumps(
    {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-06-18",
            "capabilities": {},
            "clientInfo": {"name": "smoke", "version": "0"},
        },
    }
)


def _server_version(exe: str, root: Path) -> tuple[int, ...] | None:
    """The version the server reports about itself, or None if ``exe`` serves no MCP at all."""
    try:
        done = subprocess.run(
            [exe, "mcp", "--root", str(root)],
            input=_INIT + "\n",
            capture_output=True,
            text=True,
            timeout=60,
        )
        version = json.loads(done.stdout.splitlines()[0])["result"]["serverInfo"]["version"]
        return tuple(int(part) for part in version.split(".")[:3])
    except (IndexError, KeyError, ValueError, OSError, subprocess.TimeoutExpired):
        return None


@pytest.fixture
def okf_1_1(tmp_path: Path) -> str:
    """Absolute path of an ``okf`` >= 1.1.0 on PATH, looked up OUTSIDE this environment's bin."""
    outside = os.pathsep.join(
        p for p in os.environ.get("PATH", "").split(os.pathsep) if p and p != VENV_BIN
    )
    exe = shutil.which("okf", path=outside)
    if exe is None:
        pytest.skip(
            "no `okf` on PATH outside this environment -- the smoke test needs okf >= 1.1.0"
        )
    probe_root = tmp_path / "empty-root"
    probe_root.mkdir()
    version = _server_version(exe, probe_root)
    if version is None or version < (1, 1, 0):
        pytest.skip(f"`{exe}` reports version {version} over MCP -- the smoke test needs >= 1.1.0")
    return exe


@pytest.fixture
def invented_root(tmp_path: Path, okf_1_1: str) -> Path:
    """A two-document collection about an invented island, built with ``okf build``."""
    inbox = tmp_path / "inbox"
    inbox.mkdir()
    (inbox / "lighthouse.md").write_text(
        "# Lighthouse keeping\n\n## Lamp maintenance\n\n"
        "The quartz lamp at Vardholm point is polished every Tuesday with walrus wax.\n",
        encoding="utf-8",
    )
    (inbox / "ferry.md").write_text(
        "# Ferry timetable\n\n## Winter crossings\n\n"
        "The Vardholm ferry sails twice a day in winter, leaving at dawn and dusk.\n",
        encoding="utf-8",
    )
    root = tmp_path / "root"
    subprocess.run(
        [okf_1_1, "build", str(inbox), "--bundle", str(root / "vardholm"), "--gate", "none"]
        + ["--bundle-id", "vardholm", "--okf-version", "1.0"],
        check=True,
        capture_output=True,
        timeout=120,
    )
    return root


def _config(tmp_path: Path, command: str, root: Path, allowed: list[str]) -> Path:
    path = tmp_path / "okf.mcp.json"
    server = {
        "name": "okf",
        "transport": "stdio",
        "command": command,
        "args": ["mcp", "--root", str(root)],
        "allowed_tools": allowed,
        "timeout_seconds": 60,
    }
    path.write_text(json.dumps({"servers": [server]}), encoding="utf-8")
    return path


def _text(result: Any) -> str:
    return "\n".join(getattr(item, "text", None) or "" for item in result)


async def test_okf_1_1_by_full_path_offers_four_tools_lists_and_answers(
    tmp_path: Path, okf_1_1: str, invented_root: Path
) -> None:
    (tool,) = build_mcp_tools(load_mcp_config(_config(tmp_path, okf_1_1, invented_root, FOUR)))
    async with tool:
        # (a) the four are offered
        assert sorted(f.name for f in tool.functions) == sorted(FOUR)
        # (b) okf_list names the invented collection
        listed = json.loads((await tool.call_tool("okf_list"))[0].text)
        assert [b["bundle_id"] for b in listed["bundles"]] == ["vardholm"]
        # (c) one okf_ask in the fixture's words delivers the excerpt
        asked = _text(
            await tool.call_tool(
                "okf_ask", bundle="vardholm", questions=["How is the quartz lamp polished?"]
            )
        )
        assert "walrus wax" in asked


async def test_allowed_tools_is_load_bearing_against_the_real_server(
    tmp_path: Path, okf_1_1: str, invented_root: Path
) -> None:
    """The server offers exactly the four, so (a) alone cannot show the allowlist filters.
    A narrower list must narrow what the agents are offered."""
    config = _config(tmp_path, okf_1_1, invented_root, ["okf_list", "okf_ask"])
    (tool,) = build_mcp_tools(load_mcp_config(config))
    async with tool:
        assert sorted(f.name for f in tool.functions) == ["okf_ask", "okf_list"]


@pytest.mark.xfail(
    strict=True,
    raises=ToolException,
    reason=(
        "MEASURED 2026-09-21: with the README's bare `okf` command and this environment's bin "
        "first on PATH (as under `uv run` or an activated venv), the name resolves to the `okf` "
        "script of po's own dependency llm-ingestion-okf 0.8.5, which has no `mcp` subcommand: "
        "stderr `okf: error: argument command: invalid choice: 'mcp'`, and MAF raises "
        "ToolException '... failed to initialize: Connection closed' on `async with`."
    ),
)
async def test_readme_bare_okf_command_reaches_the_1_1_server(
    tmp_path: Path, okf_1_1: str, invented_root: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("PATH", VENV_BIN + os.pathsep + os.environ.get("PATH", ""))
    (tool,) = build_mcp_tools(load_mcp_config(_config(tmp_path, "okf", invented_root, FOUR)))
    async with tool:
        assert sorted(f.name for f in tool.functions) == sorted(FOUR)