1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
"""Tests for MCP client integration and settings API."""
from __future__ import annotations
import tempfile
from pathlib import Path
import pytest
from starlette.testclient import TestClient
from ollama_web.app import create_app
from ollama_web.config import settings
from ollama_web.mcp import (
_clean_schema,
_normalized_tool_name,
_parse_tool_name,
load_mcp_config,
save_mcp_config,
)
from ollama_web.tools.registry import ToolRegistry
@pytest.fixture
def client(monkeypatch):
monkeypatch.setattr(settings, "pin", "123456")
monkeypatch.setattr(settings, "secret_key", "test-secret")
monkeypatch.setattr(settings, "allowed_origins", [])
with tempfile.TemporaryDirectory() as tmp:
monkeypatch.setattr(settings, "data_dir", tmp)
client = TestClient(create_app())
login = client.post("/api/auth/login", json={"pin": "123456"})
assert login.status_code == 200
client.csrf = login.headers["x-csrf-token"]
yield client
def test_load_save_mcp_config():
with tempfile.TemporaryDirectory() as tmp:
config = {
"mcpServers": {
"fs": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
},
"remote": {
"url": "https://example.com/mcp",
"headers": {"Authorization": "Bearer x"},
},
}
}
save_mcp_config(tmp, config)
loaded = load_mcp_config(tmp)
assert loaded == config
assert (Path(tmp) / "mcpServers.json").exists()
def test_load_mcp_config_missing_file():
with tempfile.TemporaryDirectory() as tmp:
loaded = load_mcp_config(tmp)
assert loaded == {"mcpServers": {}}
def test_mcp_config_api(client):
res = client.get("/api/mcp/servers")
assert res.status_code == 200
assert res.json() == {"mcpServers": {}}
payload = {
"mcpServers": {
"fs": {"command": "npx", "args": ["-y", "server"]},
"remote": {"url": "https://example.com/mcp", "timeout": 60},
}
}
res = client.put(
"/api/mcp/servers",
json=payload,
headers={"X-CSRF-Token": client.csrf},
)
assert res.status_code == 200
data = res.json()
assert data["mcpServers"]["fs"]["command"] == "npx"
assert data["mcpServers"]["remote"]["timeout"] == 60.0
res = client.get("/api/mcp/servers")
assert res.json() == data
def test_mcp_config_api_rejects_unknown_fields(client):
payload = {"mcpServers": {"fs": {"command": "npx", "invalid": 1}}}
res = client.put(
"/api/mcp/servers",
json=payload,
headers={"X-CSRF-Token": client.csrf},
)
assert res.status_code == 400
def test_mcp_config_api_requires_transport(client):
payload = {"mcpServers": {"fs": {}}}
res = client.put(
"/api/mcp/servers",
json=payload,
headers={"X-CSRF-Token": client.csrf},
)
assert res.status_code == 400
def test_normalized_tool_name():
assert _normalized_tool_name("server1", "tool_a") == "mcp__server1__tool_a"
def test_parse_tool_name():
assert _parse_tool_name("mcp__server1__tool_a") == ("server1", "tool_a")
assert _parse_tool_name("web_search") is None
def test_clean_schema():
schema = {
"type": "object",
"properties": {"path": {"type": "string"}},
"required": ["path"],
}
cleaned = _clean_schema(schema)
assert cleaned["type"] == "object"
assert cleaned["required"] == ["path"]
def test_tool_registry_mcp_tool():
reg = ToolRegistry()
def executor(full_name, arguments=""):
return "ok"
reg.register_mcp_tool(
"mcp__s__t", {"type": "function", "function": {"name": "mcp__s__t"}}, executor
)
assert "mcp__s__t" in reg.names
# Executor is registered internally for execution but must not be exposed
# to ollama as a callable (ollama needs the dict definition only).
assert len(reg.ollama_tools) == 1
result = reg.execute("mcp__s__t", {})
assert result == "ok"
|
