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

import logging
import os
import sys
import tempfile
import types
import shutil
import uuid
from pathlib import Path
from unittest.mock import Mock

import pytest
from loguru import logger
from sqlalchemy import create_engine, event
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import sessionmaker, Session

import local_deep_research.utilities.db_utils as db_utils_module
from local_deep_research.database.models import Base
from local_deep_research.database.auth_db import (
    dispose_auth_engine,
    init_auth_database,
)
from local_deep_research.web.app_factory import create_app
from local_deep_research.settings.manager import (
    SettingsManager,
)

# Import our mock fixtures
try:
    from .mock_fixtures import (
        get_mock_arxiv_response,
        get_mock_error_responses,
        get_mock_findings,
        get_mock_google_pse_response,
        get_mock_ollama_response,
        get_mock_pubmed_article,
        get_mock_pubmed_response,
        get_mock_research_history,
        get_mock_search_results,
        get_mock_semantic_scholar_response,
        get_mock_settings,
        get_mock_wikipedia_response,
    )
except ImportError:
    # Mock fixtures not yet created, skip for now
    pass


def generate_unique_test_username(prefix: str = "pytest_user") -> str:
    """Generate unique username using UUID instead of timestamp.

    This ensures no collisions when running tests in parallel with pytest-xdist.
    """
    return f"{prefix}_{uuid.uuid4().hex[:12]}"


@pytest.fixture(autouse=True)
def reset_all_singletons():
    """Reset all singletons before and after each test.

    This ensures proper isolation when running with pytest-xdist (-n auto).
    Without this, singleton state can leak between tests running in parallel,
    causing intermittent failures.
    """

    def _reset():
        # SocketIOService
        try:
            from local_deep_research.web.services.socket_service import (
                SocketIOService,
            )

            SocketIOService._instance = None
        except ImportError:
            pass

        # BackgroundJobScheduler (both class and module-level)
        try:
            from local_deep_research.scheduler import background

            # Stop any running scheduler BEFORE dropping the singleton
            # reference so its APScheduler thread doesn't emit logs to a
            # closed pytest stderr sink during teardown.
            if background.BackgroundJobScheduler._instance is not None:
                try:
                    background.BackgroundJobScheduler._instance.stop()
                except Exception:
                    # Never fail fixture teardown on scheduler edge cases.
                    pass
            background.BackgroundJobScheduler._instance = None
            # Also reset module-level global if it exists
            if hasattr(background, "_scheduler_instance"):
                background._scheduler_instance = None
        except ImportError:
            pass

        # QueueProcessorV2 (module-level singleton thread). Like the
        # scheduler above, create_app() starts this background thread, but
        # nothing stopped it between tests — so the first test's processor
        # ran for the whole worker, looping through SettingsManager ->
        # SQLCipher connection opens on the shared db_manager concurrently
        # with every later test (and emitting logs to a closed pytest
        # stderr sink at teardown). Stop it here so each test starts clean;
        # the next create_app() restarts it (stop() clears the running
        # guard). Tests that exercise the queue patch this object, so
        # stopping the real thread does not affect them.
        try:
            from local_deep_research.web.queue.processor_v2 import (
                queue_processor,
            )

            queue_processor.stop()
        except ImportError:
            pass
        except Exception:
            # Never fail fixture teardown on queue-processor edge cases.
            pass

        # ProviderDiscovery - reset both class-level and module-level state
        try:
            from local_deep_research.llm.providers import auto_discovery
            from local_deep_research.llm.providers.auto_discovery import (
                ProviderDiscovery,
            )

            # Reset class-level singleton
            ProviderDiscovery._instance = None
            ProviderDiscovery._providers = {}
            if hasattr(ProviderDiscovery, "_discovered"):
                ProviderDiscovery._discovered = False

            # Reset the module-level singleton instance's state
            if hasattr(auto_discovery, "provider_discovery"):
                auto_discovery.provider_discovery._discovered = False
                auto_discovery.provider_discovery._providers = {}

            # Re-register the built-in providers. get_llm() has no fallback
            # construction path anymore, so a test that called
            # clear_llm_registry() would otherwise break every later test
            # that uses a built-in provider (order-dependent failures).
            from local_deep_research.llm.providers import discover_providers

            discover_providers(force_refresh=True)
        except ImportError:
            pass

        # AccountLockoutManager singleton
        try:
            from local_deep_research.security import account_lockout

            account_lockout._manager = None
        except ImportError:
            pass

        # AdaptiveRateLimitTracker: no singleton to reset — get_tracker()
        # returns a fresh instance each call.

    _reset()
    yield
    _reset()


