"""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 (why the README asks for a full path): with a 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_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)