leocrm/tests/test_unified_search_phase_e.py

"""Phase E tests for the Unified Search plugin.

Covers provider capability flags, RRF multi-fusion, chunking, lifecycle,
API filters, AI tool, new providers, and sensitive-field exclusion.
"""

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.permission_registry import init_permission_registry
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,
    rrf_fusion_multi,
)
from app.plugins.builtins.unified_search.chunking import chunk_text
from app.plugins.builtins.unified_search.lifecycle import (
    handle_entity_delete,
    handle_entity_restore,
    rebuild_index,
    remove_from_index,
)
from app.plugins.builtins.unified_search.ai_tool import (
    TOOL_NAME,
    TOOL_DESCRIPTION,
    unified_search_tool,
    unified_search_handler,
)
from app.plugins.builtins.unified_search.providers.agent_memory_provider import (
    AgentMemorySearchProvider,
)
from app.plugins.builtins.unified_search.providers.ai_chat_provider import (
    AIChatSearchProvider,
)
from app.plugins.builtins.unified_search.providers.workflow_provider import (
    WorkflowSearchProvider,
)
from app.plugins.builtins.unified_search.providers.contact_provider import (
    ContactSearchProvider,
)
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


# ─── Unified Search Fixtures (same as test_unified_search.py) ───


@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)
    init_permission_registry(active_plugin_names={"unified_search"})
    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()
    await login_client(search_client, "admin@tenanta.com")
    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


# ─── 1. Provider Capability Flags ───


def test_all_providers_have_capability_attributes():
    """All registered providers expose supports_fts/vector/rag/graph attributes."""
    registry = SearchProviderRegistry()
    for provider_cls in [
        ContactSearchProvider,
        AgentMemorySearchProvider,
        AIChatSearchProvider,
        WorkflowSearchProvider,
    ]:
        p = provider_cls()
        registry.register(p)

    for provider in registry.get_all():
        assert hasattr(provider, "supports_fts")
        assert hasattr(provider, "supports_vector")
        assert hasattr(provider, "supports_rag")
        assert hasattr(provider, "supports_graph")
        assert isinstance(provider.supports_fts, bool)
        assert isinstance(provider.supports_vector, bool)
        assert isinstance(provider.supports_rag, bool)
        assert isinstance(provider.supports_graph, bool)


def test_get_providers_by_capability():
    """get_providers_by_capability returns only providers supporting the capability."""
    registry = SearchProviderRegistry()
    fts_only = MagicMock()
    fts_only.entity_type = "fts_only"
    fts_only.supports_fts = True
    fts_only.supports_vector = False
    fts_only.supports_rag = False
    fts_only.supports_graph = False

    vector_only = MagicMock()
    vector_only.entity_type = "vector_only"
    vector_only.supports_fts = False
    vector_only.supports_vector = True
    vector_only.supports_rag = False
    vector_only.supports_graph = False

    registry.register(fts_only)
    registry.register(vector_only)

    fts_providers = registry.get_providers_by_capability("fts")
    assert len(fts_providers) == 1
    assert fts_providers[0].entity_type == "fts_only"

    vec_providers = registry.get_providers_by_capability("vector")
    assert len(vec_providers) == 1
    assert vec_providers[0].entity_type == "vector_only"

    rag_providers = registry.get_providers_by_capability("rag")
    assert rag_providers == []

    graph_providers = registry.get_providers_by_capability("graph")
    assert graph_providers == []


def test_get_capabilities_returns_flags():
    """get_capabilities returns correct flags for each entity type."""
    registry = SearchProviderRegistry()
    contact = ContactSearchProvider()
    ai_chat = AIChatSearchProvider()
    registry.register(contact)
    registry.register(ai_chat)

    contact_caps = registry.get_capabilities("contact")
    assert contact_caps == {"fts": True, "vector": True, "rag": False, "graph": False}

    ai_chat_caps = registry.get_capabilities("ai_chat")
    assert ai_chat_caps == {"fts": True, "vector": False, "rag": False, "graph": False}

    # Unknown entity type returns all False
    unknown_caps = registry.get_capabilities("unknown")
    assert unknown_caps == {"fts": False, "vector": False, "rag": False, "graph": False}


