leocrm/tests/test_unified_search.py

"""Tests for the Unified Search plugin — covers ACs 1-12 and unit tests.

Tests hybrid search infrastructure: API endpoints, RRF fusion, autocomplete,
text extraction, and provider registry.
"""

from __future__ import annotations

import json
import uuid
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker

from app.core.db import close_engine, reset_engine_for_testing
from app.core.service_container import get_container
from app.main import create_app
from app.plugins.builtins.unified_search import UnifiedSearchPlugin
from app.plugins.builtins.unified_search.provider_registry import (
    SearchProviderRegistry,
    get_search_registry,
)
from app.plugins.builtins.unified_search.search_engine import rrf_fusion
from app.plugins.builtins.unified_search.text_extraction import extract_text_from_file
from app.plugins.registry import reset_registry_for_testing
from app.services.plugin_service import reset_plugin_service_for_testing
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users

from app.plugins.builtins.unified_search.embedding import (
    generate_embedding,
    generate_embeddings_batch,
    index_entity,
)
from app.plugins.builtins.unified_search.query_understanding import (
    llm_analyze_query,
    llm_aggregate_results,
    _fallback_query_analysis,
    _fallback_aggregate,
)
from app.plugins.builtins.unified_search.search_engine import (
    hybrid_search,
    find_similar_all_types,
    autocomplete,
)
from app.plugins.builtins.unified_search.jobs import (
    index_mails,
    index_contact,
    index_company,
    index_event,
    reindex,
    embedding_batch,
)
from app.plugins.builtins.unified_search.providers.contact_provider import (
    ContactSearchProvider,
)
from app.plugins.builtins.unified_search.providers.company_provider import (
    CompanySearchProvider,
)
from app.plugins.builtins.unified_search.providers.mail_provider import (
    MailSearchProvider,
)
from app.plugins.builtins.unified_search.providers.file_provider import (
    FileSearchProvider,
)
from app.plugins.builtins.unified_search.providers.event_provider import (
    EventSearchProvider,
)

# ─── Unified Search Fixtures ───


@pytest_asyncio.fixture
async def search_app(engine: AsyncEngine, redis_client):
    """FastAPI app with UnifiedSearch plugin registered, installed, and activated."""
    reset_engine_for_testing(engine)
    app = create_app()
    registry = reset_registry_for_testing()
    registry.initialize(engine, app)
    container = get_container()
    await container.initialize()
    registry.register_plugin(UnifiedSearchPlugin())
    reset_plugin_service_for_testing(registry)

    sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
    async with sf() as session:
        await registry.install(session, "unified_search")
        await registry.activate(session, "unified_search")
        await session.commit()

    yield app
    await close_engine()


