search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
1"""tests for pub-search MCP server."""
2
3import pytest
4from mcp.types import TextContent
5
6from fastmcp.client import Client
7from fastmcp.client.transports import FastMCPTransport
8
9from pub_search._types import Document, EndpointTiming, PopularSearch, SearchResult, Stats, Tag
10from pub_search.server import mcp
11
12
13class TestTypes:
14 """tests for type definitions."""
15
16 def test_search_result(self):
17 """SearchResult can be constructed."""
18 r = SearchResult(
19 type="article",
20 uri="at://did:plc:abc/pub.leaflet.document/123",
21 did="did:plc:abc",
22 title="test article",
23 snippet="this is a test...",
24 createdAt="2025-01-01T00:00:00Z",
25 rkey="123",
26 basePath="gyst.leaflet.pub",
27 platform="leaflet",
28 url="https://gyst.leaflet.pub/123",
29 )
30 assert r.type == "article"
31 assert r.uri == "at://did:plc:abc/pub.leaflet.document/123"
32 assert r.title == "test article"
33 assert r.platform == "leaflet"
34 assert r.url == "https://gyst.leaflet.pub/123"
35
36 def test_search_result_looseleaf(self):
37 """SearchResult supports looseleaf type."""
38 r = SearchResult(
39 type="looseleaf",
40 uri="at://did:plc:abc/pub.leaflet.document/456",
41 did="did:plc:abc",
42 title="standalone doc",
43 snippet="no publication...",
44 rkey="456",
45 )
46 assert r.type == "looseleaf"
47 assert r.basePath == ""
48
49 def test_search_result_publication(self):
50 """SearchResult supports publication type."""
51 r = SearchResult(
52 type="publication",
53 uri="at://did:plc:abc/pub.leaflet.publication/789",
54 did="did:plc:abc",
55 title="my blog",
56 snippet="a personal blog...",
57 rkey="789",
58 basePath="/blog",
59 )
60 assert r.type == "publication"
61
62 def test_tag(self):
63 """Tag can be constructed."""
64 t = Tag(tag="python", count=42)
65 assert t.tag == "python"
66 assert t.count == 42
67
68 def test_popular_search(self):
69 """PopularSearch can be constructed."""
70 p = PopularSearch(query="rust async", count=100)
71 assert p.query == "rust async"
72 assert p.count == 100
73
74 def test_stats_minimal(self):
75 """Stats can be constructed with just documents/publications."""
76 s = Stats(documents=1000, publications=50)
77 assert s.documents == 1000
78 assert s.publications == 50
79 assert s.embeddings == 0
80 assert s.timing == {}
81
82 def test_stats_full(self):
83 """Stats can be constructed with all fields from API."""
84 s = Stats(
85 documents=6527,
86 publications=2335,
87 embeddings=6527,
88 searches=5321,
89 errors=0,
90 started_at=1767333441,
91 cache_hits=978,
92 cache_misses=627,
93 timing={
94 "search_keyword": EndpointTiming(
95 count=320, avg_ms=140.1, p50_ms=7.7, p95_ms=616.2, p99_ms=1090.1, max_ms=7294.9
96 ),
97 },
98 )
99 assert s.embeddings == 6527
100 assert s.cache_hits == 978
101 assert s.timing["search_keyword"].p50_ms == 7.7
102
103 def test_document(self):
104 """Document can be constructed with full content."""
105 d = Document(
106 uri="at://did:plc:abc/pub.leaflet.document/123",
107 title="full article",
108 content="this is the full content of the article...",
109 createdAt="2025-01-01T00:00:00Z",
110 tags=["python", "tutorial"],
111 publicationUri="at://did:plc:abc/pub.leaflet.publication/blog",
112 )
113 assert d.uri == "at://did:plc:abc/pub.leaflet.document/123"
114 assert "full content" in d.content
115 assert "python" in d.tags
116
117
118class TestMcpServerImports:
119 """tests for MCP server module imports."""
120
121 def test_mcp_server_imports(self):
122 """mcp server can be imported without errors."""
123 from pub_search import mcp
124
125 assert mcp.name == "pub-search"
126
127 def test_exports(self):
128 """all expected exports are available."""
129 from pub_search import main, mcp
130
131 assert mcp is not None
132 assert main is not None
133 assert callable(main)
134
135
136class TestMcpServerRegistration:
137 """tests for MCP server tool/prompt/resource registration."""
138
139 @pytest.fixture
140 def client(self):
141 """Create a FastMCP client for testing."""
142 return Client(transport=FastMCPTransport(mcp))
143
144 async def test_list_tools(self, client):
145 """verify all expected tools are registered."""
146 async with client:
147 tools = await client.list_tools()
148
149 tool_names = {t.name for t in tools}
150 expected = {"search", "get_document", "find_similar", "get_tags", "get_stats", "get_popular"}
151 assert expected == tool_names
152
153 async def test_list_prompts(self, client):
154 """verify prompts are registered."""
155 async with client:
156 prompts = await client.list_prompts()
157
158 prompt_names = {p.name for p in prompts}
159 assert "usage_guide" in prompt_names
160 assert "search_tips" in prompt_names
161
162 async def test_list_resources(self, client):
163 """verify resources are registered."""
164 async with client:
165 resources = await client.list_resources()
166
167 resource_uris = {str(r.uri) for r in resources}
168 assert "pub-search://stats" in resource_uris
169
170 async def test_usage_guide_prompt_content(self, client):
171 """usage_guide prompt returns helpful content."""
172 async with client:
173 result = await client.get_prompt("usage_guide")
174
175 assert len(result.messages) > 0
176 content = result.messages[0].content
177 assert isinstance(content, TextContent)
178 assert "pub-search" in content.text
179 assert "search" in content.text
180
181 async def test_search_tips_prompt_content(self, client):
182 """search_tips prompt returns helpful content."""
183 async with client:
184 result = await client.get_prompt("search_tips")
185
186 assert len(result.messages) > 0
187 content = result.messages[0].content
188 assert isinstance(content, TextContent)
189 assert "search" in content.text.lower()