# ─── 2. RRF Multi-Fusion ───


def test_rrf_fusion_multi_two_lists():
    """rrf_fusion_multi fuses two result lists."""
    list_a = [{"id": "1", "title": "A"}, {"id": "2", "title": "B"}]
    list_b = [{"id": "2", "title": "B"}, {"id": "3", "title": "C"}]

    fused = rrf_fusion_multi([("a", list_a), ("b", list_b)])
    ids = [str(r.get("id", "")) for r in fused]
    assert "1" in ids
    assert "2" in ids
    assert "3" in ids
    # Item 2 appears in both lists → highest score
    assert str(fused[0].get("id", "")) == "2"
    assert fused[0]["_score"] > fused[1]["_score"]


def test_rrf_fusion_multi_three_lists():
    """rrf_fusion_multi fuses three result lists."""
    list_a = [{"id": "1"}]
    list_b = [{"id": "1"}, {"id": "2"}]
    list_c = [{"id": "1"}, {"id": "2"}, {"id": "3"}]

    fused = rrf_fusion_multi([("a", list_a), ("b", list_b), ("c", list_c)])
    ids = [str(r.get("id", "")) for r in fused]
    assert "1" in ids
    assert "2" in ids
    assert "3" in ids
    # Item 1 appears in all 3 lists → highest score
    assert str(fused[0].get("id", "")) == "1"


def test_rrf_fusion_multi_four_lists():
    """rrf_fusion_multi fuses four result lists."""
    list_a = [{"id": "1"}]
    list_b = [{"id": "1"}, {"id": "2"}]
    list_c = [{"id": "1"}, {"id": "2"}, {"id": "3"}]
    list_d = [{"id": "1"}, {"id": "2"}, {"id": "3"}, {"id": "4"}]

    fused = rrf_fusion_multi([("a", list_a), ("b", list_b), ("c", list_c), ("d", list_d)])
    ids = [str(r.get("id", "")) for r in fused]
    assert "1" in ids
    assert "2" in ids
    assert "3" in ids
    assert "4" in ids
    assert str(fused[0].get("id", "")) == "1"


def test_rrf_fusion_multi_items_in_multiple_lists_score_higher():
    """Items appearing in multiple lists get higher scores."""
    list_a = [{"id": "1"}, {"id": "2"}]
    list_b = [{"id": "1"}]

    fused = rrf_fusion_multi([("a", list_a), ("b", list_b)])
    by_id = {str(r.get("id", "")): r["_score"] for r in fused}
    # Item 1 appears in both lists → higher score than item 2 (only in list a)
    assert by_id["1"] > by_id["2"]


def test_rrf_fusion_backward_compatibility():
    """rrf_fusion remains backward compatible with the multi-fusion wrapper."""
    fts_results = [{"id": "1", "title": "A"}, {"id": "2", "title": "B"}]
    vec_results = [{"id": "2", "title": "B"}, {"id": "3", "title": "C"}]

    fused = rrf_fusion(fts_results, vec_results, "contact")
    ids = [str(r.get("id", "")) for r in fused]
    assert "1" in ids
    assert "2" in ids
    assert "3" in ids
    # Item 2 appears in both → highest score
    assert str(fused[0].get("id", "")) == "2"
    # _entity_type is set for backward compatibility
    assert all(r.get("_entity_type") == "contact" for r in fused)


def test_rrf_fusion_multi_empty_inputs():
    """rrf_fusion_multi with empty lists returns empty list."""
    assert rrf_fusion_multi([]) == []
    assert rrf_fusion_multi([("a", []), ("b", [])]) == []


# ─── 3. Chunking ───


def test_chunk_text_empty():
    """chunk_text with empty text returns empty list."""
    assert chunk_text("") == []
    assert chunk_text("   ") == []
    assert chunk_text(None) == []