@pytest.fixture(autouse=True)
def database_operation_timeout():
    """Set shorter timeouts for database operations in tests.

    This helps tests fail fast instead of hanging indefinitely when
    database contention occurs.
    """
    original = os.environ.get("LDR_DB_BUSY_TIMEOUT")
    os.environ["LDR_DB_BUSY_TIMEOUT"] = "5000"  # 5 seconds in tests
    yield
    if original:
        os.environ["LDR_DB_BUSY_TIMEOUT"] = original
    else:
        os.environ.pop("LDR_DB_BUSY_TIMEOUT", None)


def pytest_configure(config):
    """Configure pytest with custom markers."""
    config.addinivalue_line(
        "markers",
        "requires_llm: mark test as requiring a real LLM (skipped when LDR_TESTING_WITH_MOCKS=true)",
    )
    config.addinivalue_line(
        "markers",
        "integration: mark test as an integration test requiring live network access (skipped when LDR_TESTING_WITH_MOCKS=true)",
    )
    # In CI, LDR_TESTING_WITH_MOCKS is set via Docker environment variables
    # For local testing, set it here if not already set
    if not os.environ.get("LDR_TESTING_WITH_MOCKS"):
        os.environ["LDR_TESTING_WITH_MOCKS"] = "true"


@pytest.fixture(autouse=True)
def skip_if_no_real_llm(request):
    """Skip tests marked with @pytest.mark.requires_llm when running with mocks."""
    if request.node.get_closest_marker("requires_llm"):
        if os.environ.get("LDR_TESTING_WITH_MOCKS", "").lower() == "true":
            pytest.skip("Test requires real LLM but running with mocks")


@pytest.fixture(autouse=True)
def skip_integration_in_mock_mode(request):
    """Skip integration tests when running with mocks (CI default)."""
    if request.node.get_closest_marker("integration"):
        if os.environ.get("LDR_TESTING_WITH_MOCKS", "true").lower() == "true":
            pytest.skip(
                "Integration test skipped in mock mode "
                "(set LDR_TESTING_WITH_MOCKS=false to run)"
            )


@pytest.fixture
def temp_data_dir():
    """Create a temporary data directory for testing."""
    temp_dir = tempfile.mkdtemp()
    yield Path(temp_dir)
    # Use ignore_errors=True because SQLite WAL/SHM files may still be held
    # open by database connections that are disposed later in teardown order
    # (cleanup_database_connections autouse fixture runs after this fixture).
    shutil.rmtree(temp_dir, ignore_errors=True)


@pytest.fixture(autouse=True)
def cleanup_database_connections():
    """Clean up database connections before and after each test.

    This fixture ensures proper cleanup of database connections with
    logging for debugging CI issues.
    """
    # Import here to avoid circular imports
    from local_deep_research.database.encrypted_db import db_manager
    from local_deep_research.web.auth.session_manager import session_manager

    # Close all connections and sessions before test
    db_manager.close_all_databases()
    session_manager.sessions.clear()

    # Dispose auth engine so it will be recreated with correct path
    dispose_auth_engine()

    yield

    # Close all connections and sessions after test
    db_manager.close_all_databases()
    session_manager.sessions.clear()

    # Dispose auth engine after test
    dispose_auth_engine()


@pytest.fixture
def app(temp_data_dir, monkeypatch):
    """Create a Flask app configured for testing."""
    # Override data directory
    monkeypatch.setenv("LDR_DATA_DIR", str(temp_data_dir))

    # Production default (256000) makes PBKDF2 dominate wall-clock in
    # fixtures that register + log in users (authenticated_client runs
    # it twice per test). Under xdist CPU contention from leaked auth
    # background threads, setup blows past the 60s pytest-timeout and
    # the worker dies — taking the rest of its tests with it and
    # dropping coverage below the 50% gate. sqlcipher_utils declares
    # MIN_KDF_ITERATIONS_TESTING=1 specifically to support this.
    monkeypatch.setenv("LDR_DB_CONFIG_KDF_ITERATIONS", "1000")

    # Note: PYTEST_CURRENT_TEST is automatically set by pytest, which
    # app_factory.py checks to disable secure cookies for testing

    # Create app with testing config
    app, _ = create_app()
    app.config["TESTING"] = True
    app.config["WTF_CSRF_ENABLED"] = False
    app.config["WTF_CSRF_CHECK_DEFAULT"] = False
    app.config["SESSION_COOKIE_SECURE"] = False  # For testing without HTTPS
    # Override PREFERRED_URL_SCHEME so test client defaults to HTTP, not HTTPS
    # Tests that need HTTPS should explicitly set environ_base={"wsgi.url_scheme": "https"}
    app.config["PREFERRED_URL_SCHEME"] = "http"

    # Initialize auth database in test directory
    init_auth_database()

    return app


