test: harden Redis-backed test isolation under pytest-xdist (#546) (#636)

## Motivation

The integration suite runs under `pytest -n auto`, and now and then a
single matrix job fails on a test that passes everywhere else — and
passes again on rerun. The most recent example was
`test_hybrid_query_with_geo_filter` returning 0 results instead of 3 on
exactly one Python/redis-py/Redis combination. These failures read as
version-specific, but they aren't: they're a test-isolation problem
wearing a disguise.

The key fact is how the suite isolates workers. Each xdist worker gets
its **own** Redis container, so two tests can only interfere when they
run on the **same** worker *and* reach for the **same** Redis key names.
For that to actually corrupt a result, two things have to line up:

1. a **shared resource name** (a name scoped only to the worker, or a
hard-coded one), and
2. a **way for state to outlive its test** — cleanup that only runs on
the success path, or an index recreated with `overwrite=True` but no
`drop=True`, leaving stale documents under a reused prefix.

When both hold, one test sees another's leftovers — empty results,
inflated counts, dtype mismatches — and which test loses depends
entirely on the order the scheduler picked. That's the retry-passing,
pseudo-random signature.

#543 introduced a `redis_test_name` helper that mints per-test-unique
names and began migrating fixtures onto it. This PR finishes that
migration for the modules that can genuinely cause flakiness, and
tightens their recreate/cleanup so leftovers can't survive.

## What changed

**Recreate group** — per-test names + `create(overwrite=True,
drop=True)`:
- `test_hybrid.py` and `test_aggregation.py` (these two shared the
*identical* `user_index_{worker_id}` / `v1_{worker_id}` — the direct
cause of the geo-filter failure), plus `test_search_results.py`,
`test_svs_integration.py`, and the shared `conftest.py` index fixtures.

**Fixed-name collisions** — per-test names + guaranteed teardown:
- `from_existing_complex` (hard-coded `"test"`) and
`no_proactive_module_validation` (`"my_index"`) in the sync + async
search-index suites.
- The three `test_embedcache_warnings.py` tests sharing `"test_cache"`.
- The shared `"doc"` prefix / `"test-index"` name across
`test_no_proactive_module_checks.py`.

**Raw `worker_id` collision** — unique names + cleanup:
- The two `test_llmcache.py` tests that both created
`float64_cache_{worker_id}` and never cleaned up.

**Cluster tests** — the one place cross-worker collisions are real,
since every cluster test points at a single hard-coded
`redis://localhost:7001`: `test_redis_cluster_support.py` and
`test_cluster_pipelining.py` now use per-test names, `drop=True`, and
`try/finally` cleanup.

## Scope and follow-ups

- The `SemanticRouter.delete()` leak of its `{name}:route_config` key is
a product-code gap, not a test-only fix, so it's split into #634 (which
also owns the dependent `test_key_separator_handling.py` cleanup).
- The remaining #546 items are **benign hygiene** — leaks under
per-test-unique names that can't contaminate another test, only
accumulate harmless keys in a worker's throwaway Redis. Their exact
remaining scope is enumerated in a comment on #546.

## Testing

Honest caveat for reviewers: this was developed in an environment
without Docker, so I could **not** run the integration suite. `black`,
`isort`, and `mypy` pass. Please run `make test` and a repeated `pytest
-n auto` pass to confirm the flake is gone before merging — this is a
test-only change, so behavior is fully validated by the suite itself.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Low Risk**
> Changes are confined to the test suite; production library behavior is
unchanged.
> 
> **Overview**
> **Hardens integration tests** so parallel `pytest-xdist` workers and
shared Redis/cluster endpoints do not leave stale keys or collide on
index/cache names.
> 
> Fixtures and tests now use **`redis_test_name`** (per-test unique
index/prefix/cache names) instead of **`worker_id`-only** or hard-coded
names like `"test"` / `"my_index"` / `"test_cache"`. Shared
**`conftest`** index fixtures (`flat_index`, `hnsw_index`, async
variants) and modules such as **`test_hybrid`**, **`test_aggregation`**,
**`test_search_results`**, and **`test_svs_integration`** adopt that
naming and call **`create(overwrite=True, drop=True)`** so documents
under a reused prefix are cleared before load.
> 
> **Teardown** is tightened with **`try/finally`**,
**`delete(drop=True)`**, and cache **`clear()`** / **`aclear()`** on
failure paths (e.g. `from_existing_complex`, module-validation tests,
**`test_llmcache`** dtype caches, **`test_embedcache_warnings`**).
**Cluster** tests use per-test names and **`drop=True`** because all
workers share **`localhost:7001`**.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
c4e6b33. 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