def test_chunk_text_short():
    """chunk_text with text shorter than chunk_size returns a single chunk."""
    chunks = chunk_text("Hello world", chunk_size=1000, overlap=200)
    assert len(chunks) == 1
    assert chunks[0]["chunk_index"] == 0
    assert chunks[0]["chunk_text"] == "Hello world"
    assert "chunk_hash" in chunks[0]


def test_chunk_text_long():
    """chunk_text splits long text into multiple overlapping chunks."""
    text = "word " * 500  # ~2500 chars
    chunks = chunk_text(text, chunk_size=1000, overlap=200)
    assert len(chunks) > 1
    # Chunks overlap: chunk 1 starts at chunk_size - overlap
    assert chunks[1]["chunk_text"].startswith(chunks[0]["chunk_text"][-200:])


def test_chunk_text_exact_multiple():
    """chunk_text with text exactly matching chunk_size returns a single chunk."""
    text = "a" * 1000
    chunks = chunk_text(text, chunk_size=1000, overlap=200)
    assert len(chunks) == 1
    assert chunks[0]["chunk_text"] == text


def test_chunk_overlap_correct():
    """chunk overlap is correct between consecutive chunks."""
    text = "word " * 1000  # ~5000 chars
    chunks = chunk_text(text, chunk_size=1000, overlap=200)
    assert len(chunks) > 1
    # Verify overlap: the tail of chunk N equals the head of chunk N+1
    for i in range(1, len(chunks)):
        prev_tail = chunks[i - 1]["chunk_text"][-200:]
        assert chunks[i]["chunk_text"].startswith(prev_tail)


def test_chunk_hash_deterministic():
    """chunk_hash is deterministic for the same input."""
    text = "Some document text for chunking"
    c1 = chunk_text(text, chunk_size=100, overlap=20)
    c2 = chunk_text(text, chunk_size=100, overlap=20)
    assert c1 == c2
    assert c1[0]["chunk_hash"] == c2[0]["chunk_hash"]


def test_chunk_text_normalizes_whitespace():
    """chunk_text normalizes whitespace to avoid degenerate chunks."""
    chunks = chunk_text("Hello   world\n\n  test", chunk_size=1000, overlap=200)
    assert len(chunks) == 1
    assert chunks[0]["chunk_text"] == "Hello world test"


# ─── 4. Lifecycle ───


async def _create_contact_with_embedding(db_session: AsyncSession) -> tuple[uuid.UUID, uuid.UUID]:
    """Create a tenant, user, and contact with an embedding set."""
    from app.models.contact import Contact
    from app.models.tenant import Tenant
    from app.models.user import User, UserTenant
    from app.core.auth import hash_password

    tenant = Tenant(name="Lifecycle Tenant", slug="lifecycle-tenant")
    db_session.add(tenant)
    await db_session.flush()
    user = User(
        email="lifecycle@example.com",
        name="Lifecycle",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        preferences={},
    )
    db_session.add(user)
    await db_session.flush()
    db_session.add(UserTenant(user_id=user.id, tenant_id=tenant.id, is_default=True, role="admin"))
    await db_session.flush()
    contact = Contact(
        tenant_id=tenant.id,
        firstname="Lifecycle",
        surname="Test",
        email_1="lifecycle@example.com",
        created_by=user.id,
        updated_by=user.id,
    )
    db_session.add(contact)
    await db_session.flush()
    # Set embedding + search_tsv so we can verify removal.
    # The test DB uses create_all (no Alembic migrations), so indexed_at
    # (added by migration 0005) may be missing — add it if needed.
    from sqlalchemy import text as sql_text
    await db_session.execute(
        sql_text(
            "ALTER TABLE contacts ADD COLUMN IF NOT EXISTS indexed_at TIMESTAMPTZ"
        )
    )
    # The test DB (create_all) defines search_tsv as a generated column, but
    # production (migration 0001) uses a plain column maintained by a trigger.
    # The lifecycle code sets search_tsv = NULL, which requires a plain column.
    # Convert it to match production schema (PG 13+ DROP EXPRESSION) and drop
    # the recompute trigger so the NULL set by remove_from_index is preserved.
    # The DO block makes this idempotent across test runs (schema persists).
    await db_session.execute(
        sql_text(
            "DO $$ BEGIN "
            "IF EXISTS (SELECT 1 FROM pg_attribute a "
            "          WHERE a.attrelid = 'contacts'::regclass "
            "            AND a.attname = 'search_tsv' "
            "            AND a.attgenerated <> '') THEN "
            "  ALTER TABLE contacts ALTER COLUMN search_tsv DROP EXPRESSION; "
            "END IF; "
            "END $$;"
        )
    )
    await db_session.execute(
        sql_text("DROP TRIGGER IF EXISTS contacts_tsv_update ON contacts")
    )
    await db_session.execute(
        sql_text(
            "UPDATE contacts SET embedding = cast(:emb AS vector), "
            "search_tsv = to_tsvector('pg_catalog.german', :tsv), indexed_at = now() "
            "WHERE id = :eid"
        ),
        {"emb": str([0.1] * 768), "tsv": "lifecycle test", "eid": contact.id},
    )
    await db_session.commit()
    return tenant.id, contact.id


