rsearch/tests/test_providers.py

# retoor <retoor@molodetz.nl>

import asyncio
import pytest
from rsearch.app import MultiSearch


@pytest.fixture
def searcher():
    return MultiSearch()


@pytest.fixture
def query():
    return "python programming"


@pytest.fixture
def count():
    return 3


class TestProviders:

    @pytest.mark.asyncio
    async def test_brave_search(self, searcher, query, count):
        result = await asyncio.wait_for(searcher.brave_search(query, count), timeout=20)
        assert result is not None
        assert result.get("success") is True
        assert result.get("count", 0) > 0
        assert result.get("source") == "brave"
        assert len(result.get("results", [])) > 0
        first = result["results"][0]
        assert "title" in first
        assert "url" in first
        assert "description" in first

    @pytest.mark.asyncio
    async def test_duckduckgo_html_search(self, searcher, query, count):
        result = await asyncio.wait_for(searcher.duckduckgo_html_search(query, count), timeout=20)
        assert result is not None
        assert result.get("success") is True
        assert result.get("count", 0) > 0
        assert result.get("source") == "duckduckgo_html"
        assert len(result.get("results", [])) > 0
        first = result["results"][0]
        assert "title" in first
        assert "url" in first
        assert first["url"].startswith("http")

    @pytest.mark.asyncio
    async def test_bing_search(self, searcher, query, count):
        result = await asyncio.wait_for(searcher.bing_search(query, count), timeout=20)
        assert result is not None
        assert result.get("success") is True
        assert result.get("count", 0) > 0
        assert result.get("source") == "bing"
        assert len(result.get("results", [])) > 0

    @pytest.mark.asyncio
    async def test_mojeek_search(self, searcher, query, count):
        result = await asyncio.wait_for(searcher.mojeek_search(query, count), timeout=20)
        assert result is not None
        assert result.get("success") is True
        assert result.get("count", 0) > 0
        assert result.get("source") == "mojeek"
        assert len(result.get("results", [])) > 0
        first = result["results"][0]
        assert not first["title"].startswith("http"), "Title should not be a URL"

    @pytest.mark.asyncio
    async def test_duckduckgo_api_search(self, searcher, query, count):
        result = await asyncio.wait_for(searcher.duckduckgo_search(query, count), timeout=20)
        if result is not None:
            assert result.get("source") == "duckduckgo"
            if result.get("count", 0) > 0:
                assert result.get("success") is True

    @pytest.mark.asyncio
    async def test_wikipedia_search(self, searcher, query, count):
        result = await asyncio.wait_for(searcher.wikipedia_search(query, count), timeout=20)
        assert result is not None
        assert result.get("success") is True
        assert result.get("count", 0) > 0
        assert result.get("source") == "wikipedia"
        assert len(result.get("results", [])) > 0
        first = result["results"][0]
        assert "wikipedia.org" in first["url"]

    @pytest.mark.asyncio
    async def test_wikidata_search(self, searcher, query, count):
        result = await asyncio.wait_for(searcher.wikidata_search(query, count), timeout=20)
        assert result is not None
        assert result.get("success") is True
        assert result.get("count", 0) > 0
        assert result.get("source") == "wikidata"
        assert len(result.get("results", [])) > 0


class TestSearchAggregator:

    @pytest.mark.asyncio
    async def test_search_returns_results(self, searcher, query, count):
        result = await searcher.search(query, count)
        assert result is not None
        assert result.get("success") is True
        assert result.get("count", 0) > 0
        assert result.get("query") == query
        assert "timestamp" in result
        assert "results" in result

    @pytest.mark.asyncio
    async def test_search_empty_query(self, searcher, count):
        result = await searcher.search("", count)
        assert result is not None
        assert result.get("success") is False
        assert result.get("error") == "Empty query"

    @pytest.mark.asyncio
    async def test_search_count_limit(self, searcher, query):
        result = await searcher.search(query, 5)
        assert result is not None
        if result.get("success"):
            assert result.get("count", 0) <= 5

    @pytest.mark.asyncio
    async def test_result_format(self, searcher, query, count):
        result = await searcher.search(query, count)
        assert "query" in result
        assert "source" in result
        assert "count" in result
        assert "results" in result
        assert "timestamp" in result
        assert "success" in result
        assert "error" in result


class TestAllProviders:

    @pytest.mark.asyncio
    async def test_all_providers_return_valid_format(self, searcher, query, count):
        providers = [
            ("brave", searcher.brave_search),
            ("duckduckgo_html", searcher.duckduckgo_html_search),
            ("bing", searcher.bing_search),
            ("mojeek", searcher.mojeek_search),
            ("duckduckgo", searcher.duckduckgo_search),
            ("wikipedia", searcher.wikipedia_search),
            ("wikidata", searcher.wikidata_search),
        ]

        for name, fn in providers:
            try:
                result = await asyncio.wait_for(fn(query, count), timeout=20)
                if result is not None:
                    assert "query" in result, f"{name}: missing query"
                    assert "source" in result, f"{name}: missing source"
                    assert "count" in result, f"{name}: missing count"
                    assert "results" in result, f"{name}: missing results"
                    assert "timestamp" in result, f"{name}: missing timestamp"
                    assert "success" in result, f"{name}: missing success"
            except asyncio.TimeoutError:
                pytest.skip(f"{name} timed out")


if __name__ == "__main__":
    pytest.main([__file__, "-v"])