learningcircuit--local-deep-research/tests/advanced_search_system/test_parallel_search.py

"""Smoke tests for the run_parallel_searches helper.

These verify the contract documented in the module:

* empty input short-circuits to an empty list
* non-empty input returns one ``(query, payload)`` tuple per query
* ``search_fn`` may return any payload type (list, dict, etc.)
* ``context_factory=None`` path works (no Flask context required)
* ``context_factory`` is invoked once per query so each worker gets a
  fresh Flask app context — sharing one context across concurrent workers
  raises ``ValueError: <Token> was created in a different Context``
"""

import threading

from flask import Flask

from local_deep_research.advanced_search_system.parallel_search import (
    run_parallel_searches,
)


def test_empty_queries_returns_empty_list():
    assert run_parallel_searches([], lambda q: [q]) == []


def test_returns_one_tuple_per_query():
    results = run_parallel_searches(["a", "b", "c"], lambda q: [q.upper()])
    assert sorted(results) == [("a", ["A"]), ("b", ["B"]), ("c", ["C"])]


def test_payload_can_be_dict():
    results = run_parallel_searches(
        ["x", "y"], lambda q: {"question": q, "results": [q]}
    )
    assert sorted(results) == [
        ("x", {"question": "x", "results": ["x"]}),
        ("y", {"question": "y", "results": ["y"]}),
    ]


def test_max_workers_passed_through():
    # Smoke test: helper accepts max_workers without error.
    results = run_parallel_searches(["a", "b", "c"], lambda q: q, max_workers=2)
    assert sorted(results) == [("a", "a"), ("b", "b"), ("c", "c")]


def test_default_max_workers():
    # When max_workers is None, defaults to len(queries). Smoke test only.
    results = run_parallel_searches(["a"], lambda q: q)
    assert results == [("a", "a")]


def test_context_factory_called_once_per_query():
    """The factory is invoked once per query (a fresh context per worker),
    never a single shared instance."""
    from unittest.mock import MagicMock

    # side_effect returns a brand-new context-manager mock on every call.
    factory = MagicMock(side_effect=lambda: MagicMock())
    results = run_parallel_searches(
        ["a", "b", "c"], lambda q: q, context_factory=factory
    )
    assert factory.call_count == 3
    assert sorted(results) == [("a", "a"), ("b", "b"), ("c", "c")]


def test_fresh_context_per_worker_survives_concurrency():
    """Regression for the shared-AppContext crash.

    Each worker must receive its OWN Flask app context. We force every
    worker to sit inside its pushed context simultaneously (via a barrier);
    with a per-call factory this completes cleanly. The previous design
    shared one AppContext across all workers, which raised
    ``ValueError: <Token ...> was created in a different Context`` once two
    workers entered/left it concurrently.
    """
    app = Flask(__name__)
    queries = ["a", "b", "c", "d"]
    barrier = threading.Barrier(len(queries), timeout=10)

    def search_fn(q):
        # Block until ALL workers are inside their context at once, forcing
        # the concurrent push/pop that crashed the shared-context design.
        barrier.wait()
        return q

    results = run_parallel_searches(
        queries, search_fn, context_factory=app.app_context
    )
    assert sorted(results) == [("a", "a"), ("b", "b"), ("c", "c"), ("d", "d")]