@pytest.mark.asyncio
async def test_remove_from_index_sets_embedding_and_tsv_null(db_session: AsyncSession):
    """remove_from_index sets embedding and TSV to NULL."""
    tenant_id, contact_id = await _create_contact_with_embedding(db_session)

    await remove_from_index(db_session, "contact", contact_id, tenant_id)

    from sqlalchemy import text as sql_text
    result = await db_session.execute(
        sql_text("SELECT embedding, search_tsv, indexed_at FROM contacts WHERE id = :eid"),
        {"eid": contact_id},
    )
    row = result.mappings().first()
    assert row is not None
    assert row["embedding"] is None
    assert row["search_tsv"] is None
    assert row["indexed_at"] is None


@pytest.mark.asyncio
async def test_remove_from_index_unknown_entity_type(db_session: AsyncSession):
    """remove_from_index with unknown entity type does nothing (no error)."""
    await remove_from_index(db_session, "unknown_type", uuid.uuid4(), uuid.uuid4())


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.embedding.index_entity", new_callable=AsyncMock, return_value=True)
async def test_rebuild_index_regenerates_embedding(mock_index_entity, db_session: AsyncSession):
    """rebuild_index regenerates the embedding via index_entity."""
    tenant_id, contact_id = await _create_contact_with_embedding(db_session)

    success = await rebuild_index(db_session, "contact", contact_id, tenant_id)
    assert success is True
    mock_index_entity.assert_called_once()


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.embedding.index_entity", new_callable=AsyncMock, return_value=False)
async def test_rebuild_index_failure_returns_false(mock_index_entity, db_session: AsyncSession):
    """rebuild_index returns False when index_entity fails."""
    tenant_id, contact_id = await _create_contact_with_embedding(db_session)

    success = await rebuild_index(db_session, "contact", contact_id, tenant_id)
    assert success is False


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.lifecycle.remove_from_index", new_callable=AsyncMock)
async def test_handle_entity_delete_calls_remove_from_index(mock_remove, db_session: AsyncSession):
    """handle_entity_delete calls remove_from_index."""
    tenant_id = uuid.uuid4()
    entity_id = uuid.uuid4()
    await handle_entity_delete(db_session, "contact", entity_id, tenant_id)
    mock_remove.assert_called_once_with(db_session, "contact", entity_id, tenant_id)


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.lifecycle.rebuild_index", new_callable=AsyncMock, return_value=True)
async def test_handle_entity_restore_calls_rebuild_index(mock_rebuild, db_session: AsyncSession):
    """handle_entity_restore calls rebuild_index."""
    tenant_id = uuid.uuid4()
    entity_id = uuid.uuid4()
    await handle_entity_restore(db_session, "contact", entity_id, tenant_id)
    mock_rebuild.assert_called_once_with(db_session, "contact", entity_id, tenant_id)


