a digital entity named phi that roams bsky
phi.zzstoatzz.io
1"""Tests for the search_network tool."""
2
3from unittest.mock import AsyncMock, MagicMock, patch
4
5import httpx
6
7
8async def _search_network(query: str) -> str:
9 """Extracted search_network logic matching agent.py implementation."""
10 try:
11 async with httpx.AsyncClient(timeout=15) as client:
12 r = await client.get(
13 "https://api.semble.so/api/search/semantic",
14 params={"query": query, "limit": 10},
15 )
16 r.raise_for_status()
17 data = r.json()
18
19 # response is {urls: [...], pagination: {...}}
20 items = data.get("urls") if isinstance(data, dict) else data
21 if not items:
22 return f"no network results for '{query}'"
23
24 lines = []
25 for item in items:
26 meta = item.get("metadata", {})
27 title = meta.get("title") or item.get("title") or "untitled"
28 url = item.get("url", "")
29 saves = item.get("urlLibraryCount") or 0
30 desc = meta.get("description") or ""
31 line = f"{title}"
32 if url:
33 line += f" — {url}"
34 if saves:
35 line += f" ({saves} saves)"
36 if desc:
37 line += f"\n {desc[:200]}"
38 lines.append(line)
39 return "\n\n".join(lines)
40 except Exception as e:
41 return f"network search failed: {e}"
42
43
44def _mock_response(status_code: int, json_data=None):
45 resp = MagicMock(spec=httpx.Response)
46 resp.status_code = status_code
47 resp.json.return_value = json_data if json_data is not None else {"urls": []}
48 resp.raise_for_status = MagicMock()
49 if status_code >= 400:
50 resp.raise_for_status.side_effect = httpx.HTTPStatusError(
51 "error", request=MagicMock(), response=resp
52 )
53 return resp
54
55
56def _semble_response(items):
57 """Wrap items in semble's {urls: [...], pagination: {...}} envelope."""
58 return {
59 "urls": items,
60 "pagination": {
61 "currentPage": 1,
62 "totalPages": 1,
63 "totalCount": len(items),
64 "hasMore": False,
65 "limit": 10,
66 },
67 }
68
69
70class TestSearchNetworkFormatting:
71 async def test_formats_results_with_all_fields(self):
72 resp = _mock_response(
73 200,
74 _semble_response(
75 [
76 {
77 "url": "https://atproto.com",
78 "metadata": {
79 "title": "AT Protocol",
80 "description": "Federated social networking protocol",
81 },
82 "urlLibraryCount": 5,
83 },
84 {
85 "url": "https://docs.bsky.app",
86 "metadata": {
87 "title": "Bluesky Docs",
88 "description": "Documentation for Bluesky",
89 },
90 "urlLibraryCount": 3,
91 },
92 ]
93 ),
94 )
95 with patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=resp):
96 result = await _search_network("atproto")
97 assert "AT Protocol — https://atproto.com (5 saves)" in result
98 assert "Federated social networking protocol" in result
99 assert "Bluesky Docs — https://docs.bsky.app (3 saves)" in result
100
101 async def test_formats_results_with_minimal_fields(self):
102 resp = _mock_response(
103 200,
104 _semble_response(
105 [
106 {
107 "url": "https://example.com/music",
108 "metadata": {"title": "some note about music"},
109 },
110 ]
111 ),
112 )
113 with patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=resp):
114 result = await _search_network("music")
115 assert "some note about music" in result
116 assert "saves" not in result
117
118 async def test_empty_results(self):
119 resp = _mock_response(200, _semble_response([]))
120 with patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=resp):
121 result = await _search_network("nonexistent")
122 assert result == "no network results for 'nonexistent'"
123
124 async def test_api_failure(self):
125 resp = _mock_response(500)
126 with patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=resp):
127 result = await _search_network("anything")
128 assert result.startswith("network search failed:")
129
130 async def test_network_error(self):
131 with patch(
132 "httpx.AsyncClient.get",
133 new_callable=AsyncMock,
134 side_effect=httpx.ConnectError("connection refused"),
135 ):
136 result = await _search_network("anything")
137 assert result.startswith("network search failed:")
138
139 async def test_untitled_fallback(self):
140 resp = _mock_response(
141 200,
142 _semble_response(
143 [
144 {"url": "https://example.com", "metadata": {}},
145 ]
146 ),
147 )
148 with patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=resp):
149 result = await _search_network("test")
150 assert "untitled — https://example.com" in result