tests: eliminate NLTK stopwords download race under pytest-xdist (#644)

## Motivation

`TextQuery`, `HybridQuery`, and `AggregateHybridQuery` default to
`stopwords="english"` and lazily download the NLTK corpus the first time
they need it. Under pytest-xdist, several workers can hit the missing
corpus at the same moment and call `nltk.download()` concurrently —
their unzips clobber each other in the shared `nltk_data` directory, and
a worker then reads a half-written corpus. It surfaces only on Python
3.14 because there `sentence-transformers` is skipped, so nothing warms
the corpus serially before the parallel tests start; on 3.10–3.13 the
HuggingFace path fetches it first and the race never opens. It's
intermittent (only some 3.14 jobs, across `redis:8.2` and `redis:latest`
alike), which is the giveaway that Redis has nothing to do with it.

## Fix

Two complementary changes, following the pattern the unit tests already
use (they pass `stopwords=None` to avoid NLTK entirely):

**Pre-download the corpus once, before the workers race for it.** A new
session-scoped autouse fixture fetches `stopwords` a single time,
serialized across xdist workers with a file lock, so exactly one process
does the download and the rest find it already present. This covers
every test that legitimately needs english stopwords — including the
ones that assert on stopword filtering (`test_empty_query_string`,
`test_hybrid_query_stopwords`) and the count-sensitive `TextQuery` tests
in `test_query.py`, whose result counts depend on filtering and so must
keep real stopwords. Adds `filelock` (already present transitively) as
an explicit dev dependency.

**Decouple the tests that don't actually need stopwords.** The hybrid
tests in `test_aggregation.py` match text with an optional `~@field`
clause, so the result set is driven by the vector and filters, not the
text tokens — dropping stopwords there changes nothing about the
assertions. Passing `stopwords=None` on those constructions removes NLTK
from that file except the two tests that specifically exercise the
feature, shrinking the surface that can ever race.

## Verification

Ran the affected tests under `-n 8` with the stopwords corpus deleted
beforehand to simulate a cold runner. The fixture downloads once, no
worker sees a partial corpus, and all tests pass — including the
count-sensitive `TextQuery` tests, confirming english-stopword filtering
is still intact.

Independent of the `redis:latest` WORKERS fix (#641); this addresses a
separate, pre-existing flake.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Low Risk**
> Test-only and dev-dependency changes; no production library behavior
is modified.
> 
> **Overview**
> Fixes intermittent CI failures when parallel pytest workers
concurrently download the NLTK **stopwords** corpus into a shared
`nltk_data` directory.
> 
> A new session-scoped autouse fixture **`ensure_nltk_stopwords`** in
`conftest.py` ensures the corpus exists once before tests run; under
**pytest-xdist**, a **`filelock`** serializes the download so only one
worker fetches it. **`filelock`** is added as an explicit dev dependency
in **`pyproject.toml`** / **`uv.lock`**.
> 
> Hybrid aggregation integration tests that do not assert on stopword
behavior now pass **`stopwords=None`** on **`AggregateHybridQuery`** so
those cases skip NLTK entirely; tests that exercise stopword filtering
keep default or explicit stopword lists.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
17f5435. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: vishal-bala

pyproject.toml

@@ -105,6 +105,7 @@ dev = [
     "types-pyyaml",
     "types-pyopenssl",
     "testcontainers>=4.3.1,<5",
+    "filelock>=3.12,<4",
     "cryptography>=44.0.1",
     "codespell>=2.4.1,<3",
     "langcache>=0.9.0",

tests/conftest.py

@@ -53,6 +53,43 @@ def set_tokenizers_parallelism():
     os.environ["TOKENIZERS_PARALLELISM"] = "false"
 
 
+@pytest.fixture(scope="session", autouse=True)
+def ensure_nltk_stopwords(tmp_path_factory, worker_id):
+    """Pre-download the NLTK ``stopwords`` corpus once, before tests run.
+
+    Query classes that take a string ``stopwords`` value ("english" by
+    default) lazily download the corpus on first use. Under pytest-xdist
+    several workers can hit the missing corpus simultaneously and call
+    ``nltk.download()`` concurrently, corrupting the shared ``nltk_data``
+    directory and producing intermittent "stopwords/english not found"
+    failures. Fetch it here once, serializing across workers with a file lock
+    so exactly one process performs the download.
+    """
+    try:
+        import nltk
+    except ImportError:
+        return
+
+    def _ensure() -> None:
+        try:
+            nltk.data.find("corpora/stopwords")
+        except LookupError:
+            nltk.download("stopwords", quiet=True)
+
+    # Not running under xdist: download in-process.
+    if worker_id == "master":
+        _ensure()
+        return
+
+    # Under xdist: the first worker to grab the lock downloads; the others
+    # wait, then find the corpus already present.
+    from filelock import FileLock
+
+    root_tmp = tmp_path_factory.getbasetemp().parent
+    with FileLock(str(root_tmp / "nltk_stopwords.lock")):
+        _ensure()
+
+
 @pytest.fixture(scope="session", autouse=True)
 def redis_container(worker_id):
     """

tests/integration/test_aggregation.py

@@ -96,6 +96,7 @@ def test_hybrid_query(index):
         vector=vector,
         vector_field_name=vector_field,
         return_fields=return_fields,
+        stopwords=None,
     )
 
     results = index.query(hybrid_query)
@@ -122,6 +123,7 @@ def test_hybrid_query(index):
         vector=vector,
         vector_field_name=vector_field,
         num_results=3,
+        stopwords=None,
     )
 
     results = index.query(hybrid_query)
@@ -177,6 +179,7 @@ def test_hybrid_query_with_filter(index):
         vector_field_name=vector_field,
         filter_expression=filter_expression,
         return_fields=return_fields,
+        stopwords=None,
     )
 
     results = index.query(hybrid_query)
@@ -203,6 +206,7 @@ def test_hybrid_query_with_geo_filter(index):
         vector_field_name=vector_field,
         filter_expression=filter_expression,
         return_fields=return_fields,
+        stopwords=None,
     )
 
     results = index.query(hybrid_query)
@@ -226,6 +230,7 @@ def test_hybrid_query_alpha(index, alpha):
         vector=vector,
         vector_field_name=vector_field,
         alpha=alpha,
+        stopwords=None,
     )
 
     results = index.query(hybrid_query)
@@ -291,6 +296,7 @@ def test_hybrid_query_with_text_filter(index):
         alpha=0.5,
         filter_expression=filter_expression,
         return_fields=["job", "description"],
+        stopwords=None,
     )
 
     results = index.query(hybrid_query)
@@ -309,6 +315,7 @@ def test_hybrid_query_with_text_filter(index):
         alpha=0.5,
         filter_expression=filter_expression,
         return_fields=["description"],
+        stopwords=None,
     )
 
     results = index.query(hybrid_query)
@@ -339,6 +346,7 @@ def test_hybrid_query_word_weights(index, scorer):
         return_fields=return_fields,
         text_scorer=scorer,
         text_weights=weights,
+        stopwords=None,
     )
 
     weighted_results = index.query(weighted_query)
@@ -353,6 +361,7 @@ def test_hybrid_query_word_weights(index, scorer):
         return_fields=return_fields,
         text_scorer=scorer,
         text_weights={},
+        stopwords=None,
     )
 
     unweighted_results = index.query(unweighted_query)
@@ -372,6 +381,7 @@ def test_hybrid_query_word_weights(index, scorer):
         return_fields=return_fields,
         text_scorer=scorer,
         text_weights=weights,
+        stopwords=None,
     )
 
     weighted_results = index.query(weighted_query)
@@ -386,6 +396,7 @@ def test_hybrid_query_word_weights(index, scorer):
         return_fields=return_fields,
         text_scorer=scorer,
         text_weights=None,
+        stopwords=None,
     )
 
     new_query.set_text_weights(weights)