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

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 {