@pytest.fixture
def client(app):
    """Create a test client."""
    return app.test_client()


@pytest.fixture
def app_with_csrf(temp_data_dir, monkeypatch):
    """Flask app with CSRF protection ENABLED.

    The default `app` fixture disables CSRF for ergonomic testing of every
    other code path; this opt-in fixture is for tests that specifically
    verify CSRF token enforcement on mutating routes. Do not use for
    general request testing — those tests should not be coupled to CSRF.
    """
    monkeypatch.setenv("LDR_DATA_DIR", str(temp_data_dir))
    monkeypatch.setenv("LDR_DB_CONFIG_KDF_ITERATIONS", "1000")
    app, _ = create_app()
    app.config["TESTING"] = True
    # Deliberately do NOT disable CSRF — this fixture exists to verify it.
    app.config["SESSION_COOKIE_SECURE"] = False
    app.config["PREFERRED_URL_SCHEME"] = "http"
    init_auth_database()
    return app


@pytest.fixture
def authenticated_client(app, temp_data_dir):
    """Create a test client with an authenticated user."""
    # Create unique test username using UUID to avoid conflicts in parallel tests
    test_username = generate_unique_test_username()
    test_password = "TestPass123"

    # Clear any existing user database
    encrypted_db_dir = temp_data_dir / "encrypted_databases"
    if encrypted_db_dir.exists():
        import shutil

        try:
            shutil.rmtree(encrypted_db_dir)
        except Exception as e:
            logger.warning(f"Could not remove encrypted_db_dir: {e}")

    # Create a test client
    client = app.test_client()

    # Register and login the user through the normal flow
    with client:
        # Register new unique 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"Registration failed with status {register_response.status_code}: "
                f"{register_response.data.decode()[:500]}"
            )

        # 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"Login failed with status {login_response.status_code}: "
                f"{login_response.data.decode()[:500]}"
            )

    return client