@pytest_asyncio.fixture
async def search_client(search_app) -> AsyncClient:
    """HTTP test client with unified search plugin active."""
    transport = ASGITransport(app=search_app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


@pytest_asyncio.fixture
async def search_authed_client(
    search_client: AsyncClient, db_session: AsyncSession
) -> tuple[AsyncClient, dict]:
    """Authenticated admin client with seeded data and unified search plugin active."""
    seed = await seed_tenant_and_users(db_session)
    # Grant is_system_admin to admin user so search:read/search:admin permissions pass
    from sqlalchemy import update
    from app.models.user import User
    await db_session.execute(
        update(User)
        .where(User.email == "admin@tenanta.com")
        .values(is_system_admin=True)
    )
    await db_session.commit()
    # Login and capture CSRF token
    login_resp = await search_client.post(
        "/api/v1/auth/login",
        json={"email": "admin@tenanta.com", "password": "TestPass123!"},
        headers=ORIGIN_HEADER,
    )
    assert login_resp.status_code == 200, f"Login failed: {login_resp.text}"
    csrf_token = login_resp.json().get("csrf_token", "")
    search_client.headers["X-CSRF-Token"] = csrf_token
    return search_client, seed


@pytest.fixture(autouse=True)
async def mock_external_calls():
    """Mock all external API calls (LiteLLM, job queue) for all tests."""
    mock_resp = MagicMock()
    mock_resp.choices = [MagicMock()]
    mock_resp.choices[0].message.content = json.dumps({
        "normalized_query": "test",
        "entities": {},
        "intent": "search",
        "semantic_terms": [],
        "suggested_filters": {},
    })

    mock_emb_resp = MagicMock()
    mock_emb_resp.data = [{"embedding": [0.1] * 768}]

    with (
        patch("litellm.acompletion", new_callable=AsyncMock, return_value=mock_resp),
        patch("litellm.aembedding", new_callable=AsyncMock, return_value=mock_emb_resp),
        patch("app.core.jobs.enqueue_job", new_callable=AsyncMock, return_value="job-123"),
    ):
        yield


# ─── AC1: POST /api/v1/search → 200 + results array ───


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.routes.hybrid_search", new_callable=AsyncMock)
async def test_ac1_search_returns_results(
    mock_hybrid_search,
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC1: POST /api/v1/search with query returns 200 + results array."""
    client, _ = search_authed_client
    mock_hybrid_search.return_value = [
        {
            "entity_type": "company",
            "entity_id": str(uuid.uuid4()),
            "title": "Company Alpha",
            "snippet": "IT company",
            "score": 0.95,
            "data": {"industry": "IT"},
        }
    ]

    resp = await client.post(
        "/api/v1/search",
        json={"query": "Company"},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert "results" in data
    assert isinstance(data["results"], list)
    assert len(data["results"]) > 0
    assert data["results"][0]["entity_type"] == "company"
    assert data["results"][0]["title"] == "Company Alpha"
    assert "score" in data["results"][0]
    assert "query" in data
    assert "normalized_query" in data
    assert "summary" in data


# ─── AC2: entity_types filter ───


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.routes.hybrid_search", new_callable=AsyncMock)
async def test_ac2_search_entity_types_filter(
    mock_hybrid_search,
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC2: Search with entity_types filters correctly."""
    client, _ = search_authed_client
    mock_hybrid_search.return_value = []

    resp = await client.post(
        "/api/v1/search",
        json={"query": "test", "entity_types": ["company"]},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    # Verify hybrid_search was called with entity_types=["company"]
    call_kwargs = mock_hybrid_search.call_args
    assert call_kwargs.kwargs.get("entity_types") == ["company"]


# ─── AC3: No auth → 401 ───


@pytest.mark.asyncio
async def test_ac3_search_no_auth_401(search_client: AsyncClient):
    """AC3: Search without auth returns 401 or 403 (CSRF may block first)."""
    resp = await search_client.post(
        "/api/v1/search",
        json={"query": "test"},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code in (401, 403)


# ─── AC4: Tenant isolation ───


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.routes.hybrid_search", new_callable=AsyncMock)
async def test_ac4_search_tenant_isolation(
    mock_hybrid_search,
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC4: Search only returns results from the user's tenant."""
    client, seed = search_authed_client
    mock_hybrid_search.return_value = []

    resp = await client.post(
        "/api/v1/search",
        json={"query": "test"},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200

    # Verify hybrid_search was called with tenant A's ID
    call_args = mock_hybrid_search.call_args
    tenant_id_arg = call_args.kwargs.get("tenant_id")
    assert tenant_id_arg == seed["tenant_a"].id


# ─── AC5: Empty query → 422 ───


@pytest.mark.asyncio
async def test_ac5_search_empty_query_422(search_authed_client: tuple[AsyncClient, dict]):
    """AC5: Empty query returns 422 (min_length=1)."""
    client, _ = search_authed_client
    resp = await client.post(
        "/api/v1/search",
        json={"query": ""},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 422


# ─── AC6: GET /api/v1/search/suggest → suggestions ───


@pytest.mark.asyncio
async def test_ac6_suggest_returns_suggestions(
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC6: GET /api/v1/search/suggest returns 200 + suggestions list."""
    client, _ = search_authed_client
    resp = await client.get(
        "/api/v1/search/suggest",
        params={"q": "comp"},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert "suggestions" in data
    assert isinstance(data["suggestions"], list)


# ─── AC7: Empty suggest query → 422 ───


@pytest.mark.asyncio
async def test_ac7_suggest_empty_query_422(
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC7: GET /api/v1/search/suggest with empty query returns 422 (min_length=1)."""
    client, _ = search_authed_client
    resp = await client.get(
        "/api/v1/search/suggest",
        params={"q": ""},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 422


# ─── AC8: POST /api/v1/search/similar → similar results ───


@pytest.mark.asyncio
async def test_ac8_similar_returns_results(
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC8: POST /api/v1/search/similar returns 200 + similar dict."""
    client, seed = search_authed_client
    resp = await client.post(
        "/api/v1/search/similar",
        json={
            "entity_type": "company",
            "entity_id": str(seed["company_a"].id),
            "limit": 5,
        },
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert "similar" in data
    assert isinstance(data["similar"], dict)


# ─── AC9: Similar tenant isolation ───


@pytest.mark.asyncio
async def test_ac9_similar_tenant_isolation(
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC9: Similar search respects tenant isolation."""
    client, seed = search_authed_client
    # Use company from tenant B
    resp = await client.post(
        "/api/v1/search/similar",
        json={
            "entity_type": "company",
            "entity_id": str(seed["company_b"].id),
            "limit": 5,
        },
        headers=ORIGIN_HEADER,
    )
    # Should return 200 but with empty similar (tenant B company not accessible from tenant A)
    assert resp.status_code == 200
    data = resp.json()
    assert isinstance(data["similar"], dict)
    # Company B should not be found in tenant A's context
    for etype, items in data["similar"].items():
        for item in items:
            assert item.get("entity_id") != str(seed["company_b"].id)


# ─── AC10: GET /api/v1/search/providers → provider list ───


@pytest.mark.asyncio
async def test_ac10_list_providers(
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC10: GET /api/v1/search/providers returns 200 + active provider list."""
    client, _ = search_authed_client
    resp = await client.get(
        "/api/v1/search/providers",
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert isinstance(data, list)
    # After activation, providers should be registered
    if len(data) > 0:
        assert "entity_type" in data[0]
        assert "plugin_name" in data[0]
        assert "is_active" in data[0]
        assert data[0]["is_active"] is True


# ─── AC11: POST /api/v1/search/reindex as admin → 200 ───


@pytest.mark.asyncio
async def test_ac11_reindex_admin_200(
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC11: Admin can trigger reindex (200)."""
    client, _ = search_authed_client
    resp = await client.post(
        "/api/v1/search/reindex",
        json={"entity_types": ["company"]},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["status"] == "ok"
    assert "entity_types" in data


# ─── AC12: POST /api/v1/search/reindex as viewer → 403 ───


@pytest.mark.asyncio
async def test_ac12_reindex_viewer_403(
    search_client: AsyncClient, db_session: AsyncSession
):
    """AC12: Non-admin (viewer) gets 403 on reindex."""
    await seed_tenant_and_users(db_session)
    await login_client(search_client, "viewer@tenanta.com")

    resp = await search_client.post(
        "/api/v1/search/reindex",
        json={"entity_types": ["company"]},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 403


# ─── Unit Tests: RRF Fusion ───


def test_rrf_fusion_combines_fts_and_vector_scores():
    """RRF fusion combines FTS + Vector scores correctly."""
    fts_results = [
        {"id": "1", "title": "Result A", "score": 0.9},
        {"id": "2", "title": "Result B", "score": 0.8},
    ]
    vec_results = [
        {"id": "2", "title": "Result B", "score": 0.95},
        {"id": "3", "title": "Result C", "score": 0.7},
    ]

    fused = rrf_fusion(fts_results, vec_results, "company")

    # All 3 unique IDs should be present
    ids = [str(r.get("id", "")) for r in fused]
    assert "1" in ids
    assert "2" in ids
    assert "3" in ids

    # Results should be sorted by fused score descending
    assert fused[0]["_score"] >= fused[-1]["_score"]

    # ID 2 appears in both FTS and vector, so should have highest score
    assert str(fused[0].get("id", "")) == "2"


def test_rrf_fusion_empty_inputs():
    """RRF fusion with empty inputs returns empty list."""
    fused = rrf_fusion([], [], "company")
    assert fused == []


def test_rrf_fusion_fts_only():
    """RRF fusion with only FTS results returns them sorted by score."""
    fts_results = [
        {"id": "1", "title": "A"},
        {"id": "2", "title": "B"},
    ]
    fused = rrf_fusion(fts_results, [], "company")
    assert len(fused) == 2
    # First result (rank 0) should have higher score
    assert fused[0]["_score"] > fused[1]["_score"]


# ─── Unit Tests: Text Extraction ───


@pytest.mark.asyncio
async def test_extract_text_from_text_file(tmp_path):
    """Extract text from a plain text file."""
    test_file = tmp_path / "test.txt"
    test_file.write_text("Hello World\nThis is a test.", encoding="utf-8")

    result = await extract_text_from_file(str(test_file), "text/plain")
    assert "Hello World" in result
    assert "test" in result


@pytest.mark.asyncio
async def test_extract_text_from_pdf_mocked(tmp_path):
    """Extract text from PDF with mocked PyMuPDF."""
    test_file = tmp_path / "test.pdf"
    test_file.write_bytes(b"%PDF-1.4 fake")

    mock_page = MagicMock()
    mock_page.get_text.return_value = "PDF content text"
    mock_doc = MagicMock()
    mock_doc.__iter__ = MagicMock(return_value=iter([mock_page]))
    mock_doc.close = MagicMock()

    with patch("fitz.open", return_value=mock_doc):
        result = await extract_text_from_file(str(test_file), "application/pdf")
    assert "PDF content text" in result


@pytest.mark.asyncio
async def test_extract_text_unsupported_mime():
    """Unsupported MIME type returns empty string."""
    result = await extract_text_from_file("/tmp/nonexistent", "application/octet-stream")
    assert result == ""


@pytest.mark.asyncio
async def test_extract_text_image_returns_empty(tmp_path):
    """Image MIME type returns empty string (OCR not implemented)."""
    test_file = tmp_path / "image.png"
    test_file.write_bytes(b"\x89PNG\r\n\x1a\n")
    result = await extract_text_from_file(str(test_file), "image/png")
    assert result == ""


# ─── Unit Tests: Provider Registry ───


def test_provider_register_and_unregister():
    """Provider can be registered and unregistered."""
    registry = SearchProviderRegistry()

    # Create a mock provider
    mock_provider = MagicMock()
    mock_provider.entity_type = "test_entity"

    # Register
    registry.register(mock_provider)
    assert registry.get("test_entity") is mock_provider

    # Unregister
    registry.unregister("test_entity")
    assert registry.get("test_entity") is None


def test_provider_get_all():
    """get_all returns all registered providers."""
    registry = SearchProviderRegistry()

    p1 = MagicMock()
    p1.entity_type = "contact"
    p2 = MagicMock()
    p2.entity_type = "company"

    registry.register(p1)
    registry.register(p2)

    all_providers = registry.get_all()
    assert len(all_providers) == 2
    entity_types = [p.entity_type for p in all_providers]
    assert "contact" in entity_types
    assert "company" in entity_types


def test_provider_clear():
    """clear removes all providers."""
    registry = SearchProviderRegistry()
    p = MagicMock()
    p.entity_type = "test"
    registry.register(p)
    assert len(registry.get_all()) == 1

    registry.clear()
    assert len(registry.get_all()) == 0


def test_provider_get_entity_types():
    """get_entity_types returns list of registered entity types."""
    registry = SearchProviderRegistry()
    p1 = MagicMock()
    p1.entity_type = "contact"
    p2 = MagicMock()
    p2.entity_type = "company"
    registry.register(p1)
    registry.register(p2)

    types = registry.get_entity_types()
    assert "contact" in types
    assert "company" in types
    assert len(types) == 2


# ─── Unit Tests: Autocomplete ───


@pytest.mark.asyncio
async def test_autocomplete_empty_query_returns_empty(db_session: AsyncSession):
    """Autocomplete with empty query returns empty list."""
    from app.plugins.builtins.unified_search.search_engine import autocomplete

    result = await autocomplete(db_session, "", uuid.uuid4(), 10)
    assert result == []


@pytest.mark.asyncio
async def test_autocomplete_returns_titles(db_session: AsyncSession):
    """Autocomplete returns titles from FTS prefix search."""
    from app.plugins.builtins.unified_search.search_engine import autocomplete

    # With no providers registered and no data, should return empty list
    result = await autocomplete(db_session, "test", uuid.uuid4(), 10)
    assert isinstance(result, list)
    assert len(result) <= 10


# ─── Extended: Embedding Pipeline ───


@pytest.mark.asyncio
async def test_generate_embedding_success():
    """generate_embedding returns a vector list."""
    result = await generate_embedding("hello world")
    assert isinstance(result, list)
    assert len(result) > 0
    assert all(isinstance(v, float) for v in result)


@pytest.mark.asyncio
async def test_generate_embedding_empty_text():
    """generate_embedding with empty text still returns a list (mocked)."""
    result = await generate_embedding("")
    assert isinstance(result, list)


@pytest.mark.asyncio
async def test_generate_embedding_api_failure():
    """generate_embedding returns [] on API failure."""
    with patch("litellm.aembedding", new_callable=AsyncMock, side_effect=Exception("API error")):
        result = await generate_embedding("test")
    assert result == []


@pytest.mark.asyncio
async def test_generate_embeddings_batch_success():
    """generate_embeddings_batch returns a list of vectors."""
    mock_batch_resp = MagicMock()
    mock_batch_resp.data = [
        {"embedding": [0.1] * 768},
        {"embedding": [0.2] * 768},
    ]
    with patch("litellm.aembedding", new_callable=AsyncMock, return_value=mock_batch_resp):
        result = await generate_embeddings_batch(["hello", "world"])
    assert isinstance(result, list)
    assert len(result) == 2
    assert all(isinstance(v, list) for v in result)


@pytest.mark.asyncio
async def test_generate_embeddings_batch_empty():
    """generate_embeddings_batch with empty list returns []."""
    mock_empty_resp = MagicMock()
    mock_empty_resp.data = []
    with patch("litellm.aembedding", new_callable=AsyncMock, return_value=mock_empty_resp):
        result = await generate_embeddings_batch([])
    assert isinstance(result, list)
    assert len(result) == 0


@pytest.mark.asyncio
async def test_index_entity_success(db_session: AsyncSession):
    """index_entity stores embedding in DB and returns True."""
    from app.models.contact import Contact
    from app.models.tenant import Tenant
    from app.models.user import User
    from app.core.auth import hash_password

    tenant = Tenant(name="Test Tenant", slug="test-tenant")
    db_session.add(tenant)
    await db_session.flush()
    user = User(
        tenant_id=tenant.id,
        email="test@example.com",
        name="Test",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        is_active=True,
        preferences={},
    )
    db_session.add(user)
    await db_session.flush()
    contact = Contact(
        tenant_id=tenant.id,
        first_name="John",
        last_name="Doe",
        email="john@example.com",
        created_by=user.id,
        updated_by=user.id,
    )
    db_session.add(contact)
    await db_session.commit()

    # Register provider so index_entity can find it
    registry = get_search_registry()
    registry.clear()
    registry.register(ContactSearchProvider())

    result = await index_entity("contact", contact.id, tenant.id, db_session)
    assert result is True
    registry.clear()


@pytest.mark.asyncio
async def test_index_entity_no_provider(db_session: AsyncSession):
    """index_entity returns False for unknown entity_type."""
    registry = get_search_registry()
    registry.clear()
    result = await index_entity("unknown_type", uuid.uuid4(), uuid.uuid4(), db_session)
    assert result is False


# ─── Extended: Query Understanding ───


@pytest.mark.asyncio
async def test_llm_analyze_query_success():
    """llm_analyze_query returns analyzed query dict."""
    result = await llm_analyze_query("Finde alle Kontakte bei Company Alpha")
    assert isinstance(result, dict)
    assert "normalized_query" in result
    assert "entities" in result
    assert "intent" in result
    assert "semantic_terms" in result
    assert "suggested_filters" in result


@pytest.mark.asyncio
async def test_llm_analyze_query_fallback():
    """llm_analyze_query returns fallback on LLM failure."""
    with patch("litellm.acompletion", new_callable=AsyncMock, side_effect=Exception("LLM error")):
        result = await llm_analyze_query("test query")
    assert result == _fallback_query_analysis("test query")


@pytest.mark.asyncio
async def test_llm_aggregate_results_success():
    """llm_aggregate_results returns summary + facets."""
    mock_agg_resp = MagicMock()
    mock_agg_resp.choices = [MagicMock()]
    mock_agg_resp.choices[0].message.content = json.dumps({
        "summary": "2 Ergebnisse gefunden",
        "facets": {"types": {"company": 1, "contact": 1}},
        "suggestions": ["Filter by company"],
    })
    with patch("litellm.acompletion", new_callable=AsyncMock, return_value=mock_agg_resp):
        results = [
            {"entity_type": "company", "title": "Company Alpha"},
            {"entity_type": "contact", "title": "John Doe"},
        ]
        result = await llm_aggregate_results(results, "test")
    assert isinstance(result, dict)
    assert "summary" in result
    assert "facets" in result
    assert "suggestions" in result
    assert result["summary"] == "2 Ergebnisse gefunden"


@pytest.mark.asyncio
async def test_llm_aggregate_results_empty():
    """llm_aggregate_results with empty results returns fallback."""
    result = await llm_aggregate_results([], "test")
    assert result == _fallback_aggregate([], "test")


def test_fallback_query_analysis():
    """Fallback query analysis has correct structure."""
    result = _fallback_query_analysis("my query")
    assert result["normalized_query"] == "my query"
    assert result["entities"] == {}
    assert result["intent"] == "search"
    assert result["semantic_terms"] == []
    assert result["suggested_filters"] == {}


def test_fallback_aggregate():
    """Fallback aggregate has correct structure."""
    result = _fallback_aggregate([{"title": "x"}], "test")
    assert "summary" in result
    assert "facets" in result
    assert "suggestions" in result
    assert "1 Ergebnisse gefunden" in result["summary"]


# ─── Extended: Search Engine ───


@pytest.mark.asyncio
async def test_hybrid_search_with_results(db_session: AsyncSession):
    """hybrid_search returns results when data exists."""
    from app.models.contact import Contact
    from app.models.tenant import Tenant
    from app.models.user import User
    from app.core.auth import hash_password

    tenant = Tenant(name="Test HS", slug="test-hs")
    db_session.add(tenant)
    await db_session.flush()
    user = User(
        tenant_id=tenant.id,
        email="hs@example.com",
        name="HS",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        is_active=True,
        preferences={},
    )
    db_session.add(user)
    await db_session.flush()
    contact = Contact(
        tenant_id=tenant.id,
        first_name="Search",
        last_name="Test",
        email="searchtest@example.com",
        created_by=user.id,
        updated_by=user.id,
    )
    db_session.add(contact)
    await db_session.commit()

    registry = get_search_registry()
    registry.clear()
    registry.register(ContactSearchProvider())

    query_analysis = {"normalized_query": "search", "semantic_terms": []}
    results = await hybrid_search(db_session, query_analysis, tenant.id, limit=10)
    assert isinstance(results, list)
    registry.clear()


@pytest.mark.asyncio
async def test_hybrid_search_no_results(db_session: AsyncSession):
    """hybrid_search returns empty list on empty DB."""
    registry = get_search_registry()
    registry.clear()
    registry.register(ContactSearchProvider())
    query_analysis = {"normalized_query": "nonexistent", "semantic_terms": []}
    results = await hybrid_search(db_session, query_analysis, uuid.uuid4(), limit=10)
    assert isinstance(results, list)
    assert len(results) == 0
    registry.clear()


@pytest.mark.asyncio
async def test_find_similar_all_types(db_session: AsyncSession):
    """find_similar_all_types returns similar entities dict."""
    result = await find_similar_all_types(db_session, "contact", uuid.uuid4(), uuid.uuid4(), limit=5)
    assert isinstance(result, dict)


@pytest.mark.asyncio
async def test_find_similar_no_embedding(db_session: AsyncSession):
    """find_similar_all_types returns {} when entity has no embedding."""
    result = await find_similar_all_types(db_session, "contact", uuid.uuid4(), uuid.uuid4(), limit=5)
    assert result == {}


# ─── Extended: Jobs ───


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.jobs.get_session_factory")
@patch("app.plugins.builtins.unified_search.embedding.index_entity", new_callable=AsyncMock, return_value=True)
async def test_index_mails(mock_index_entity, mock_factory, db_session: AsyncSession):
    """index_mails calls index_entity for each mail."""
    from app.plugins.builtins.mail.models import Mail, MailAccount, MailFolder
    from app.models.tenant import Tenant
    from app.models.user import User
    from app.core.auth import hash_password

    tenant = Tenant(name="Mail Tenant", slug="mail-tenant")
    db_session.add(tenant)
    await db_session.flush()
    user = User(
        tenant_id=tenant.id,
        email="mail@example.com",
        name="Mail",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        is_active=True,
        preferences={},
    )
    db_session.add(user)
    await db_session.flush()
    account = MailAccount(
        tenant_id=tenant.id,
        user_id=user.id,
        email_address="mail@example.com",
        display_name="Mail Account",
        imap_host="localhost",
        imap_port=993,
        imap_ssl=True,
        smtp_host="localhost",
        smtp_port=587,
        smtp_tls=True,
        username="mail@example.com",
        encrypted_password="encrypted",
        is_shared=False,
        is_active=True,
    )
    db_session.add(account)
    await db_session.flush()
    folder = MailFolder(
        tenant_id=tenant.id,
        account_id=account.id,
        name="INBOX",
        imap_name="INBOX",
        is_standard=True,
        unread_count=0,
        total_count=0,
    )
    db_session.add(folder)
    await db_session.flush()
    mail = Mail(
        tenant_id=tenant.id,
        account_id=account.id,
        folder_id=folder.id,
        message_id="msg-1",
        thread_id="thread-1",
        from_address="sender@example.com",
        to_addresses="recipient@example.com",
        cc_addresses="",
        bcc_addresses="",
        subject="Test Mail",
        body_text="Test body",
        body_html="",
        body_html_sanitized="",
        is_seen=False,
        is_flagged=False,
        is_draft=False,
        is_answered=False,
        is_forwarded=False,
        has_attachments=False,
        size_bytes=0,
        received_at=datetime.now(UTC),
    )
    db_session.add(mail)
    await db_session.commit()

    # Mock factory to return a session that uses the test engine
    sf = async_sessionmaker(bind=db_session.bind, expire_on_commit=False, class_=AsyncSession)
    mock_factory.return_value = sf

    await index_mails({}, [str(mail.id)])
    mock_index_entity.assert_called()


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.jobs.get_session_factory")
@patch("app.plugins.builtins.unified_search.embedding.index_entity", new_callable=AsyncMock, return_value=True)
async def test_index_contact(mock_index_entity, mock_factory, db_session: AsyncSession):
    """index_contact calls index_entity."""
    from app.models.contact import Contact
    from app.models.tenant import Tenant
    from app.models.user import User
    from app.core.auth import hash_password

    tenant = Tenant(name="Contact Tenant", slug="contact-tenant")
    db_session.add(tenant)
    await db_session.flush()
    user = User(
        tenant_id=tenant.id,
        email="contact@example.com",
        name="Contact",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        is_active=True,
        preferences={},
    )
    db_session.add(user)
    await db_session.flush()
    contact = Contact(
        tenant_id=tenant.id,
        first_name="Index",
        last_name="Contact",
        email="indexcontact@example.com",
        created_by=user.id,
        updated_by=user.id,
    )
    db_session.add(contact)
    await db_session.commit()

    sf = async_sessionmaker(bind=db_session.bind, expire_on_commit=False, class_=AsyncSession)
    mock_factory.return_value = sf

    await index_contact({}, str(contact.id))
    mock_index_entity.assert_called()


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.jobs.get_session_factory")
@patch("app.plugins.builtins.unified_search.embedding.index_entity", new_callable=AsyncMock, return_value=True)
async def test_index_company(mock_index_entity, mock_factory, db_session: AsyncSession):
    """index_company calls index_entity."""
    from app.models.company import Company
    from app.models.tenant import Tenant
    from app.models.user import User
    from app.core.auth import hash_password

    tenant = Tenant(name="Company Tenant", slug="company-tenant")
    db_session.add(tenant)
    await db_session.flush()
    user = User(
        tenant_id=tenant.id,
        email="company@example.com",
        name="Company",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        is_active=True,
        preferences={},
    )
    db_session.add(user)
    await db_session.flush()
    company = Company(
        tenant_id=tenant.id,
        name="Index Company",
        industry="IT",
        created_by=user.id,
        updated_by=user.id,
    )
    db_session.add(company)
    await db_session.commit()

    sf = async_sessionmaker(bind=db_session.bind, expire_on_commit=False, class_=AsyncSession)
    mock_factory.return_value = sf

    await index_company({}, str(company.id))
    mock_index_entity.assert_called()


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.jobs.get_session_factory")
@patch("app.plugins.builtins.unified_search.embedding.index_entity", new_callable=AsyncMock, return_value=True)
async def test_reindex(mock_index_entity, mock_factory, db_session: AsyncSession):
    """reindex iterates over all entities of a type."""
    from app.models.company import Company
    from app.models.tenant import Tenant
    from app.models.user import User
    from app.core.auth import hash_password

    tenant = Tenant(name="Reindex Tenant", slug="reindex-tenant")
    db_session.add(tenant)
    await db_session.flush()
    user = User(
        tenant_id=tenant.id,
        email="reindex@example.com",
        name="Reindex",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        is_active=True,
        preferences={},
    )
    db_session.add(user)
    await db_session.flush()
    company = Company(
        tenant_id=tenant.id,
        name="Reindex Co",
        industry="IT",
        created_by=user.id,
        updated_by=user.id,
    )
    db_session.add(company)
    await db_session.commit()

    sf = async_sessionmaker(bind=db_session.bind, expire_on_commit=False, class_=AsyncSession)
    mock_factory.return_value = sf

    await reindex({}, "company")
    mock_index_entity.assert_called()


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.jobs.get_session_factory")
@patch("app.plugins.builtins.unified_search.embedding.index_entity", new_callable=AsyncMock, return_value=True)
async def test_embedding_batch(mock_index_entity, mock_factory, db_session: AsyncSession):
    """embedding_batch finds entities without embedding."""
    from app.models.company import Company
    from app.models.tenant import Tenant
    from app.models.user import User
    from app.core.auth import hash_password

    tenant = Tenant(name="Batch Tenant", slug="batch-tenant")
    db_session.add(tenant)
    await db_session.flush()
    user = User(
        tenant_id=tenant.id,
        email="batch@example.com",
        name="Batch",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        is_active=True,
        preferences={},
    )
    db_session.add(user)
    await db_session.flush()
    company = Company(
        tenant_id=tenant.id,
        name="Batch Co",
        industry="IT",
        created_by=user.id,
        updated_by=user.id,
    )
    db_session.add(company)
    await db_session.commit()

    sf = async_sessionmaker(bind=db_session.bind, expire_on_commit=False, class_=AsyncSession)
    mock_factory.return_value = sf

    await embedding_batch({})
    mock_index_entity.assert_called()


# ─── Extended: Provider-Specific Tests ───


@pytest.mark.asyncio
async def test_contact_provider_search_fts(db_session: AsyncSession):
    """ContactProvider FTS search returns results."""
    provider = ContactSearchProvider()
    result = await provider.search_fts(db_session, "test", uuid.uuid4(), 10)
    assert isinstance(result, list)


@pytest.mark.asyncio
async def test_company_provider_search_fts(db_session: AsyncSession):
    """CompanyProvider FTS search returns results."""
    provider = CompanySearchProvider()
    result = await provider.search_fts(db_session, "test", uuid.uuid4(), 10)
    assert isinstance(result, list)


@pytest.mark.asyncio
async def test_mail_provider_search_fts(db_session: AsyncSession):
    """MailProvider FTS search returns results."""
    provider = MailSearchProvider()
    result = await provider.search_fts(db_session, "test", uuid.uuid4(), 10)
    assert isinstance(result, list)


@pytest.mark.asyncio
async def test_file_provider_search_fts(db_session: AsyncSession):
    """FileProvider FTS search returns results."""
    provider = FileSearchProvider()
    result = await provider.search_fts(db_session, "test", uuid.uuid4(), 10)
    assert isinstance(result, list)


@pytest.mark.asyncio
async def test_event_provider_search_fts(db_session: AsyncSession):
    """EventProvider FTS search returns results."""
    provider = EventSearchProvider()
    result = await provider.search_fts(db_session, "test", uuid.uuid4(), 10)
    assert isinstance(result, list)


def test_contact_provider_get_embedding_text():
    """ContactProvider get_embedding_text is a callable method."""
    provider = ContactSearchProvider()
    assert hasattr(provider, "get_embedding_text")
    assert provider.entity_type == "contact"


def test_company_provider_get_embedding_text():
    """CompanyProvider get_embedding_text is a callable method."""
    provider = CompanySearchProvider()
    assert hasattr(provider, "get_embedding_text")
    assert provider.entity_type == "company"


def test_mail_provider_get_embedding_text():
    """MailProvider get_embedding_text is a callable method."""
    provider = MailSearchProvider()
    assert hasattr(provider, "get_embedding_text")
    assert provider.entity_type == "mail"


def test_file_provider_get_embedding_text():
    """FileProvider get_embedding_text is a callable method."""
    provider = FileSearchProvider()
    assert hasattr(provider, "get_embedding_text")
    assert provider.entity_type == "file"


def test_event_provider_get_embedding_text():
    """EventProvider get_embedding_text is a callable method."""
    provider = EventSearchProvider()
    assert hasattr(provider, "get_embedding_text")
    assert provider.entity_type == "event"


def test_contact_provider_to_search_result():
    """ContactProvider to_search_result returns correct dict."""
    provider = ContactSearchProvider()
    entity = {"id": "123", "first_name": "John", "last_name": "Doe", "email": "john@example.com"}
    result = provider.to_search_result(entity)
    assert result["entity_type"] == "contact"
    assert result["entity_id"] == "123"
    assert result["title"] == "John Doe"
    assert result["snippet"] == "john@example.com"
    assert result["score"] == 0.0


def test_company_provider_to_search_result():
    """CompanyProvider to_search_result returns correct dict."""
    provider = CompanySearchProvider()
    entity = {"id": "456", "name": "Acme Corp", "description": "IT company"}
    result = provider.to_search_result(entity)
    assert result["entity_type"] == "company"
    assert result["entity_id"] == "456"
    assert result["title"] == "Acme Corp"
    assert "IT company" in result["snippet"]


def test_mail_provider_to_search_result():
    """MailProvider to_search_result returns correct dict."""
    provider = MailSearchProvider()
    entity = {"id": "789", "subject": "Test Subject", "body_text": "Body text here"}
    result = provider.to_search_result(entity)
    assert result["entity_type"] == "mail"
    assert result["entity_id"] == "789"
    assert result["title"] == "Test Subject"
    assert "Body text" in result["snippet"]


def test_file_provider_to_search_result():
    """FileProvider to_search_result returns correct dict."""
    provider = FileSearchProvider()
    entity = {"id": "abc", "name": "document.pdf", "content_text": "File content"}
    result = provider.to_search_result(entity)
    assert result["entity_type"] == "file"
    assert result["entity_id"] == "abc"
    assert result["title"] == "document.pdf"
    assert "File content" in result["snippet"]


def test_event_provider_to_search_result():
    """EventProvider to_search_result returns correct dict."""
    provider = EventSearchProvider()
    entity = {"id": "evt1", "title": "Meeting", "description": "Team meeting"}
    result = provider.to_search_result(entity)
    assert result["entity_type"] == "event"
    assert result["entity_id"] == "evt1"
    assert result["title"] == "Meeting"
    assert "Team meeting" in result["snippet"]


# ─── Extended: API Tests ───


@pytest.mark.asyncio
async def test_ac13_stats_endpoint(
    search_authed_client: tuple[AsyncClient, dict],
):
    """AC13: GET /api/v1/search/stats returns 200 + statistics."""
    client, _ = search_authed_client
    resp = await client.get(
        "/api/v1/search/stats",
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert isinstance(data, dict)
    # Stats should contain table keys or recent_logs
    assert "recent_logs" in data


@pytest.mark.asyncio
async def test_ac14_toggle_provider(
    search_authed_client: tuple[AsyncClient, dict],
    db_session: AsyncSession,
):
    """AC14: POST /api/v1/search/providers/{type}/toggle as admin returns 200."""
    client, seed = search_authed_client
    # Pre-insert a provider row so the UPDATE path is taken (avoids INSERT without id)
    from sqlalchemy import text as sql_text
    from tests.conftest import _get_sync_engine

    sync_eng = _get_sync_engine()
    with sync_eng.connect() as conn:
        conn.execute(
            sql_text(
                "INSERT INTO unified_search_providers (id, tenant_id, entity_type, plugin_name, is_active, config) "
                "VALUES (:id, :tid, 'contact', 'unified_search', true, '{}'::jsonb) "
                "ON CONFLICT DO NOTHING"
            ),
            {"id": str(uuid.uuid4()), "tid": str(seed["tenant_a"].id)},
        )
        conn.commit()
    sync_eng.dispose()

    resp = await client.post(
        "/api/v1/search/providers/contact/toggle",
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    assert "entity_type" in data
    assert "is_active" in data
    assert data["entity_type"] == "contact"


@pytest.mark.asyncio
async def test_ac15_toggle_provider_viewer_403(
    search_client: AsyncClient, db_session: AsyncSession
):
    """AC15: POST /providers/{type}/toggle as viewer → 403."""
    await seed_tenant_and_users(db_session)
    await login_client(search_client, "viewer@tenanta.com")
    resp = await search_client.post(
        "/api/v1/search/providers/contact/toggle",
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 403