pyMCPBroker/tests/test_broker_end_to_end.py

from __future__ import annotations

import json
import shlex
import sys
from pathlib import Path

from fastapi.testclient import TestClient

from pyMCPBroker.app import create_app
from pyMCPBroker.broker import Broker
from pyMCPBroker.config import load_config



def make_config(tmp_path: Path) -> Path:
    server = Path(__file__).with_name("fake_mcp_server.py")
    command = f"{shlex.quote(sys.executable)} {shlex.quote(str(server))}"
    config = {
        "tree": [
            {
                "path": "/repo",
                "type": "node",
                "summary": "Repository operations",
                "children": [
                    {
                        "path": "/repo/read",
                        "type": "node",
                        "summary": "Read repository data",
                        "source": {
                            "backend": "stdio",
                            "command": command,
                            "tool_filter": ["get_*", "list_*", "!delete_*"],
                            "path_aliases": {"get_file_contents": "get_file"},
                            "tool_overrides": {
                                "get_file_contents": {
                                    "summary": "Read one file from a repository",
                                    "max_output_chars": 1200,
                                    "example_args": {
                                        "owner": "myorg",
                                        "repo": "demo-repo",
                                        "ref": "main",
                                        "filePath": "README.md",
                                    },
                                }
                            },
                        },
                    }
                ],
            }
        ]
    }
    path = tmp_path / "config.json"
    path.write_text(json.dumps(config), encoding="utf-8")
    return path


def test_meta_end_to_end(tmp_path: Path) -> None:
    cfg = load_config(make_config(tmp_path))
    broker = Broker(cfg)
    app = create_app(broker)

    with TestClient(app) as client:
        r = client.post("/meta_tree", json={"path": "/"})
        assert r.status_code == 200
        assert r.json()["children"][0]["path"] == "/repo"

        r = client.post("/meta_tree", json={"path": "/repo/read"})
        assert r.status_code == 200
        body = r.json()
        child_summaries = {child["path"]: child["summary"] for child in body["children"]}
        assert "/repo/read/get_file" in child_summaries
        assert "/repo/read/list_branches" in child_summaries
        assert child_summaries["/repo/read/list_branches"] == "List all branches in a repository for a given owner and repo."

        r = client.post("/meta_desc", json={"path": "/repo/read/get_file"})
        body = r.json()
        assert body["summary"] == "Read one file from a repository"
        assert body["args_schema"]["required"] == ["owner", "repo", "ref", "filePath"]
        assert body["example_args"]["repo"] == "demo-repo"

        r = client.post(
            "/meta_call",
            json={
                "path": "/repo/read/get_file",
                "args": {"owner": "acme", "repo": "demo", "ref": "main", "filePath": "README.md"},
            },
        )
        body = r.json()
        assert body["ok"] is True
        assert body["result"]["truncated"] is True
        assert body["result"]["result"]["structuredContent"]["repo"] == "demo"


def test_meta_call_invalid_args(tmp_path: Path) -> None:
    cfg = load_config(make_config(tmp_path))
    broker = Broker(cfg)
    app = create_app(broker)

    with TestClient(app) as client:
        r = client.post("/meta_call", json={"path": "/repo/read/get_file", "args": {"owner": "acme"}})
        body = r.json()
        assert r.status_code == 400
        assert body["error_code"] == "invalid_arguments"
        assert "required property" in body["message"]


def test_meta_tree_rejects_tool_path(tmp_path: Path) -> None:
    cfg = load_config(make_config(tmp_path))
    broker = Broker(cfg)
    app = create_app(broker)

    with TestClient(app) as client:
        r = client.post("/meta_tree", json={"path": "/repo/read/get_file"})
        assert r.status_code == 400
        assert r.json()["error_code"] == "not_a_node"


def test_secret_auth(tmp_path: Path) -> None:
    cfg = load_config(make_config(tmp_path))
    broker = Broker(cfg)
    app = create_app(broker, shared_secret="sekret")

    with TestClient(app) as client:
        assert client.post("/meta_tree", json={"path": "/"}).status_code == 401
        assert client.post("/meta_tree", json={"path": "/"}, headers={"Authorization": "Bearer sekret"}).status_code == 200


def test_no_filter_exposes_all_tools(tmp_path: Path) -> None:
    server = Path(__file__).with_name("fake_mcp_server.py")
    command = f"{shlex.quote(sys.executable)} {shlex.quote(str(server))}"
    path = tmp_path / "config.json"
    path.write_text(
        json.dumps(
            {
                "tree": [
                    {
                        "path": "/repo",
                        "type": "node",
                        "source": {"backend": "stdio", "command": command},
                    }
                ]
            }
        ),
        encoding="utf-8",
    )
    cfg = load_config(path)
    broker = Broker(cfg)
    app = create_app(broker)

    with TestClient(app) as client:
        r = client.post("/meta_tree", json={"path": "/repo"})
        body = r.json()
        child_paths = {child["path"] for child in body["children"]}
        assert "/repo/get_file_contents" in child_paths
        assert "/repo/delete_file" in child_paths


def test_explicit_root_is_optional(tmp_path: Path) -> None:
    path = tmp_path / "config.json"
    path.write_text(
        json.dumps(
            {
                "tree": {
                    "path": "/",
                    "type": "node",
                    "summary": "Configured root",
                    "children": [
                        {"path": "/repo", "type": "node", "summary": "Repository operations"}
                    ],
                }
            }
        ),
        encoding="utf-8",
    )
    cfg = load_config(path)
    broker = Broker(cfg)
    app = create_app(broker)

    with TestClient(app) as client:
        r = client.post("/meta_tree", json={"path": "/"})
        assert r.status_code == 200
        assert r.json()["children"][0]["path"] == "/repo"


def test_summary_falls_back_to_description_when_title_is_missing(tmp_path: Path) -> None:
    server = Path(__file__).with_name("fake_mcp_server.py")
    command = f"{shlex.quote(sys.executable)} {shlex.quote(str(server))}"
    path = tmp_path / "config.json"
    path.write_text(
        json.dumps(
            {
                "tree": [
                    {
                        "path": "/repo",
                        "type": "node",
                        "source": {
                            "backend": "stdio",
                            "command": command,
                            "tool_filter": ["list_*"],
                        },
                    }
                ]
            }
        ),
        encoding="utf-8",
    )
    cfg = load_config(path)
    broker = Broker(cfg)
    app = create_app(broker)

    with TestClient(app) as client:
        r = client.post("/meta_tree", json={"path": "/repo"})
        body = r.json()
        assert r.status_code == 200
        assert body["children"][0]["path"] == "/repo/list_branches"
        assert body["children"][0]["summary"] == "List all branches in a repository for a given owner and repo."