learningcircuit--local-deep-research/tests/web/test_startup_import_weight.py

# allow: no-sut-import — guardian; runs create_app()/imports the SUT startup
# path in a fresh subprocess (in-process imports would mask the regression)
# and asserts the sentence-transformers / langchain-text-splitters stack is
# absent from sys.modules.
"""Regression guard for the app-startup import weight.

Background
----------
``create_app()`` runs on every app startup. It used to load the
sentence-transformers / langchain-text-splitters stack (~2s+, and on the
scheduler path the full transformers/torch stack, ~6-10s) at import time,
which on slow CI runners pushed startup past the 20s ``faulthandler``
watchdog (``responsive-test-gate / ui-tests``) and hung the app. The
original faulthandler dump showed the main thread stuck importing
``scheduler/background.py`` (it transitively imported the RAG service ->
``embeddings.splitters`` -> ``langchain_text_splitters`` -> sentence
transformers). The lazy text-splitter fix landed in #4829; see also the
earlier #4490 / #4431.

What this guard does (and does NOT) assert
------------------------------------------
#4829 removed the **sentence-transformers / langchain-text-splitters**
stack from the startup path (it is now imported lazily, only when text
splitting / RAG indexing actually runs). That is the invariant guarded
here.

It does NOT assert ``transformers`` / ``torch`` are absent from
``create_app()``. Those are STILL imported at startup through a separate,
pre-existing path not addressed by the lazy-splitter fix:
``web/app_factory.py`` register_blueprints -> most route blueprints ->
``search_system`` / ``config.llm_config`` ->
``langchain_core.language_models``, whose ``base`` module imports
``transformers`` at module load (``from transformers import
GPT2TokenizerFast`` — inside a ``try/except ImportError``, but transformers
is installed here, so it fires). ``BaseChatModel`` is imported at module
level by ~15 core modules and most startup route blueprints, so removing
transformers/torch from ``create_app()`` is an architectural change well
beyond the splitter fix. Asserting their absence here would be a guaranteed
false failure.

Both checks run in a FRESH subprocess: the broader pytest session has
almost always already imported these modules into ``sys.modules``, which
would mask any regression in-process.

If a check fails, some module on the startup path grew a module-level
import that drags in the splitter stack. Fix it by deferring that import
to the function where it is used (an RAG / text-splitting job), not by
relaxing this test.
"""

import subprocess
import sys

import pytest

# Probe 1: run the REAL create_app() in a clean interpreter and assert the
# splitter stack never loads. Covers the ENTIRE startup surface (every
# blueprint + middleware + queue processor), so there is no "which module did
# we forget to import" blind spot. Side-effecting singletons are patched out
# and all state is redirected to a throwaway data dir for determinism.
_PROBE_CREATE_APP = """
import sys, os, shutil, tempfile
_data_dir = tempfile.mkdtemp(prefix="ldr_startup_guard_")
os.environ["LDR_DATA_DIR"] = _data_dir
os.environ.setdefault("TESTING", "1")
from unittest.mock import patch, MagicMock

try:
    with (
        patch("local_deep_research.web.app_factory.SocketIOService"),
        patch("local_deep_research.web.queue.processor_v2.queue_processor"),
        patch("atexit.register"),
        patch(
            "local_deep_research.scheduler.background.get_background_job_scheduler",
            return_value=MagicMock(),
        ),
    ):
        from local_deep_research.web.app_factory import create_app
        create_app()

    stack = ("sentence_transformers", "langchain_text_splitters")
    loaded = sorted(m for m in stack if m in sys.modules)
    if loaded:
        print("SPLITTER_STACK_IMPORTED:" + ",".join(loaded))
        sys.exit(1)
    print("CREATE_APP_SPLITTER_CLEAN")
finally:
    # create_app() writes a .secret_key into LDR_DATA_DIR; clean it up so the
    # guard doesn't leak a temp dir per run on persistent dev machines.
    shutil.rmtree(_data_dir, ignore_errors=True)
"""

# Probe 2: the original faulthandler hang site. Importing scheduler.background
# in isolation must stay fully decoupled from the ML stack — not even
# transformers/torch at module load. NOTE: #4829 already decoupled the
# scheduler -> RAG service -> splitter chain, so a regression in
# text_splitter_registry is caught by Probe 1, not here. This probe guards the
# scheduler module's OWN import surface — it trips if a NEW heavy import lands
# directly on scheduler.background's chain (a tighter, non-redundant contract).
_PROBE_SCHEDULER = """
import sys
import local_deep_research.scheduler.background  # noqa: F401

heavy = ("transformers", "sentence_transformers", "torch", "langchain_text_splitters")
loaded = sorted(m for m in heavy if m in sys.modules)
if loaded:
    print("SCHEDULER_IMPORT_HEAVY:" + ",".join(loaded))
    sys.exit(1)
print("SCHEDULER_IMPORT_CLEAN")
"""


def _run(probe: str) -> subprocess.CompletedProcess:
    # Keep this comfortably below pyproject's global pytest timeout (180s,
    # thread method). If a probe ever genuinely hangs, we want subprocess.run
    # to raise a clean TimeoutExpired here — yielding a localized failure with
    # the partial output — rather than pytest-timeout firing first and
    # hard-killing the whole xdist worker (losing its other tests/coverage).
    try:
        return subprocess.run(
            [sys.executable, "-c", probe],
            capture_output=True,
            text=True,
            timeout=120,
        )
    except subprocess.TimeoutExpired as exc:
        pytest.fail(
            "Startup-import probe did not finish within 120s — likely a hang on "
            f"the app-startup path.\npartial stdout:\n{exc.stdout}\n"
            f"partial stderr:\n{exc.stderr}"
        )


def test_create_app_does_not_import_splitter_stack():
    """Running create_app() must not load sentence-transformers / splitters."""
    result = _run(_PROBE_CREATE_APP)
    assert (
        result.returncode == 0 and "CREATE_APP_SPLITTER_CLEAN" in result.stdout
    ), (
        "create_app() pulled the sentence-transformers / langchain-text-splitters "
        "stack onto the startup path. This reintroduces the startup-watchdog hang "
        "fixed in #4829 — defer the offending top-level import to its call "
        f"site.\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}"
    )


def test_scheduler_import_is_ml_free():
    """Importing scheduler.background (the original hang site) loads no ML stack."""
    result = _run(_PROBE_SCHEDULER)
    assert (
        result.returncode == 0 and "SCHEDULER_IMPORT_CLEAN" in result.stdout
    ), (
        "Importing local_deep_research.scheduler.background pulled in the ML stack. "
        "Keep the splitter import lazy (inside get_text_splitter) — and any RAG "
        "service import out of module scope — so the scheduler import on the "
        f"startup critical path stays light.\n"
        f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
    )