62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
"""Tests for duckduckgo-mcp (run with duckduckgo-mcp/.venv: needs mcp 1.x FastMCP).
|
|
|
|
duckduckgo-mcp\\.venv\\Scripts\\python.exe -m pytest tests/test_duckduckgo_mcp.py -q
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("mcp.server.fastmcp")
|
|
pytest.importorskip("duckduckgo_mcp")
|
|
|
|
from mcp.server.fastmcp.exceptions import ToolError # noqa: E402
|
|
|
|
from duckduckgo_mcp import server as srv # noqa: E402
|
|
|
|
|
|
class FakeDDGS:
|
|
def __init__(self, fail: Exception | None = None):
|
|
self.fail = fail
|
|
self.threads: list[str] = []
|
|
|
|
def text(self, query, **kwargs):
|
|
self.threads.append(threading.current_thread().name)
|
|
if self.fail:
|
|
raise self.fail
|
|
return [{"title": "T", "href": "https://example.org", "body": "B"}]
|
|
|
|
def news(self, query, **kwargs):
|
|
return self.text(query, **kwargs)
|
|
|
|
|
|
def test_search_runs_off_the_event_loop(monkeypatch):
|
|
fake = FakeDDGS()
|
|
monkeypatch.setattr(srv, "_get_ddgs", lambda: fake)
|
|
out = asyncio.run(srv.mcp.call_tool("ddg_search", {"query": "python"}))
|
|
assert "example.org" in str(out)
|
|
assert fake.threads and fake.threads[0] != threading.main_thread().name
|
|
|
|
|
|
def test_failures_are_tool_errors(monkeypatch):
|
|
monkeypatch.setattr(srv, "_get_ddgs", lambda: FakeDDGS(RuntimeError("Ratelimit")))
|
|
with pytest.raises(ToolError, match="Ratelimit"):
|
|
asyncio.run(srv.mcp.call_tool("ddg_search", {"query": "python"}))
|
|
with pytest.raises(ToolError, match="Ratelimit"):
|
|
asyncio.run(srv.mcp.call_tool("ddg_answers", {"query": "python"}))
|
|
|
|
|
|
def test_input_validation(monkeypatch):
|
|
monkeypatch.setattr(srv, "_get_ddgs", lambda: FakeDDGS())
|
|
with pytest.raises(ToolError, match="empty"):
|
|
asyncio.run(srv.mcp.call_tool("ddg_news", {"query": " "}))
|
|
with pytest.raises(ToolError, match="safesearch"):
|
|
asyncio.run(srv.mcp.call_tool("ddg_search", {"query": "x", "safesearch": "strict"}))
|
|
|
|
|
|
def test_tool_annotations():
|
|
for t in asyncio.run(srv.mcp.list_tools()):
|
|
assert t.annotations.readOnlyHint is True and t.annotations.openWorldHint is True, t.name
|