tests/conftest.py

@@ -518,7 +518,7 @@ def flat_index(sample_data, redis_url, redis_test_name):
     )
 
     # create the index (no data yet)
-    index.create(overwrite=True)
+    index.create(overwrite=True, drop=True)
 
     # Prepare and load the data
     def hash_preprocess(item: dict) -> dict:
@@ -575,7 +575,7 @@ async def async_flat_index(sample_data, redis_url, redis_test_name):
     )
 
     # create the index (no data yet)
-    await index.create(overwrite=True)
+    await index.create(overwrite=True, drop=True)
 
     # Prepare and load the data
     def hash_preprocess(item: dict) -> dict:
@@ -631,7 +631,7 @@ async def async_hnsw_index(sample_data, redis_url, redis_test_name):
     )
 
     # create the index (no data yet)
-    await index.create(overwrite=True)
+    await index.create(overwrite=True, drop=True)
 
     # Prepare and load the data
     def hash_preprocess(item: dict) -> dict:
@@ -687,7 +687,7 @@ def hnsw_index(sample_data, redis_url, redis_test_name):
     )
 
     # create the index (no data yet)
-    index.create(overwrite=True)
+    index.create(overwrite=True, drop=True)
 
     # Prepare and load the data
     def hash_preprocess(item: dict) -> dict:

tests/integration/test_aggregation.py

@@ -8,13 +8,13 @@
 
 
 @pytest.fixture
