Import tests: tests/test_sources

Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.

tests/test_sources.py

@@ -0,0 +1,214 @@
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+from axon.models import SearchResult
+
+
+# ── Brave ──────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_brave_returns_empty_without_api_key(monkeypatch):
+    monkeypatch.setattr("axon.services.sources.brave.settings.brave_api_key", None)
+    from axon.services.sources.brave import search_brave
+    results = await search_brave("test")
+    assert results == []
+
+
+@pytest.mark.asyncio
+async def test_brave_parses_web_results(monkeypatch):
+    import axon.services.sources.brave as brave_mod
+
+    fake_data = {
+        "web": {
+            "results": [
+                {
+                    "title": "Hello",
+                    "url": "https://example.com",
+                    "description": "A description",
+                    "page_age": "2024-01-15",
+                    "meta_url": {"favicon": "https://example.com/fav.ico"},
+                }
+            ]
+        }
+    }
+
+    class FakeResponse:
+        def raise_for_status(self): pass
+        def json(self): return fake_data
+
+    class FakeClient:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *_): pass
+        async def get(self, *args, **kwargs): return FakeResponse()
+
+    monkeypatch.setattr(brave_mod, "settings", type("S", (), {"brave_api_key": "fake-key"})())
+    monkeypatch.setattr("axon.services.sources.brave.httpx.AsyncClient", FakeClient)
+
+    results = await brave_mod.search_brave("hello")
+    assert len(results) == 1
+    assert results[0].url == "https://example.com"
+    assert results[0].title == "Hello"
+    assert results[0].content == "A description"
+    assert results[0].published_date == "2024-01-15"
+    assert results[0].favicon == "https://example.com/fav.ico"
+
+
+@pytest.mark.asyncio
+async def test_brave_returns_empty_on_http_error(monkeypatch):
+    import httpx
+    monkeypatch.setattr("axon.services.sources.brave.settings.brave_api_key", "fake-key")
+
+    class FailClient:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *_): pass
+        async def get(self, *args, **kwargs):
+            raise httpx.ConnectError("network error")
+
+    monkeypatch.setattr("axon.services.sources.brave.httpx.AsyncClient", FailClient)
+
+    from axon.services.sources.brave import search_brave
+    results = await search_brave("test")
+    assert results == []
+
+
+# ── DDG ────────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_ddg_parses_text_results(monkeypatch):
+    fake_items = [
+        {"title": "DDG Result", "href": "https://ddg.com/page", "body": "Some content"},
+    ]
+
+    class FakeDDGS:
+        def __enter__(self): return self
+        def __exit__(self, *_): pass
+        def text(self, *args, **kwargs): return iter(fake_items)
+        def news(self, *args, **kwargs): return iter([])
+
+    monkeypatch.setattr("axon.services.sources.ddg.DDGS", FakeDDGS)
+
+    from axon.services.sources.ddg import search_ddg
+    results = await search_ddg("test")
+    assert len(results) == 1
+    assert results[0].url == "https://ddg.com/page"
+    assert results[0].content == "Some content"
+
+
+@pytest.mark.asyncio
+async def test_ddg_news_uses_url_key(monkeypatch):
+    fake_news = [
+        {"title": "News", "url": "https://news.com/story", "body": "News body", "date": "2024-01-01"},
+    ]
+
+    class FakeDDGS:
+        def __enter__(self): return self
+        def __exit__(self, *_): pass
+        def text(self, *args, **kwargs): return iter([])
+        def news(self, *args, **kwargs): return iter(fake_news)
+
+    monkeypatch.setattr("axon.services.sources.ddg.DDGS", FakeDDGS)
+
+    from axon.services.sources.ddg import search_ddg
+    results = await search_ddg("news query", topic="news")
+    assert len(results) == 1
+    assert results[0].url == "https://news.com/story"
+    assert results[0].content == "News body"
+    assert results[0].published_date == "2024-01-01"
+
+
+@pytest.mark.asyncio
+async def test_ddg_returns_empty_on_error(monkeypatch):
+    class FailDDGS:
+        def __enter__(self): return self
+        def __exit__(self, *_): pass
+        def text(self, *args, **kwargs): raise RuntimeError("DDG down")
+
+    monkeypatch.setattr("axon.services.sources.ddg.DDGS", FailDDGS)
+
+    from axon.services.sources.ddg import search_ddg
+    results = await search_ddg("test")
+    assert results == []
+
+
+@pytest.mark.asyncio
+async def test_ddg_passes_timelimit(monkeypatch):
+    """Verify that time_range parameter is correctly mapped to timelimit for DDGS."""
+    captured_timelimit = {}
+
+    class FakeDDGS:
+        def __enter__(self): return self
+        def __exit__(self, *_): pass
+        def text(self, *args, **kwargs):
+            captured_timelimit["value"] = kwargs.get("timelimit")
+            return iter([])
+        def news(self, *args, **kwargs): return iter([])
+
+    monkeypatch.setattr("axon.services.sources.ddg.DDGS", FakeDDGS)
+
+    from axon.services.sources.ddg import search_ddg
+    await search_ddg("test", time_range="week")
+    assert captured_timelimit.get("value") == "w"
+
+
+# ── Serper ─────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_serper_returns_empty_without_api_key(monkeypatch):
+    monkeypatch.setattr("axon.services.sources.serper.settings.serper_api_key", None)
+    from axon.services.sources.serper import search_serper
+    results = await search_serper("test")
+    assert results == []
+
+
+@pytest.mark.asyncio
+async def test_serper_parses_organic_results(monkeypatch):
+    monkeypatch.setattr("axon.services.sources.serper.settings.serper_api_key", "fake-key")
+
+    fake_data = {
+        "organic": [
+            {"title": "Serper Result", "link": "https://s.com", "snippet": "Snippet text", "date": "Jan 1, 2024"},
+        ]
+    }
+
+    class FakeResponse:
+        def raise_for_status(self): pass
+        def json(self): return fake_data
+
+    class FakeClient:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *_): pass
+        async def post(self, *args, **kwargs): return FakeResponse()
+
+    monkeypatch.setattr("axon.services.sources.serper.httpx.AsyncClient", FakeClient)
+
+    from axon.services.sources.serper import search_serper
+    results = await search_serper("test")
+    assert len(results) == 1
+    assert results[0].url == "https://s.com"
+    assert results[0].content == "Snippet text"
+
+
+@pytest.mark.asyncio
+async def test_serper_uses_news_endpoint(monkeypatch):
+    """Verify that topic='news' uses the /news endpoint."""
+    monkeypatch.setattr("axon.services.sources.serper.settings.serper_api_key", "fake-key")
+
+    captured_url = {}
+    fake_data = {"news": [{"title": "News", "link": "https://n.com", "snippet": "News snippet"}]}
+
+    class FakeResponse:
+        def raise_for_status(self): pass
+        def json(self): return fake_data
+
+    class FakeClient:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *_): pass
+        async def post(self, url, *args, **kwargs):
+            captured_url["value"] = url
+            return FakeResponse()
+
+    monkeypatch.setattr("axon.services.sources.serper.httpx.AsyncClient", FakeClient)
+
+    from axon.services.sources.serper import search_serper
+    results = await search_serper("test", topic="news")
+    assert "/news" in captured_url.get("value", "")
+    assert len(results) == 1