tests/test_exa_search.py

from __future__ import annotations

import asyncio
from types import SimpleNamespace

import httpx

from app.routes import chat
from app.schemas.chat import WebCitation
from app.services.exa_search import (
    ExaSearchUnavailable,
    WebSearchResult,
    WebSearchSource,
    search_exa_web,
)


class _FakeResponse:
    def raise_for_status(self) -> None:
        return None

    def json(self) -> dict[str, object]:
        return {
            "results": [
                {
                    "title": "A recent explainer",
                    "url": "https://example.com/explainer",
                    "publishedDate": "2026-07-12T00:00:00.000Z",
                    "highlights": ["A useful secondary explanation."],
                },
                {
                    "title": "Official examination notice",
                    "url": "https://education.gov.in/exam-notice",
                    "publishedDate": "2026-07-13T00:00:00.000Z",
                    "author": "Ministry of Education",
                    "highlights": ["The official notice and its effective date."],
                },
                {
                    "title": "Unsafe result",
                    "url": "javascript:alert(1)",
                    "highlights": ["This must never reach the client."],
                },
            ]
        }


class _FakeClient:
    last_request: dict[str, object] = {}

    def __init__(self, **kwargs: object) -> None:
        self.kwargs = kwargs

    def __enter__(self) -> "_FakeClient":
        return self

    def __exit__(self, *args: object) -> None:
        return None

    def post(self, path: str, **kwargs: object) -> _FakeResponse:
        _FakeClient.last_request = {"path": path, **kwargs}
        return _FakeResponse()


def _settings(api_key: str | None = "test-server-key") -> SimpleNamespace:
    return SimpleNamespace(
        exa_api_key=api_key,
        exa_base_url="https://api.exa.ai",
        exa_search_timeout_seconds=5,
        exa_search_max_results=6,
    )


def test_exa_normalises_safe_citations_and_ranks_official_first(monkeypatch) -> None:
    monkeypatch.setattr(httpx, "Client", _FakeClient)

    result = search_exa_web("latest official exam notice", settings=_settings())

    assert [source.publisher for source in result.sources] == [
        "education.gov.in",
        "example.com",
    ]
    assert result.sources[0].is_official is True
    assert "Official examination notice" in result.as_prompt_context()
    assert _FakeClient.last_request["path"] == "/search"
    request_json = _FakeClient.last_request["json"]
    assert isinstance(request_json, dict)
    assert request_json["type"] == "auto"
    assert request_json["contents"] == {"highlights": True}


def test_exa_requires_a_server_side_key() -> None:
    try:
        search_exa_web("latest exam notice", settings=_settings(api_key=None))
    except ExaSearchUnavailable as exc:
        assert "not configured" in str(exc)
    else:
        raise AssertionError("Search must stay disabled without a backend key")


def test_only_current_information_questions_receive_web_context(monkeypatch) -> None:
    calls: list[str] = []
    source = WebSearchSource(
        title="Official result",
        url="https://education.gov.in/result",
        publisher="education.gov.in",
        published_date="2026-07-14",
        author=None,
        snippet="The official current result.",
        is_official=True,
    )

    def fake_search(query: str) -> WebSearchResult:
        calls.append(query)
        return WebSearchResult(query=query, sources=(source,))

    monkeypatch.setattr(chat, "search_exa_web", fake_search)

    untouched = asyncio.run(
        chat._prepare_current_info_context(
            intent="study_explain",
            question="Explain sound waves",
            system_prompt="teacher prompt",
            user_message="Explain sound waves",
            evidence_label="No source attached",
        )
    )
    assert untouched == (
        "teacher prompt",
        "Explain sound waves",
        "No source attached",
        [],
    )
    assert calls == []

    prompt, message, label, citations = asyncio.run(
        chat._prepare_current_info_context(
            intent="current_info",
            question="Search the web for the latest official result",
            system_prompt=chat._SYSTEM_PROMPTS["current_info"],
            user_message="Search the web for the latest official result",
            evidence_label="No source attached",
        )
    )
    assert calls == ["Search the web for the latest official result"]
    assert label == "Searched the web"
    assert isinstance(citations[0], WebCitation)
    assert citations[0].publisher == "education.gov.in"
    assert "[1]" in prompt
    assert "education.gov.in/result" in message


def test_chat_intent_router_keeps_web_search_explicit() -> None:
    assert chat._detect_intent("Search the web for today's CBSE notice", False) == "current_info"
    assert chat._detect_intent("What is the latest Kerala SSLC exam timetable?", False) == "current_info"
    assert chat._detect_intent("What is the current Kerala SSLC syllabus?", False) == "current_info"
    assert chat._detect_intent("Explain the current in an electric circuit", False) == "study_explain"
    assert chat._detect_intent("Update my study plan for physics", False) == "study_explain"
    assert chat._detect_intent("What should I study today?", False) == "study_explain"
    assert chat._detect_intent("Help me revise my recent mistakes", False) == "study_explain"
    assert chat._detect_intent("Explain Sound Waves from my textbook", True) == "source_answer"
    # The specialised notes intent is retained, but selected-source retrieval
    # still grounds it; the important regression guard is that "today" does
    # not escape to Exa.
    assert chat._detect_intent("Summarize today's lesson from this PDF", True) == "notes"
    assert chat._detect_intent("Search the web for today's notice", True) == "current_info"


def test_current_information_search_failure_is_visible(monkeypatch) -> None:
    def unavailable(_query: str) -> WebSearchResult:
        raise ExaSearchUnavailable("Web search could not be completed.")

    monkeypatch.setattr(chat, "search_exa_web", unavailable)

    prompt, message, label, citations = asyncio.run(
        chat._prepare_current_info_context(
            intent="current_info",
            question="Search the web for today's official notice",
            system_prompt=chat._SYSTEM_PROMPTS["current_info"],
            user_message="Search the web for today's official notice",
            evidence_label="No source attached",
        )
    )

    assert prompt == chat._SYSTEM_PROMPTS["current_info"]
    assert message == "Search the web for today's official notice"
    assert label == "Web search unavailable"
    assert citations == []