From eddf38104c5be99c0301b4e73d257a25e0e7a9a5 Mon Sep 17 00:00:00 2001 From: arsam Date: Fri, 18 Sep 2026 23:06:53 +0330 Subject: [PATCH] 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 | 214 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 214 insertions(+) create mode 100644 tests/test_sources.py diff --git a/tests/test_sources.py b/tests/test_sources.py new file mode 100644 index 0000000..2d718cf --- /dev/null +++ b/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