"""Fixtures for chat feature tests."""
import pytest
import uuid
from unittest.mock import MagicMock, patch
from contextlib import contextmanager
from datetime import datetime
@pytest.fixture(autouse=True)
def _stub_request_path_llm():
"""Keep the send-message request path from making real LLM calls.
build_research_context() condenses prior turns into a query-focused
summary via get_llm() before research is dispatched. Most chat tests
don't exercise that summary (they mock the research worker), so stub the
LLM to keep them fast and deterministic regardless of provider
availability. Tests that assert on the summary patch get_llm themselves,
which overrides this stub within their own context.
"""
fake_llm = MagicMock()
fake_llm.invoke.return_value = MagicMock(
content="stub conversation summary"
)
with patch(
"local_deep_research.config.llm_config.get_llm",
return_value=fake_llm,
):
yield
def generate_unique_test_username():
"""Generate a unique test username to avoid conflicts in parallel tests."""
return f"test_user_{uuid.uuid4().hex[:8]}"
def setup_query_mock_with_session(db_session_mock, session_obj):
"""
Set up query mock to return session for the call patterns used by ChatService:
- filter_by().first() — get_session, archive, delete, etc.
- filter_by().with_for_update().first() — update_accumulated_context
- execute(update(...).returning(...)).scalar_one_or_none() — add_message
(atomic increment-and-return; mutates session_obj.message_count to
mirror the SQL UPDATE side effect so callers can assert on it).
Args:
db_session_mock: The mock database session
session_obj: The mock session object to return (or None for not found)
"""
query_mock = MagicMock()
filter_result = MagicMock()
# Set up both patterns to return the same session
filter_result.first.return_value = session_obj
filter_result.with_for_update.return_value.first.return_value = session_obj
query_mock.filter_by.return_value = filter_result
db_session_mock.query.return_value = query_mock
def _execute_side_effect(_stmt, *args, **kwargs):
result = MagicMock()
if session_obj is None:
result.scalar_one_or_none.return_value = None
else:
current = session_obj.message_count or 0
session_obj.message_count = current + 1
result.scalar_one_or_none.return_value = session_obj.message_count
return result
db_session_mock.execute.side_effect = _execute_side_effect
return query_mock
@pytest.fixture
def mock_user_db_session():
"""Mock the database session context manager."""
session_mock = MagicMock()
@contextmanager
def _mock_session(username, password=None):
yield session_mock
with patch(
"src.local_deep_research.chat.service.get_user_db_session",
side_effect=_mock_session,
):
yield session_mock
@pytest.fixture
def chat_service(setup_database_for_all_tests):
"""Provide ChatService with test database."""
from src.local_deep_research.chat.service import ChatService
return ChatService(username="test_user")
@pytest.fixture
def csrf_authenticated_client(app_with_csrf, temp_data_dir):
"""Authenticated test client against the CSRF-enabled app.
Returns ``(client, csrf_token)``. The token is the value the client
must send as ``X-CSRFToken`` (or as a ``csrf_token`` form field)
for mutating requests to be accepted; tests that omit the header
verify that Flask-WTF rejects the request.
Mirrors the registration/login flow of the no-CSRF
``authenticated_client`` fixture in tests/conftest.py — auth pages
use ``WTForms``-managed CSRF tokens so the standard
register-then-login pattern still works against ``app_with_csrf``
when we read the token from the form's ````
first.
"""
import re
import shutil
test_username = generate_unique_test_username()
test_password = "TestPass123"
encrypted_db_dir = temp_data_dir / "encrypted_databases"
if encrypted_db_dir.exists():
try:
shutil.rmtree(encrypted_db_dir)
except Exception:
pass
client = app_with_csrf.test_client()
csrf_input_re = re.compile(
rb'name="csrf_token"\s+(?:type="hidden"\s+)?value="([^"]+)"'
)
csrf_meta_re = re.compile(rb'name="csrf-token"\s+content="([^"]+)"')
def _extract_csrf(resp_data):
m = csrf_input_re.search(resp_data) or csrf_meta_re.search(resp_data)
return m.group(1).decode() if m else None
# 1. GET the register form to obtain its CSRF token, then POST
# register. Successful registration auto-logs the user in (route
# redirects to "/"), so a separate login step is not needed.
reg_form = client.get("/auth/register")
reg_token = _extract_csrf(reg_form.data)
if not reg_token:
raise RuntimeError("CSRF-fixture: no token in /auth/register form")
register_resp = client.post(
"/auth/register",
data={
"username": test_username,
"password": test_password,
"confirm_password": test_password,
"acknowledge": "true",
"csrf_token": reg_token,
},
follow_redirects=False,
)
if register_resp.status_code not in [200, 302]:
raise RuntimeError(
f"CSRF-fixture registration failed: {register_resp.status_code} "
f"{register_resp.data.decode()[:300]}"
)
# 2. GET an authenticated page to harvest the post-login CSRF token
# that mutating chat-API requests must echo back. The chat page
# embeds it in .
page = client.get("/chat/")
api_token = _extract_csrf(page.data)
if not api_token:
raise RuntimeError(
f"CSRF-fixture: no csrf-token in /chat/ "
f"(status={page.status_code}, "
f"location={page.headers.get('Location')})"
)
return client, api_token
@pytest.fixture
def sample_messages():
"""Sample message list for context manager tests."""
return [
{
"id": "msg-001",
"role": "user",
"content": "What is quantum computing?",
"message_type": "query",
"created_at": datetime(2024, 1, 15, 10, 0, 0),
},
{
"id": "msg-002",
"role": "assistant",
"content": "Quantum computing uses quantum mechanics principles like superposition and entanglement to process information. Unlike classical computers that use bits (0 or 1), quantum computers use qubits that can exist in multiple states simultaneously.",
"message_type": "response",
"research_id": "research-abc123",
"created_at": datetime(2024, 1, 15, 10, 1, 0),
},
{
"id": "msg-003",
"role": "user",
"content": "What are the practical applications?",
"message_type": "query",
"created_at": datetime(2024, 1, 15, 10, 2, 0),
},
{
"id": "msg-004",
"role": "assistant",
"content": "Practical applications include cryptography, drug discovery, financial modeling, and optimization problems.",
"message_type": "response",
"research_id": "research-def456",
"created_at": datetime(2024, 1, 15, 10, 3, 0),
},
]
@pytest.fixture
def sample_accumulated_context():
"""Sample accumulated context."""
return {
"key_entities": [
"quantum computing",
"qubits",
"superposition",
"entanglement",
],
"topics": ["physics", "computing", "cryptography"],
"summary": "Discussion about quantum computing fundamentals and applications.",
}
@pytest.fixture
def mock_chat_session():
"""Create a mock ChatSession object."""
session = MagicMock()
session.id = "session-test-123"
session.title = "Test Chat Session"
session.status = "active"
session.message_count = 0
session.accumulated_context = {}
session.created_at = datetime(2024, 1, 15, 10, 0, 0)
session.updated_at = datetime(2024, 1, 15, 10, 0, 0).isoformat()
return session
@pytest.fixture
def mock_settings_manager():
"""Mock the SettingsManager."""
manager_mock = MagicMock()
manager_mock.get_all_settings.return_value = {
"llm.provider": {"value": "ollama"},
"llm.model": {"value": "gemma:latest"},
"search.tool": {"value": "searxng"},
"search.iterations": {"value": 2},
"search.questions_per_iteration": {"value": 3},
}
return manager_mock
@pytest.fixture
def long_query_text():
"""A query text longer than 100 characters for title truncation tests."""
return (
"Can you explain the fundamental principles of quantum computing and how "
"it differs from classical computing, including practical applications "
"in cryptography, drug discovery, and financial modeling?"
)
@pytest.fixture
def many_messages():
"""Generate more than 10 messages for limit testing."""
messages = []
for i in range(15):
messages.append(
{
"id": f"msg-{i:03d}",
"role": "user" if i % 2 == 0 else "assistant",
"content": f"Message content {i}",
"message_type": "query" if i % 2 == 0 else "response",
"research_id": f"research-{i:03d}" if i % 2 == 1 else None,
"created_at": datetime(2024, 1, 15, 10, i, 0),
}
)
return messages
# =============================================================================
# Security Test Fixtures
# =============================================================================
@pytest.fixture
def second_user_client(app, temp_data_dir):
"""
Create a second authenticated test client for cross-user tests.
This creates a separate user to test user isolation and
cross-user access prevention.
"""
# Create unique test username
test_username = generate_unique_test_username()
test_password = "testpassword456"
# Create a test client
client = app.test_client()
# Register and login the second user
with client:
# Register new user
register_response = client.post(
"/auth/register",
data={
"username": test_username,
"password": test_password,
"confirm_password": test_password,
"acknowledge": "true",
},
follow_redirects=False,
)
if register_response.status_code not in [200, 302]:
raise Exception(
f"Second user registration failed: {register_response.status_code}"
)
# Login user
login_response = client.post(
"/auth/login",
data={"username": test_username, "password": test_password},
follow_redirects=False,
)
if login_response.status_code not in [200, 302]:
raise Exception(
f"Second user login failed: {login_response.status_code}"
)
return client
# =============================================================================
# XSS Test Fixtures
# =============================================================================
@pytest.fixture
def xss_payloads():
"""Common XSS attack vectors for testing."""
return {
"script_tags": [
"",
"",
"",
"",
],
"event_handlers": [
"",
"