@pytest.fixture()
def setup_database_for_all_tests(
    tmp_path_factory, mocker
):  # Use function-scoped mocker so patches don't leak to other tests
    """
    Provides a database setup for a temporary SQLite file database for the entire test session.
    It patches db_utils.get_db_session and db_utils.get_settings_manager to use this test DB.
    """

    # Call cache_clear on the functions from db_utils_module.
    # This ensures any pre-existing cached instances are gone.
    # We must ensure db_utils_module is imported before this point.
    try:
        if hasattr(db_utils_module.get_db_session, "cache_clear"):
            db_utils_module.get_db_session.cache_clear()
        if hasattr(db_utils_module.get_settings_manager, "cache_clear"):
            db_utils_module.get_settings_manager.cache_clear()
        # get_setting_from_db_main_thread has been removed

    except Exception as e:
        logger.warning(f"Failed to clear db_utils caches aggressively: {e}")
        # This shouldn't prevent test run, but indicates a problem with cache_clear

    # Debug tmp_path_factory behavior
    temp_dir = tmp_path_factory.mktemp("db_test_data")
    db_file = temp_dir / "test_settings.db"
    db_url = f"sqlite:///{db_file}"

    engine = None
    try:
        engine = create_engine(db_url)
    except Exception:
        logger.exception("Failed to create SQLAlchemy engine")
        raise

    # Enable SQLite FK enforcement on every connection. Mirrors production
    # (sqlcipher_utils.apply_performance_pragmas) and the FK-aware
    # fixtures in tests/database/test_research_strategy_fk_regression.py
    # and tests/database/test_chat_models.py. Fixture is function-scoped
    # so the listener is registered fresh per test — no leakage.
    # NB: parameter is named `dbapi` (not `dbapi_connection`) to sidestep
    # the custom-checks raw-SQL detector, which uses an unanchored regex
    # `conn.execute` that flags ANY identifier ending in "conn.execute".
    # The PRAGMA is the canonical way to enable FK enforcement on SQLite
    # and has no ORM equivalent.
    @event.listens_for(engine, "connect")
    def _enable_fk(dbapi, _connection_record):
        dbapi.execute("PRAGMA foreign_keys = ON")

    try:
        Base.metadata.create_all(engine)
    except SQLAlchemyError:
        logger.exception("SQLAlchemyError during Base.metadata.create_all")
        raise
    except Exception:
        logger.exception("Unexpected error during Base.metadata.create_all")
        raise

    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    temp_session = SessionLocal()
    temp_settings_manager = SettingsManager(db_session=temp_session)

    try:
        temp_settings_manager.load_from_defaults_file(commit=True)
    except Exception:
        logger.exception("Failed to load default settings")
        temp_session.rollback()  # Rollback if default loading fails
        raise  # Re-raise to fail the test if default loading is critical
    finally:
        temp_session.close()  # Close the temporary session used for loading defaults

    # Clear caches and patch
    db_utils_module.get_db_session.cache_clear()

    mock_get_db_session = mocker.patch(
        "local_deep_research.utilities.db_utils.get_db_session"
    )
    mock_get_db_session.side_effect = SessionLocal

    mock_get_settings_manager = mocker.patch(
        "local_deep_research.utilities.db_utils.get_settings_manager"
    )

    def _settings_with_maybe_fake_db(
        db_session: Session | None = None, *_, **__
    ) -> SettingsManager:
        if db_session is None:
            # Use the mock.
            db_session = mock_get_db_session()
        return SettingsManager(db_session=db_session)

    mock_get_settings_manager.side_effect = _settings_with_maybe_fake_db

    yield SessionLocal  # Yield the SessionLocal class for individual tests to create sessions

    if engine:
        engine.dispose()  # Dispose the engine to close all connections
    # tmp_path_factory handles deleting the temporary directory and its contents


@pytest.fixture
def mock_db_session(mocker):
    return mocker.MagicMock()


@pytest.fixture
def mock_logger(mocker):
    return mocker.patch("local_deep_research.settings.manager.logger")


# ============== LLM and Search Mock Fixtures (inspired by scottvr) ==============


@pytest.fixture
def mock_llm():
    """Create a mock LLM for testing."""
    mock = Mock()
    mock.invoke.return_value = Mock(content="Mocked LLM response")
    return mock


@pytest.fixture
def mock_search():
    """Create a mock search engine for testing."""
    mock = Mock()
    mock.run.return_value = get_mock_search_results()
    return mock


@pytest.fixture
def mock_search_system():
    """Create a mock search system for testing."""
    mock = Mock()
    mock.analyze_topic.return_value = get_mock_findings()
    mock.all_links_of_system = [
        {"title": "Source 1", "link": "https://example.com/1"},
        {"title": "Source 2", "link": "https://example.com/2"},
    ]
    return mock


# ============== API Response Mock Fixtures ==============


@pytest.fixture
def mock_wikipedia_response():
    """Mock response from Wikipedia API."""
    return get_mock_wikipedia_response()


@pytest.fixture
def mock_arxiv_response():
    """Mock response from arXiv API."""
    return get_mock_arxiv_response()


@pytest.fixture
def mock_pubmed_response():
    """Mock response from PubMed API."""
    return get_mock_pubmed_response()


@pytest.fixture
def mock_pubmed_article():
    """Mock PubMed article detail."""
    return get_mock_pubmed_article()


@pytest.fixture
def mock_semantic_scholar_response():
    """Mock response from Semantic Scholar API."""
    return get_mock_semantic_scholar_response()


@pytest.fixture
def mock_google_pse_response():
    """Mock response from Google PSE API."""
    return get_mock_google_pse_response()


@pytest.fixture
def mock_ollama_response():
    """Mock response from Ollama API."""
    return get_mock_ollama_response()


# ============== Data Structure Mock Fixtures ==============


@pytest.fixture
def mock_search_results():
    """Sample search results for testing."""
    return get_mock_search_results()


