portfolio-optimiser/tests/test_mcp_tools.py

"""Concrete MCP servers as run-path tools (Trekk B1) — config + tool construction.

Krav 3: a run must be able to reach an external service (price register, meter data, contract
registry) while it is working. MAF already ships the client (``MCPStdioTool`` /
``MCPStreamableHTTPTool``), so what this module owns is the part MAF cannot decide for us: which
servers a run may contact, which of their tools it may call, what it waits, and where the
credential comes from.

Every refusal here exists because its absence is a real hazard, not for tidiness:

* an EMPTY ``allowed_tools`` would let a server expose any tool it likes to the agents — authority
  by whatever the far end happens to serve. You must name what you allow;
* a missing/blank timeout would be an unbounded wait against a third party (the repo's fail-fast
  invariant: stop criteria are never optional);
* an UNKNOWN field is refused rather than ignored, which is what keeps a literal secret from being
  parked in the config file and silently carried along;
* a named credential env var that is not set refuses the run instead of quietly calling the
  service unauthenticated.
"""

from __future__ import annotations

import json

import pytest
from agent_framework import MCPStdioTool, MCPStreamableHTTPTool
from pydantic import ValidationError

from portfolio_optimiser.mcp_tools import (
    McpServerConfig,
    build_mcp_tools,
    load_mcp_config,
    service_labels,
)

_STDIO = {
    "name": "maalerdata",
    "transport": "stdio",
    "command": "uvx",
    "args": ["maalerdata-mcp"],
    "allowed_tools": ["read_meter"],
    "timeout_seconds": 30,
}
_HTTP = {
    "name": "prisregister",
    "transport": "http",
    "url": "https://intern.example/mcp",
    "allowed_tools": ["lookup_unit_price", "list_categories"],
    "timeout_seconds": 15,
}


def _write(tmp_path, *servers) -> str:
    path = tmp_path / "mcp.json"
    path.write_text(json.dumps({"servers": list(servers)}), encoding="utf-8")
    return str(path)


# --- config shape ------------------------------------------------------------------------------


def test_stdio_and_http_configs_load(tmp_path) -> None:
    configs = load_mcp_config(_write(tmp_path, _STDIO, _HTTP))
    assert [c.name for c in configs] == ["maalerdata", "prisregister"]
    assert configs[0].transport == "stdio"
    assert configs[1].allowed_tools == ("lookup_unit_price", "list_categories")


def test_stdio_without_command_refused(tmp_path) -> None:
    """A stdio server with nothing to launch is not a server."""
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, {**_STDIO, "command": None}))


def test_http_without_url_refused(tmp_path) -> None:
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, {**_HTTP, "url": None}))


def test_transport_and_coordinates_must_agree(tmp_path) -> None:
    """A stdio server carrying a URL (or an http server carrying a command) is ambiguous about
    what would actually be contacted — refused rather than resolved by precedence."""
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, {**_STDIO, "url": "https://intern.example/mcp"}))
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, {**_HTTP, "command": "uvx"}))


def test_unknown_transport_refused(tmp_path) -> None:
    """The transport set is CLOSED — a config can never name arbitrary machinery to load
    (the same rule ``--embedder-config`` follows)."""
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, {**_STDIO, "transport": "carrier-pigeon"}))


def test_empty_allowed_tools_refused(tmp_path) -> None:
    """Closed by default: you must name the tools a run may call. An empty list would hand the
    agents whatever the far end chooses to expose."""
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, {**_STDIO, "allowed_tools": []}))


def test_non_positive_timeout_refused(tmp_path) -> None:
    """Fail-fast: an unbounded wait against a third party is exactly what stop criteria exist
    to prevent."""
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, {**_STDIO, "timeout_seconds": 0}))


def test_unknown_field_refused(tmp_path) -> None:
    """Refused, never ignored — and this is the rule that keeps a literal secret out of the file:
    there is no field for one, and an unrecognised key does not slip through unnoticed."""
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, {**_STDIO, "credential": "sk-live-not-here-please"}))


def test_duplicate_server_names_refused(tmp_path) -> None:
    """``name`` is how a server is identified in the egress declaration; two servers on one name
    would make the announcement ambiguous about what is contacted."""
    with pytest.raises(ValidationError):
        load_mcp_config(_write(tmp_path, _STDIO, {**_STDIO, "command": "other"}))


def test_missing_config_file_refused(tmp_path) -> None:
    with pytest.raises(FileNotFoundError):
        load_mcp_config(str(tmp_path / "nope.json"))


def test_malformed_config_refused(tmp_path) -> None:
    bad = tmp_path / "mcp.json"
    bad.write_text("not json", encoding="utf-8")
    with pytest.raises(ValidationError):
        load_mcp_config(str(bad))


# --- tool construction -------------------------------------------------------------------------


def test_builds_the_right_client_per_transport() -> None:
    tools = build_mcp_tools((McpServerConfig(**_STDIO), McpServerConfig(**_HTTP)))
    assert isinstance(tools[0], MCPStdioTool)
    assert isinstance(tools[1], MCPStreamableHTTPTool)


def test_allowed_tools_and_timeout_reach_the_client() -> None:
    """The allowlist and the timeout are the two limits that matter, so they are asserted on the
    built client — not merely on the config that was meant to produce it."""
    tool = build_mcp_tools((McpServerConfig(**_HTTP),))[0]
    assert set(tool.allowed_tools or ()) == {"lookup_unit_price", "list_categories"}
    assert tool.request_timeout == 15


def test_credential_is_read_from_the_environment_at_build_time(monkeypatch) -> None:
    """The config names an env var; the value never lives in the file."""
    monkeypatch.setenv("PRISREGISTER_TOKEN", "s3cret")
    cfg = McpServerConfig(**{**_HTTP, "credential_env": "PRISREGISTER_TOKEN"})
    tool = build_mcp_tools((cfg,))[0]
    assert isinstance(tool, MCPStreamableHTTPTool)  # built without raising


def test_missing_credential_env_refuses(monkeypatch) -> None:
    """A named-but-unset credential refuses rather than calling the service unauthenticated —
    an anonymous call can succeed with the wrong scope, which is worse than not calling."""
    monkeypatch.delenv("PRISREGISTER_TOKEN", raising=False)
    cfg = McpServerConfig(**{**_HTTP, "credential_env": "PRISREGISTER_TOKEN"})
    with pytest.raises(ValueError, match="PRISREGISTER_TOKEN"):
        build_mcp_tools((cfg,))


# --- the egress declaration --------------------------------------------------------------------


def test_service_labels_name_every_server_and_its_allowed_tools() -> None:
    """These strings go straight into the run announcement, so an operator can see WHAT will be
    contacted and WHICH tools are permitted before anything is spent."""
    labels = service_labels((McpServerConfig(**_STDIO), McpServerConfig(**_HTTP)))
    assert labels == (
        "maalerdata (read_meter)",
        "prisregister (lookup_unit_price, list_categories)",
    )