# ─── 5. API Filters ───


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.routes.hybrid_search", new_callable=AsyncMock)
async def test_search_with_date_filters(mock_hybrid_search, search_authed_client: tuple[AsyncClient, dict]):
    """Search with date_from/date_to filters returns filtered results."""
    client, _ = search_authed_client
    mock_hybrid_search.return_value = [
        {
            "entity_type": "contact",
            "entity_id": str(uuid.uuid4()),
            "title": "Alpha",
            "snippet": "",
            "score": 0.9,
            "data": {},
            "_created_at": "2026-01-01T00:00:00+00:00",
            "_updated_at": "2026-01-01T00:00:00+00:00",
            "_tags": "",
        },
        {
            "entity_type": "contact",
            "entity_id": str(uuid.uuid4()),
            "title": "Beta",
            "snippet": "",
            "score": 0.8,
            "data": {},
            "_created_at": "2026-06-01T00:00:00+00:00",
            "_updated_at": "2026-06-01T00:00:00+00:00",
            "_tags": "",
        },
    ]

    resp = await client.post(
        "/api/v1/search",
        json={"query": "test", "date_from": "2026-03-01", "date_to": "2026-12-31"},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    results = data["results"]
    assert len(results) == 1
    assert results[0]["title"] == "Beta"


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.routes.hybrid_search", new_callable=AsyncMock)
async def test_search_with_tags_filter(mock_hybrid_search, search_authed_client: tuple[AsyncClient, dict]):
    """Search with tags filter returns only matching results."""
    client, _ = search_authed_client
    mock_hybrid_search.return_value = [
        {
            "entity_type": "contact",
            "entity_id": str(uuid.uuid4()),
            "title": "Alpha",
            "snippet": "",
            "score": 0.9,
            "data": {},
            "_created_at": None,
            "_updated_at": None,
            "_tags": "vip,partner",
        },
        {
            "entity_type": "contact",
            "entity_id": str(uuid.uuid4()),
            "title": "Beta",
            "snippet": "",
            "score": 0.8,
            "data": {},
            "_created_at": None,
            "_updated_at": None,
            "_tags": "lead",
        },
    ]

    resp = await client.post(
        "/api/v1/search",
        json={"query": "test", "tags": ["vip"]},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    results = data["results"]
    assert len(results) == 1
    assert results[0]["title"] == "Alpha"


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.routes.hybrid_search", new_callable=AsyncMock)
async def test_search_with_sort_parameter(mock_hybrid_search, search_authed_client: tuple[AsyncClient, dict]):
    """Search with sort=name sorts results by title."""
    client, _ = search_authed_client
    mock_hybrid_search.return_value = [
        {
            "entity_type": "contact",
            "entity_id": str(uuid.uuid4()),
            "title": "Zeta",
            "snippet": "",
            "score": 0.9,
            "data": {},
            "_created_at": None,
            "_updated_at": None,
            "_tags": "",
        },
        {
            "entity_type": "contact",
            "entity_id": str(uuid.uuid4()),
            "title": "Alpha",
            "snippet": "",
            "score": 0.8,
            "data": {},
            "_created_at": None,
            "_updated_at": None,
            "_tags": "",
        },
    ]

    resp = await client.post(
        "/api/v1/search",
        json={"query": "test", "sort": "name"},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200
    data = resp.json()
    results = data["results"]
    assert len(results) == 2
    assert results[0]["title"] == "Alpha"
    assert results[1]["title"] == "Zeta"


@pytest.mark.asyncio
async def test_facets_endpoint_returns_correct_structure(search_authed_client: tuple[AsyncClient, dict]):
    """GET /api/v1/search/facets returns correct structure."""
    client, _ = search_authed_client
    resp = await client.get("/api/v1/search/facets", headers=ORIGIN_HEADER)
    assert resp.status_code == 200
    data = resp.json()
    assert "entity_types" in data
    assert "tags" in data
    assert "date_ranges" in data
    assert isinstance(data["entity_types"], list)
    assert isinstance(data["tags"], list)
    assert isinstance(data["date_ranges"], dict)


# ─── 6. AI Tool ───


def test_unified_search_tool_name_and_description():
    """unified_search_tool has correct name and description."""
    assert unified_search_tool.name == "unified_search"
    assert TOOL_NAME == "unified_search"
    assert TOOL_DESCRIPTION == unified_search_tool.description
    assert "Hybrid-Suche" in unified_search_tool.description
    assert unified_search_tool.required_permission == "search:read"
    assert unified_search_tool.category == "search"


def test_unified_search_tool_parameters():
    """unified_search_tool exposes query/entity_types/limit parameters."""
    params = unified_search_tool.parameters
    assert params["type"] == "object"
    assert "query" in params["properties"]
    assert "entity_types" in params["properties"]
    assert "limit" in params["properties"]
    assert params["required"] == ["query"]


def test_unified_search_tool_openai_schema():
    """unified_search_tool to_openai_schema returns valid function schema."""
    schema = unified_search_tool.to_openai_schema()
    assert schema["type"] == "function"
    assert schema["function"]["name"] == "unified_search"
    assert "parameters" in schema["function"]


@pytest.mark.asyncio
@patch("app.plugins.builtins.unified_search.ai_tool.hybrid_search", new_callable=AsyncMock)
@patch("app.plugins.builtins.unified_search.ai_tool.llm_analyze_query", new_callable=AsyncMock)
async def test_unified_search_handler_returns_compact_results(
    mock_llm, mock_hybrid, db_session: AsyncSession
):
    """unified_search_handler returns compact results."""
    mock_llm.return_value = {"normalized_query": "test", "semantic_terms": []}
    mock_hybrid.return_value = [
        {
            "entity_type": "contact",
            "entity_id": str(uuid.uuid4()),
            "title": "John Doe",
            "snippet": "john@example.com",
            "score": 0.95,
        }
    ]

    # Patch the session factory to use the test session
    sf = async_sessionmaker(bind=db_session.bind, expire_on_commit=False, class_=AsyncSession)
    with patch("app.core.db.get_session_factory", return_value=sf):
        result = await unified_search_handler(
            {"query": "John", "limit": 5},
            {"tenant_id": str(uuid.uuid4()), "user_id": str(uuid.uuid4()), "is_system_admin": True},
        )

    data = json.loads(result)
    assert "count" in data
    assert "results" in data
    assert data["count"] == 1
    assert data["results"][0]["entity_type"] == "contact"
    assert data["results"][0]["title"] == "John Doe"
    assert "score" in data["results"][0]


@pytest.mark.asyncio
async def test_unified_search_handler_missing_query():
    """unified_search_handler returns error for missing query."""
    result = await unified_search_handler({}, {})
    data = json.loads(result)
    assert "error" in data
    assert data["error"] == "query is required"


@pytest.mark.asyncio
async def test_unified_search_handler_missing_tenant():
    """unified_search_handler returns error for missing tenant context."""
    result = await unified_search_handler({"query": "test"}, {})
    data = json.loads(result)
    assert "error" in data
    assert data["error"] == "missing tenant context"


# ─── 7. New Providers ───


def test_agent_memory_provider_imports_and_flags():
    """AgentMemorySearchProvider imports correctly with correct flags."""
    provider = AgentMemorySearchProvider()
    assert provider.entity_type == "agent_memory"
    assert provider.supports_fts is True
    assert provider.supports_vector is True
    assert provider.supports_rag is False
    assert provider.supports_graph is False


def test_ai_chat_provider_imports_and_flags():
    """AIChatSearchProvider imports correctly with correct flags."""
    provider = AIChatSearchProvider()
    assert provider.entity_type == "ai_chat"
    assert provider.supports_fts is True
    assert provider.supports_vector is False
    assert provider.supports_rag is False
    assert provider.supports_graph is False


def test_workflow_provider_imports_and_flags():
    """WorkflowSearchProvider imports correctly with correct flags."""
    provider = WorkflowSearchProvider()
    assert provider.entity_type == "workflow"
    assert provider.supports_fts is True
    assert provider.supports_vector is False
    assert provider.supports_rag is False
    assert provider.supports_graph is False


def test_new_providers_to_search_result():
    """New providers produce correct search result dicts."""
    agent_memory = AgentMemorySearchProvider()
    result = agent_memory.to_search_result({"id": "1", "content": "Remembered fact", "memory_type": "fact"})
    assert result["entity_type"] == "agent_memory"
    assert result["entity_id"] == "1"
    assert result["title"] == "Remembered fact"
    assert result["data"]["memory_type"] == "fact"

    ai_chat = AIChatSearchProvider()
    result = ai_chat.to_search_result({"id": "2", "content": "Chat message", "role": "user", "session_title": "Session"})
    assert result["entity_type"] == "ai_chat"
    assert result["entity_id"] == "2"
    assert result["title"] == "Session"
    assert result["data"]["role"] == "user"

    workflow = WorkflowSearchProvider()
    result = workflow.to_search_result({"id": "3", "name": "Workflow A", "description": "Desc", "trigger_event": "contact.created"})
    assert result["entity_type"] == "workflow"
    assert result["entity_id"] == "3"
    assert result["title"] == "Workflow A"
    assert result["data"]["trigger_event"] == "contact.created"


# ─── 8. Sensitive Fields Exclusion ───


def test_sensitive_fields_not_in_search_tsv():
    """Sensitive fields are excluded from search_tsv via filter_for_search."""
    from app.core.sensitive_data import filter_for_search, get_sensitive_fields

    sensitive = get_sensitive_fields("contact")
    assert "password_hash" in sensitive
    assert "smtp_password" in sensitive
    assert "api_key" in sensitive

    data = {
        "displayname": "John Doe",
        "email_1": "john@example.com",
        "password_hash": "secret-hash",
        "smtp_password": "secret-pw",
        "api_key": "secret-key",
    }
    filtered = filter_for_search(data, "contact")
    assert "displayname" in filtered
    assert "email_1" in filtered
    assert "password_hash" not in filtered
    assert "smtp_password" not in filtered
    assert "api_key" not in filtered


def test_sensitive_fields_not_in_embedding_text():
    """Sensitive fields are excluded from embedding text via filter_for_embeddings."""
    from app.core.sensitive_data import filter_for_embeddings

    data = {
        "displayname": "John Doe",
        "email_1": "john@example.com",
        "password_hash": "secret-hash",
        "smtp_password": "secret-pw",
        "api_key": "secret-key",
    }
    filtered = filter_for_embeddings(data, "contact")
    assert "displayname" in filtered
    assert "email_1" in filtered
    assert "password_hash" not in filtered
    assert "smtp_password" not in filtered
    assert "api_key" not in filtered


def test_contact_provider_embedding_text_excludes_sensitive():
    """ContactSearchProvider.get_embedding_text selects only non-sensitive columns."""
    provider = ContactSearchProvider()
    # The SQL selects only safe columns — verify no sensitive column names appear
    import inspect
    source = inspect.getsource(provider.get_embedding_text)
    assert "password_hash" not in source
    assert "smtp_password" not in source
    assert "api_key" not in source
    assert "oauth_token" not in source


def test_sensitive_fields_redacted_in_index_entity():
    """index_entity redacts sensitive fields from embedding text."""
    from app.core.sensitive_data import get_sensitive_fields
    from app.plugins.builtins.unified_search.embedding import index_entity

    # Verify the sensitive-data guard is present in index_entity source
    import inspect
    source = inspect.getsource(index_entity)
    assert "get_sensitive_fields" in source
    assert "REDACTED" in source