learningcircuit--local-deep-research/tests/chat/conftest.py

"""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 ``<input name="csrf_token">``
    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 <meta name="csrf-token">.
    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": [
            "<script>alert('XSS')</script>",
            "<script>alert(document.cookie)</script>",
            "<SCRIPT>alert('XSS')</SCRIPT>",
            "<script src='http://evil.com/xss.js'></script>",
        ],
        "event_handlers": [
            "<img src=x onerror=alert('XSS')>",
            "<svg onload=alert('XSS')>",
            "<body onload=alert('XSS')>",
            "<div onmouseover=alert('XSS')>hover</div>",
        ],
        "javascript_urls": [
            "<a href='javascript:alert(1)'>Click</a>",
            "<iframe src='javascript:alert(1)'></iframe>",
        ],
        "nested": [
            "<<script>script>alert('XSS')<</script>/script>",
            "<scr<script>ipt>alert('XSS')</scr</script>ipt>",
        ],
        "encoded": [
            "&#60;script&#62;alert('XSS')&#60;/script&#62;",
            "%3Cscript%3Ealert('XSS')%3C/script%3E",
        ],
    }


@pytest.fixture
def malicious_session_ids():
    """Session ID values that could be used in attacks."""
    return [
        # Path traversal
        "../../../etc/passwd",
        "..%2F..%2Fetc%2Fpasswd",
        "session-id/../../admin",
        # Null bytes
        "valid-id%00.txt",
        "test\x00admin",
        # SQL injection
        "' OR '1'='1",
        "'; DROP TABLE chat_sessions; --",
        "1 UNION SELECT * FROM users--",
        # XSS in IDs
        "<script>alert(1)</script>",
        "session-<img onerror=alert(1)>",
    ]


# =============================================================================
# Fuzz Test Fixtures
# =============================================================================


@pytest.fixture
def random_unicode_strings():
    """Collection of Unicode strings for fuzz testing."""
    return [
        "",  # Empty
        " ",  # Whitespace
        "\t\n\r",  # Control chars
        "Hello, World!",  # Normal ASCII
        "Привет мир",  # Cyrillic
        "你好世界",  # Chinese
        "こんにちは",  # Japanese
        "مرحبا",  # Arabic
        "🎉🔥💯",  # Emoji
        "a" * 10000,  # Very long
        "\x00\x01\x02",  # Null and control
        "line1\nline2\rline3",  # Newlines
        "'\"<>&",  # HTML special chars
    ]


@pytest.fixture
def boundary_values():
    """Boundary values for testing limits."""
    return {
        "message_lengths": [0, 1, 99, 100, 101, 9999, 10000, 10001],
        "title_lengths": [0, 1, 499, 500, 501],
        "pagination_limits": [-1, 0, 1, 20, 100, 101, 1000],
        "pagination_offsets": [-1, 0, 1, 100, 10000],
    }