test: erweitere Tests auf 103 (65 unified_search + 38 ai_proactive)
- Embedding Pipeline Tests (7) - Query Understanding Tests (6) - Search Engine Tests (4) - Jobs Tests (5) - Provider-spezifische Tests (12) - API Tests (3) - Context Tools Tests (7) - Proactive Engine Tests (10) - API Tests (6) - Jobs Tests (1) - Plugin Lifecycle Tests (3) Alle 103 Tests bestanden.
This commit is contained in:
Agent Zero
File diff suppressed because it is too large
Load Diff
@@ -8,6 +8,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
@@ -29,6 +30,46 @@ 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 ───
|
||||
|
||||
|
||||
@@ -567,3 +608,711 @@ async def test_autocomplete_returns_titles(db_session: AsyncSession):
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user