learningcircuit--local-deep-research/tests/embeddings/splitters/test_text_splitter_concurrency.py

"""Concurrency regression test for ``get_text_splitter``.

``get_text_splitter`` imports ``langchain_text_splitters`` lazily (to keep the
heavy torch/sentence-transformers stack off the app-startup path). That
package's ``__init__`` eagerly imports ~14 splitter submodules with enough
internal cross-referencing that a *cold* import from several threads at once
can observe a partially-initialized package and raise
``ImportError: cannot import name ... from partially initialized module``.

The RAG auto-index ``ThreadPoolExecutor`` and the per-user document scheduler
both build ``LibraryRAGService`` (→ ``get_text_splitter``) from multiple worker
threads, so this is reachable in production after a restart. ``get_text_splitter``
serializes the cold import behind a module-level lock to prevent it.

This must run in a *fresh* interpreter: inside the pytest process
``langchain_text_splitters`` is already imported (warm), so the race window is
gone. We therefore exercise it in a subprocess. Without the lock this fails
deterministically (~2 of 16 threads raise); with it, all threads succeed.
"""

import subprocess
import sys
import textwrap

import pytest

# allow: no-sut-import — the SUT (get_text_splitter) is imported inside the
# subprocess driver below, because the cold-import race only reproduces in a
# fresh interpreter (it is already warm in the pytest process).

# Driver run in a fresh interpreter: 16 threads, barrier-synced so they all hit
# the cold ``langchain_text_splitters`` import simultaneously. recursive/token
# only — both trigger the package init (the race source) without the
# sentence-transformers model download.
_DRIVER = textwrap.dedent(
    """
    import sys
    import threading

    from local_deep_research.embeddings.splitters import get_text_splitter

    N = 16
    barrier = threading.Barrier(N)
    errors = []

    def worker(i):
        splitter_type = ("recursive", "token")[i % 2]
        try:
            barrier.wait()
            get_text_splitter(
                splitter_type, chunk_size=128, chunk_overlap=10
            )
        except Exception as exc:  # noqa: BLE001 - report any failure
            errors.append(f"{type(exc).__name__}: {exc}")

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(N)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    if errors:
        print(f"FAIL {len(errors)}/{N}: {errors[0]}")
        sys.exit(1)
    print("OK")
    """
)


@pytest.mark.slow
def test_get_text_splitter_concurrent_cold_import_is_safe():
    """Concurrent first-calls must not race on the lazy package import."""
    result = subprocess.run(
        [sys.executable, "-c", _DRIVER],
        capture_output=True,
        text=True,
        timeout=180,
    )
    assert result.returncode == 0, (
        "Concurrent get_text_splitter cold-import raced.\n"
        f"stdout: {result.stdout}\nstderr: {result.stderr[-2000:]}"
    )
    assert "OK" in result.stdout