@pytest.fixture
def mock_findings():
    """Sample research findings for testing."""
    return get_mock_findings()


@pytest.fixture
def mock_error_responses():
    """Collection of error responses for testing."""
    return get_mock_error_responses()


# ============== Environment and Module Mock Fixtures ==============


@pytest.fixture
def mock_env_vars(monkeypatch):
    """Set up mock environment variables for testing."""
    monkeypatch.setenv("LDR_LLM__PROVIDER", "test_provider")
    monkeypatch.setenv("LDR_LLM__MODEL", "test_model")
    monkeypatch.setenv("LDR_SEARCH__TOOL", "test_tool")
    monkeypatch.setenv("LDR_SEARCH__ITERATIONS", "2")
    yield


@pytest.fixture
def mock_llm_config(monkeypatch):
    """Create and patch a mock llm_config module."""
    # Create a mock module
    mock_module = types.ModuleType("mock_llm_config")

    # Add necessary functions and variables
    def get_llm(*args, **kwargs):
        mock = Mock()
        mock.invoke.return_value = Mock(content="Mocked LLM response")
        return mock

    mock_module.get_llm = get_llm

    # Patch the module
    monkeypatch.setitem(
        sys.modules, "local_deep_research.config.llm_config", mock_module
    )
    monkeypatch.setattr("local_deep_research.config.llm_config", mock_module)

    return mock_module


# ============== Test Database Fixtures ==============


@pytest.fixture
def temp_db_path():
    """Create a temporary database file for testing."""
    fd, path = tempfile.mkstemp(suffix=".db")
    os.close(fd)
    yield path
    os.unlink(path)


@pytest.fixture
def mock_research_history():
    """Mock research history entries."""
    return get_mock_research_history()


@pytest.fixture
def mock_settings():
    """Mock settings configuration."""
    return get_mock_settings()


# ============== Loguru Logging Fixtures ==============


class PropagateHandler(logging.Handler):
    def emit(self, record):
        logging.getLogger(record.name).handle(record)


@pytest.fixture
def loguru_caplog(caplog):
    """Make pytest caplog work with loguru.

    Standard pytest caplog doesn't capture loguru logs out of the box.
    This fixture propagates loguru logs to the standard logging module
    so they can be captured by pytest's caplog fixture.

    Note: The local_deep_research package disables loguru logging by default
    (see src/local_deep_research/__init__.py). This fixture re-enables it
    for the duration of the test.

    See: https://loguru.readthedocs.io/en/stable/resources/migration.html

    Usage:
        def test_something(loguru_caplog):
            import logging
            with loguru_caplog.at_level(logging.WARNING):
                # ... code that uses loguru logging ...
            assert "expected message" in loguru_caplog.text
    """
    # Re-enable logging for local_deep_research (disabled in __init__.py)
    logger.enable("local_deep_research")

    handler_id = logger.add(
        PropagateHandler(),
        format="{message}",
        level="DEBUG",
        # diagnose=False: loguru defaults to True, which dumps repr() of
        # every traceback frame's local on exception. The propagating handler
        # forwards those into pytest's stdlib log capture, where they end up
        # in CI logs. Hygiene companion to #4185 / #4384.
        diagnose=False,
    )
    yield caplog
    logger.remove(handler_id)
    # Re-disable logging to restore original state
    logger.disable("local_deep_research")


@pytest.fixture
def loguru_caplog_full(caplog):
    """Like ``loguru_caplog`` but captures the exception block too.

    Use this in security tests that assert a credential never appears in
    log output. The default ``loguru_caplog`` fixture uses
    ``format="{message}"``, which excludes the rendered exception block
    that ``logger.exception()`` (and ``exc_info=True``) emit — so a leak
    that lives only in the traceback would false-pass.

    This fixture appends ``{exception}`` to the format and enables
    ``backtrace=True`` so the cause chain (``__cause__`` / ``__context__``)
    is rendered. ``diagnose`` stays off to match production
    (``utilities/log_utils.py`` runs ``diagnose=False`` unless
    ``LDR_APP_DEBUG`` is set).
    """
    logger.enable("local_deep_research")

    handler_id = logger.add(
        PropagateHandler(),
        format="{message}\n{exception}",
        level="DEBUG",
        backtrace=True,
        diagnose=False,
    )
    yield caplog
    logger.remove(handler_id)
    logger.disable("local_deep_research")