-def index(multi_vector_data, redis_url, worker_id):
+def index(multi_vector_data, redis_url, redis_test_name):
 
     index = SearchIndex.from_dict(
         {
             "index": {
-                "name": f"user_index_{worker_id}",
-                "prefix": f"v1_{worker_id}",
+                "name": redis_test_name("user_index"),
+                "prefix": redis_test_name("v1"),
                 "storage_type": "hash",
             },
             "fields": [
@@ -59,8 +59,9 @@ def index(multi_vector_data, redis_url, worker_id):
         redis_url=redis_url,
     )
 
-    # create the index (no data yet)
-    index.create(overwrite=True)
+    # create the index (no data yet); drop any stale docs left by an
+    # interrupted earlier run sharing this worker's Redis database
+    index.create(overwrite=True, drop=True)
 
     # prepare and load the data
     def hash_preprocess(item: dict) -> dict:

tests/integration/test_async_search_index.py

@@ -113,11 +113,11 @@ async def test_search_index_from_existing_url(async_index, redis_url):
 
 
 @pytest.mark.asyncio
-async def test_search_index_from_existing_complex(async_client):
+async def test_search_index_from_existing_complex(async_client, redis_test_name):
     schema = {
         "index": {
-            "name": "test",
-            "prefix": "test",
+            "name": redis_test_name("complex_index"),
+            "prefix": redis_test_name("complex"),
             "storage_type": "json",
         },
         "fields": [
@@ -143,33 +143,37 @@ async def test_search_index_from_existing_complex(async_client):
         ],
     }
     async_index = AsyncSearchIndex.from_dict(schema, redis_client=async_client)
-    await async_index.create(overwrite=True)
+    await async_index.create(overwrite=True, drop=True)
 
     try:
-        async_index2 = await AsyncSearchIndex.from_existing(
-            async_index.name, redis_client=async_client
-        )
-    except Exception as e:
-        pytest.skip(str(e))
-
-    # Verify index metadata matches
-    assert async_index2.schema.index.name == async_index.schema.index.name
-    assert async_index2.schema.index.prefix == async_index.schema.index.prefix
-    assert (
-        async_index2.schema.index.storage_type == async_index.schema.index.storage_type
-    )
-
-    # Verify non-vector fields are present
-    for field_name in ["user", "credit_score", "job", "age"]:
-        assert field_name in async_index2.schema.fields
+        try:
+            async_index2 = await AsyncSearchIndex.from_existing(
+                async_index.name, redis_client=async_client
+            )
+        except Exception as e:
+            pytest.skip(str(e))
+
+        # Verify index metadata matches
+        assert async_index2.schema.index.name == async_index.schema.index.name
+        assert async_index2.schema.index.prefix == async_index.schema.index.prefix
         assert (
-            async_index2.schema.fields[field_name].type
-            == async_index.schema.fields[field_name].type
+            async_index2.schema.index.storage_type
+            == async_index.schema.index.storage_type
         )
 
-    # Vector field may not be present on older Redis versions
-    if "user_embedding" in async_index2.schema.fields:
-        assert async_index2.schema.fields["user_embedding"].type == "vector"
+        # Verify non-vector fields are present
+        for field_name in ["user", "credit_score", "job", "age"]:
+            assert field_name in async_index2.schema.fields
+            assert (
+                async_index2.schema.fields[field_name].type
+                == async_index.schema.fields[field_name].type
+            )
+
+        # Vector field may not be present on older Redis versions
+        if "user_embedding" in async_index2.schema.fields:
+            assert async_index2.schema.fields["user_embedding"].type == "vector"
+    finally:
+        await async_index.delete(drop=True)
 
 
 def test_search_index_no_prefix(index_schema):
@@ -493,7 +497,9 @@ async def test_search_index_that_owns_client_disconnect_sync(index_schema, redis
 
 
 @pytest.mark.asyncio
-async def test_async_search_index_no_proactive_module_validation(redis_url):
+async def test_async_search_index_no_proactive_module_validation(
+    redis_url, redis_test_name
+):
     """
     Updated test for issue #370: AsyncSearchIndex should not validate modules proactively.
     Operations should fail naturally if modules are missing.
@@ -505,7 +511,7 @@ async def test_async_search_index_no_proactive_module_validation(redis_url):
         # Create index - validation should only set lib name, not check modules
         index = AsyncSearchIndex(
             schema=IndexSchema.from_dict(
-                {"index": {"name": "my_index"}, "fields": fields}
+                {"index": {"name": redis_test_name("my_index")}, "fields": fields}
             ),
             redis_client=client,
         )
@@ -517,8 +523,11 @@ async def test_async_search_index_no_proactive_module_validation(redis_url):
         # The actual operation (create) will succeed if modules are present
         await index.create(overwrite=True, drop=True)
 
-        # Verify index was created successfully (modules are present in test env)
-        assert await index.exists()
+        try:
+            # Verify index was created successfully (modules are present in test env)
+            assert await index.exists()
+        finally:
+            await index.delete(drop=True)
 
 
 @pytest.mark.asyncio

tests/integration/test_cluster_pipelining.py

@@ -33,13 +33,20 @@ def test_real_cluster_pipeline_get_protocol_version(redis_cluster_url):
 
 
 @pytest.mark.requires_cluster
-def test_real_searchindex_with_cluster_batch_operations(redis_cluster_url):
+def test_real_searchindex_with_cluster_batch_operations(
+    redis_cluster_url, redis_test_name
+):
     """
     Test SearchIndex.load() with Redis Cluster.
     """
     # Create schema like the user had
+    index_prefix = redis_test_name("doc")
     schema_dict = {
-        "index": {"name": "test-real-365", "prefix": "doc", "storage_type": "hash"},
+        "index": {
+            "name": redis_test_name("test-real-365"),
+            "prefix": index_prefix,
+            "storage_type": "hash",
+        },
         "fields": [
             {"name": "id", "type": "tag"},
             {"name": "text", "type": "text"},
@@ -52,7 +59,7 @@ def test_real_searchindex_with_cluster_batch_operations(redis_cluster_url):
     index = SearchIndex(schema, redis_url=redis_cluster_url)
 
     # Create the index
-    index.create(overwrite=True)
+    index.create(overwrite=True, drop=True)
 
     try:
         # Test data like user had
@@ -67,11 +74,11 @@ def test_real_searchindex_with_cluster_batch_operations(redis_cluster_url):
         )
 
         assert len(keys) == 10
-        assert all(k.startswith("doc:") for k in keys)
+        assert all(k.startswith(f"{index_prefix}:") for k in keys)
 
     finally:
         # Clean up
-        index.delete()
+        index.delete(drop=True)
 
 
 @pytest.mark.requires_cluster
@@ -114,14 +121,18 @@ def test_cluster_pipeline_protocol_version_directly():
 
 
 @pytest.mark.requires_cluster
-def test_batch_search_with_real_cluster(redis_cluster_url):
+def test_batch_search_with_real_cluster(redis_cluster_url, redis_test_name):
     """
     Test batch_search which uses get_protocol_version internally.
     """
     from redisvl.query import FilterQuery
 
     schema_dict = {
-        "index": {"name": "test-batch-365", "prefix": "batch", "storage_type": "json"},
+        "index": {
+            "name": redis_test_name("test-batch-365"),
+            "prefix": redis_test_name("batch"),
+            "storage_type": "json",
+        },
         "fields": [
             {"name": "id", "type": "tag"},
             {"name": "category", "type": "tag"},
@@ -131,7 +142,7 @@ def test_batch_search_with_real_cluster(redis_cluster_url):
     schema = IndexSchema.from_dict(schema_dict)
     index = SearchIndex(schema, redis_url=redis_cluster_url)
 
-    index.create(overwrite=True)
+    index.create(overwrite=True, drop=True)
 
     try:
         # Load test data
@@ -151,17 +162,21 @@ def test_batch_search_with_real_cluster(redis_cluster_url):
         assert len(results) == 3
 
     finally:
-        index.delete()
+        index.delete(drop=True)
 
 
 @pytest.mark.requires_cluster
 @pytest.mark.parametrize("ttl", [None, 30])
-def test_cluster_load_with_ttl(redis_cluster_url, ttl):
+def test_cluster_load_with_ttl(redis_cluster_url, ttl, redis_test_name):
     """
     Test that TTL is correctly set on keys when using load() with ttl parameter on cluster.
     """
     schema_dict = {
-        "index": {"name": "test-ttl-cluster", "prefix": "ttl", "storage_type": "hash"},
+        "index": {
+            "name": redis_test_name("test-ttl-cluster"),
+            "prefix": redis_test_name("ttl"),
+            "storage_type": "hash",
+        },
         "fields": [
             {"name": "id", "type": "tag"},
             {"name": "text", "type": "text"},
@@ -171,7 +186,7 @@ def test_cluster_load_with_ttl(redis_cluster_url, ttl):
     schema = IndexSchema.from_dict(schema_dict)
     index = SearchIndex(schema, redis_url=redis_cluster_url)
 
-    index.create(overwrite=True)
+    index.create(overwrite=True, drop=True)
 
     try:
         # Load test data with TTL parameter
@@ -190,4 +205,4 @@ def test_cluster_load_with_ttl(redis_cluster_url, ttl):
             assert abs(key_ttl - ttl) <= 5
 
     finally:
-        index.delete()
+        index.delete(drop=True)

tests/integration/test_embedcache_warnings.py

@@ -19,10 +19,14 @@ def reset_warning_flag():
 
 
 @pytest.mark.asyncio
-async def test_sync_methods_warn_with_async_only_client(async_client, caplog):
+async def test_sync_methods_warn_with_async_only_client(
+    async_client, caplog, redis_test_name
+):
     """Test that sync methods warn when only async client is provided."""
     # Initialize EmbeddingsCache with only async_redis_client
-    cache = EmbeddingsCache(name="test_cache", async_redis_client=async_client)
+    cache = EmbeddingsCache(
+        name=redis_test_name("test_cache"), async_redis_client=async_client
+    )
 
     # Mock _get_redis_client to prevent actual connection attempt
     with patch.object(cache, "_get_redis_client") as mock_get_client:
@@ -53,15 +57,15 @@ async def test_sync_methods_warn_with_async_only_client(async_client, caplog):
             assert len(caplog.records) == 0
 
 
-def test_no_warning_with_sync_client(redis_url):
+def test_no_warning_with_sync_client(redis_url, redis_test_name):
     """Test that no warning is shown when sync client is provided."""
     # Create sync redis client from redis_url
     sync_client = Redis.from_url(redis_url)
 
+    cache = EmbeddingsCache(
+        name=redis_test_name("test_cache"), redis_client=sync_client
+    )
     try:
-        # Initialize EmbeddingsCache with sync_redis_client
-        cache = EmbeddingsCache(name="test_cache", redis_client=sync_client)
-
         with patch("redisvl.utils.log.get_logger") as mock_logger:
             # Sync methods should not warn
             _ = cache.get_by_key("test_key")
@@ -70,19 +74,27 @@ def test_no_warning_with_sync_client(redis_url):
             # No warnings should have been logged
             mock_logger.return_value.warning.assert_not_called()
     finally:
+        cache.clear()
         sync_client.close()
 
 
 @pytest.mark.asyncio
-async def test_async_methods_no_warning(async_client):
+async def test_async_methods_no_warning(async_client, redis_test_name):
     """Test that async methods don't trigger warnings."""
     # Initialize EmbeddingsCache with only async_redis_client
-    cache = EmbeddingsCache(name="test_cache", async_redis_client=async_client)
+    cache = EmbeddingsCache(
+        name=redis_test_name("test_cache"), async_redis_client=async_client
+    )
 
-    with patch("redisvl.utils.log.get_logger") as mock_logger:
-        # Async methods should not warn
-        _ = await cache.aget_by_key("test_key")
-        _ = await cache.aset(content="test", model_name="model", embedding=[0.1, 0.2])
+    try:
+        with patch("redisvl.utils.log.get_logger") as mock_logger:
+            # Async methods should not warn
+            _ = await cache.aget_by_key("test_key")
+            _ = await cache.aset(
+                content="test", model_name="model", embedding=[0.1, 0.2]
+            )
 
-        # No warnings should have been logged
-        mock_logger.return_value.warning.assert_not_called()
+            # No warnings should have been logged
+            mock_logger.return_value.warning.assert_not_called()
+    finally:
+        await cache.aclear()

tests/integration/test_hybrid.py

@@ -19,12 +19,12 @@
 
 
 @pytest.fixture
-def index_schema(worker_id):
+def index_schema(redis_test_name):
     return IndexSchema.from_dict(
         {
             "index": {
-                "name": f"user_index_{worker_id}",
-                "prefix": f"v1_{worker_id}",
+                "name": redis_test_name("user_index"),
+                "prefix": redis_test_name("v1"),
                 "storage_type": "hash",
             },
             "fields": [
@@ -73,8 +73,9 @@ def index_schema(worker_id):
 def index(index_schema, multi_vector_data, redis_url):
     index = SearchIndex(schema=index_schema, redis_url=redis_url)
 
-    # create the index (no data yet)
-    index.create(overwrite=True)
+    # create the index (no data yet); drop any stale docs left by an
+    # interrupted earlier run sharing this worker's Redis database
+    index.create(overwrite=True, drop=True)
 
     # prepare and load the data
     def hash_preprocess(item: dict) -> dict:
@@ -97,7 +98,7 @@ def hash_preprocess(item: dict) -> dict:
 @pytest.fixture
 async def async_index(index_schema, multi_vector_data, async_client):
     index = AsyncSearchIndex(schema=index_schema, redis_client=async_client)
-    await index.create(overwrite=True)
+    await index.create(overwrite=True, drop=True)
 
     def hash_preprocess(item: dict) -> dict:
         return {