diff --git a/tests/conftest.py b/tests/conftest.py
index c6f8f903..c3ac61ba 100644
--- a/tests/conftest.py
+++ b/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:
diff --git a/tests/integration/test_aggregation.py b/tests/integration/test_aggregation.py
index fe26839b..02f051ec 100644
--- a/tests/integration/test_aggregation.py
+++ b/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:
diff --git a/tests/integration/test_async_search_index.py b/tests/integration/test_async_search_index.py
index c3ca3774..01e30a92 100644
--- a/tests/integration/test_async_search_index.py
+++ b/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
diff --git a/tests/integration/test_cluster_pipelining.py b/tests/integration/test_cluster_pipelining.py
index 0c09db34..0a545487 100644
--- a/tests/integration/test_cluster_pipelining.py
+++ b/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)
diff --git a/tests/integration/test_embedcache_warnings.py b/tests/integration/test_embedcache_warnings.py
index de2f9b18..13823dd5 100644
--- a/tests/integration/test_embedcache_warnings.py
+++ b/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()
diff --git a/tests/integration/test_hybrid.py b/tests/integration/test_hybrid.py
index 95f9c481..05bc66a7 100644
--- a/tests/integration/test_hybrid.py
+++ b/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 {
diff --git a/tests/integration/test_llmcache.py b/tests/integration/test_llmcache.py
index 39ec674a..9673581b 100644
--- a/tests/integration/test_llmcache.py
+++ b/tests/integration/test_llmcache.py
@@ -2,6 +2,7 @@
 import sys
 import warnings
 from collections import namedtuple
+from contextlib import suppress
 from time import sleep, time
 
 import pytest
@@ -963,27 +964,32 @@ def test_no_key_collision_on_identical_prompts(redis_url, worker_id, hf_vectoriz
     assert len(filtered_results) == 2
 
 
-def test_create_cache_with_different_vector_types(client, worker_id, redis_url):
+def test_create_cache_with_different_vector_types(client, redis_test_name, redis_url):
     skip_if_no_redis_search(client)
+    caches = []
     try:
         bfloat_cache = SemanticCache(
-            name=f"bfloat_cache_{worker_id}", dtype="bfloat16", redis_url=redis_url
+            name=redis_test_name("bfloat_cache"), dtype="bfloat16", redis_url=redis_url
         )
+        caches.append(bfloat_cache)
         bfloat_cache.store("bfloat16 prompt", "bfloat16 response")
 
         float16_cache = SemanticCache(
-            name=f"float16_cache_{worker_id}", dtype="float16", redis_url=redis_url
+            name=redis_test_name("float16_cache"), dtype="float16", redis_url=redis_url
         )
+        caches.append(float16_cache)
         float16_cache.store("float16 prompt", "float16 response")
 
         float32_cache = SemanticCache(
-            name=f"float32_cache_{worker_id}", dtype="float32", redis_url=redis_url
+            name=redis_test_name("float32_cache"), dtype="float32", redis_url=redis_url
         )
+        caches.append(float32_cache)
         float32_cache.store("float32 prompt", "float32 response")
 
         float64_cache = SemanticCache(
-            name=f"float64_cache_{worker_id}", dtype="float64", redis_url=redis_url
+            name=redis_test_name("float64_cache"), dtype="float64", redis_url=redis_url
         )
+        caches.append(float64_cache)
         float64_cache.store("float64 prompt", "float64 response")
 
         for cache in [bfloat_cache, float16_cache, float32_cache, float64_cache]:
@@ -991,9 +997,13 @@ def test_create_cache_with_different_vector_types(client, worker_id, redis_url):
             assert len(cache.check("float prompt", num_results=5)) == 1
     except:
         pytest.skip("Required Redis modules not available or version too low")
+    finally:
+        for cache in caches:
+            with suppress(Exception):
+                cache.delete()
 
 
-def test_bad_dtype_connecting_to_existing_cache(client, redis_url, worker_id):
+def test_bad_dtype_connecting_to_existing_cache(client, redis_url, redis_test_name):
     skip_if_no_redis_search(client)
     # Skip this test for Redis 6.2.x as FT.INFO doesn't return dims properly
     redis_version = client.info()["redis_version"]
@@ -1002,24 +1012,26 @@ def test_bad_dtype_connecting_to_existing_cache(client, redis_url, worker_id):
             "Redis 6.2.x FT.INFO doesn't properly return vector dims for reconnection"
         )
 
+    cache_name = redis_test_name("float64_cache")
+
     def create_cache():
-        return SemanticCache(
-            name=f"float64_cache_{worker_id}", dtype="float64", redis_url=redis_url
-        )
+        return SemanticCache(name=cache_name, dtype="float64", redis_url=redis_url)
 
     def create_same_type():
-        return SemanticCache(
-            name=f"float64_cache_{worker_id}", dtype="float64", redis_url=redis_url
-        )
+        return SemanticCache(name=cache_name, dtype="float64", redis_url=redis_url)
 
     cache = create_cache()
-    same_type = create_same_type()
-    # under the hood uses from_existing
-
-    with pytest.raises(ValueError):
-        bad_type = SemanticCache(
-            name=f"float64_cache_{worker_id}", dtype="float16", redis_url=redis_url
-        )
+    try:
+        same_type = create_same_type()
+        # under the hood uses from_existing
+
+        with pytest.raises(ValueError):
+            bad_type = SemanticCache(
+                name=cache_name, dtype="float16", redis_url=redis_url
+            )
+    finally:
+        with suppress(Exception):
+            cache.delete()
 
 
 def test_vectorizer_dtype_mismatch(redis_url, hf_vectorizer_float16, worker_id):
diff --git a/tests/integration/test_no_proactive_module_checks.py b/tests/integration/test_no_proactive_module_checks.py
index 024dff67..79175c23 100644
--- a/tests/integration/test_no_proactive_module_checks.py
+++ b/tests/integration/test_no_proactive_module_checks.py
@@ -28,11 +28,15 @@
 
 
 @pytest.fixture
-def sample_schema():
+def sample_schema(redis_test_name):
     """Create a sample index schema for testing."""
     return IndexSchema.from_dict(
         {
-            "index": {"name": "test-index", "prefix": "doc", "storage_type": "hash"},
+            "index": {
+                "name": redis_test_name("test-index"),
+                "prefix": redis_test_name("doc"),
+                "storage_type": "hash",
+            },
             "fields": [
                 {"name": "text", "type": "text"},
                 {
@@ -146,11 +150,12 @@ def test_search_index_create_with_modules(self, client, sample_schema, worker_id
             # MODULE LIST should NOT have been called
             mock_module_list.assert_not_called()
 
-        # Verify index exists
-        assert index.exists()
-
-        # Clean up
-        index.delete()
+        try:
+            # Verify index exists
+            assert index.exists()
+        finally:
+            # Clean up
+            index.delete(drop=True)
 
     async def test_async_search_index_create_with_modules(
         self, async_client, sample_schema, worker_id
@@ -174,11 +179,12 @@ async def test_async_search_index_create_with_modules(
             # MODULE LIST should NOT have been called
             mock_module_list.assert_not_called()
 
-        # Verify index exists
-        assert await index.exists()
-
-        # Clean up
-        await index.delete()
+        try:
+            # Verify index exists
+            assert await index.exists()
+        finally:
+            # Clean up
+            await index.delete(drop=True)
 
     def test_search_operations_fail_gracefully_without_modules(self):
         """Test that operations fail with clear errors when modules are missing."""
@@ -326,22 +332,24 @@ async def test_async_cleanup_without_validation(self, redis_url):
         for client in clients:
             await client.aclose()
 
-    def test_from_existing_index_no_validation(self, client, worker_id):
+    def test_from_existing_index_no_validation(self, client, redis_test_name):
         """Test that SearchIndex.from_existing doesn't validate modules."""
         # Skip if Redis Search is not available
         skip_if_no_redis_search(client)
 
+        index_name = redis_test_name("existing-index")
+
         # First create an index normally
         schema = IndexSchema.from_dict(
             {
-                "index": {"name": f"existing-index-{worker_id}", "prefix": "doc"},
+                "index": {"name": index_name, "prefix": redis_test_name("doc")},
                 "fields": [{"name": "text", "type": "text"}],
             }
         )
         index = SearchIndex(schema, redis_client=client)
 
         try:
-            index.create(overwrite=True)
+            index.create(overwrite=True, drop=True)
         except ResponseError as e:
             if "unknown command" in str(e).lower():
                 pytest.skip("Redis Search module not available")
@@ -352,16 +360,16 @@ def test_from_existing_index_no_validation(self, client, worker_id):
             with patch.object(client, "module_list") as mock_module_list:
                 # Load from existing should work without MODULE LIST
                 existing_index = SearchIndex.from_existing(
-                    f"existing-index-{worker_id}", redis_client=client
+                    index_name, redis_client=client
                 )
 
                 # MODULE LIST should NOT have been called
                 mock_module_list.assert_not_called()
 
-                assert existing_index.name == f"existing-index-{worker_id}"
+                assert existing_index.name == index_name
         finally:
             # Clean up
             try:
-                index.delete()
+                index.delete(drop=True)
             except:
                 pass  # Ignore cleanup errors
diff --git a/tests/integration/test_redis_cluster_support.py b/tests/integration/test_redis_cluster_support.py
index 0d18dea3..fc01922c 100644
--- a/tests/integration/test_redis_cluster_support.py
+++ b/tests/integration/test_redis_cluster_support.py
@@ -48,12 +48,15 @@ async def test_sync_to_async_conversion_rejects_cluster_client(redis_cluster_url
 
 
 @pytest.mark.requires_cluster
-def test_search_index_cluster_client(redis_cluster_url):
+def test_search_index_cluster_client(redis_cluster_url, redis_test_name):
     """Test that SearchIndex correctly accepts RedisCluster clients."""
     # Create a simple schema
     schema = IndexSchema.from_dict(
         {
-            "index": {"name": "test_cluster_index", "prefix": "test_cluster"},
+            "index": {
+                "name": redis_test_name("test_cluster_index"),
+                "prefix": redis_test_name("test_cluster"),
+            },
             "fields": [
                 {"name": "name", "type": "text"},
                 {"name": "age", "type": "numeric"},
@@ -63,50 +66,54 @@ def test_search_index_cluster_client(redis_cluster_url):
 
     cluster_client = RedisCluster.from_url(redis_cluster_url)
     index = SearchIndex(schema=schema, redis_client=cluster_client)
-    index.create(overwrite=True)
-    index.load([{"name": "test1", "age": 30}])
-    results = index.query(TextQuery("test1", "name"))
-    assert results[0]["name"] == "test1"
-    index.delete(drop=True)
+    try:
+        index.create(overwrite=True, drop=True)
+        index.load([{"name": "test1", "age": 30}])
+        results = index.query(TextQuery("test1", "name"))
+        assert results[0]["name"] == "test1"
+    finally:
+        index.delete(drop=True)
 
 
 @pytest.mark.requires_cluster
-def test_search_index_cluster_info(redis_cluster_url):
+def test_search_index_cluster_info(redis_cluster_url, redis_test_name):
     """Test .info() method on SearchIndex with RedisCluster client."""
+    index_name = redis_test_name("test_cluster_info")
     schema = IndexSchema.from_dict(
         {
-            "index": {"name": "test_cluster_info", "prefix": "test_info"},
+            "index": {"name": index_name, "prefix": redis_test_name("test_info")},
             "fields": [{"name": "name", "type": "text"}],
         }
     )
     client = RedisCluster.from_url(redis_cluster_url)
     index = SearchIndex(schema=schema, redis_client=client)
     try:
-        index.create(overwrite=True)
+        index.create(overwrite=True, drop=True)
         info = index.info()
         assert isinstance(info, dict)
-        assert info.get("index_name", None) == "test_cluster_info"
+        assert info.get("index_name", None) == index_name
     finally:
         index.delete(drop=True)
 
 
 @pytest.mark.requires_cluster
 @pytest.mark.asyncio
-async def test_async_search_index_cluster_info(redis_cluster_url):
+async def test_async_search_index_cluster_info(redis_cluster_url, redis_test_name):
     """Test .info() method on AsyncSearchIndex with AsyncRedisCluster client."""
+    index_name = redis_test_name("async_cluster_info")
     schema = IndexSchema.from_dict(
         {
-            "index": {"name": "async_cluster_info", "prefix": "async_info"},
+            "index": {"name": index_name, "prefix": redis_test_name("async_info")},
             "fields": [{"name": "name", "type": "text"}],
         }
     )
     client = AsyncRedisCluster.from_url(redis_cluster_url)
     index = AsyncSearchIndex(schema=schema, redis_client=client)
     try:
-        await index.create(overwrite=True)
+        await index.create(overwrite=True, drop=True)
         info = await index.info()
         assert isinstance(info, dict)
-        assert info.get("index_name", None) == "async_cluster_info"
+        assert info.get("index_name", None) == index_name
     finally:
         await index.delete(drop=True)
         await client.aclose()
@@ -114,12 +121,15 @@ async def test_async_search_index_cluster_info(redis_cluster_url):
 
 @pytest.mark.requires_cluster
 @pytest.mark.asyncio
-async def test_async_search_index_client(redis_cluster_url):
+async def test_async_search_index_client(redis_cluster_url, redis_test_name):
     """Test that AsyncSearchIndex correctly handles AsyncRedis clients."""
     # Create a simple schema
     schema = IndexSchema.from_dict(
         {
-            "index": {"name": "async_test_index", "prefix": "async_test"},
+            "index": {
+                "name": redis_test_name("async_test_index"),
+                "prefix": redis_test_name("async_test"),
+            },
             "fields": [
                 {"name": "name", "type": "text"},
                 {"name": "age", "type": "numeric"},
@@ -131,7 +141,7 @@ async def test_async_search_index_client(redis_cluster_url):
     cluster_client = AsyncRedisCluster.from_url(redis_cluster_url)
     index = AsyncSearchIndex(schema=schema, redis_client=cluster_client)
     try:
-        await index.create(overwrite=True)
+        await index.create(overwrite=True, drop=True)
         await index.load([{"name": "async_test", "age": 25}])
         results = await index.query(TextQuery("async_test", "name"))
         assert results[0]["name"] == "async_test"
@@ -143,12 +153,14 @@ async def test_async_search_index_client(redis_cluster_url):
 
 @pytest.mark.requires_cluster
 @pytest.mark.asyncio
-async def test_embeddings_cache_cluster_async(redis_cluster_url):
+async def test_embeddings_cache_cluster_async(redis_cluster_url, redis_test_name):
     """Test that EmbeddingsCache correctly handles AsyncRedisCluster clients."""
     cluster_client = RedisConnectionFactory.get_async_redis_cluster_connection(
         redis_cluster_url
     )
-    cache = EmbeddingsCache(async_redis_client=cluster_client)
+    cache = EmbeddingsCache(
+        name=redis_test_name("embedcache"), async_redis_client=cluster_client
+    )
 
     try:
         await cache.aset(
@@ -166,10 +178,12 @@ async def test_embeddings_cache_cluster_async(redis_cluster_url):
 
 
 @pytest.mark.requires_cluster
-def test_embeddings_cache_cluster_sync(redis_cluster_url):
+def test_embeddings_cache_cluster_sync(redis_cluster_url, redis_test_name):
     """Test that EmbeddingsCache correctly handles RedisCluster clients."""
     cluster_client = RedisCluster.from_url(redis_cluster_url)
-    cache = EmbeddingsCache(redis_client=cluster_client)
+    cache = EmbeddingsCache(
+        name=redis_test_name("embedcache"), redis_client=cluster_client
+    )
 
     for i in range(100):
         cache.set(
@@ -197,7 +211,9 @@ def test_embeddings_cache_cluster_sync(redis_cluster_url):
 
 
 @pytest.mark.requires_cluster
-def test_semantic_router_cluster_client(redis_cluster_url, hf_vectorizer):
+def test_semantic_router_cluster_client(
+    redis_cluster_url, hf_vectorizer, redis_test_name
+):
     """Test that SemanticRouter works correctly with RedisCluster clients."""
     routes = [
         Route(
@@ -214,7 +230,7 @@ def test_semantic_router_cluster_client(redis_cluster_url, hf_vectorizer):
     ]
     client = RedisCluster.from_url(redis_cluster_url)
 
-    router_name = "test_cluster_router"
+    router_name = redis_test_name("test_cluster_router")
     router = SemanticRouter(
         name=router_name,
         routes=routes,
diff --git a/tests/integration/test_search_index.py b/tests/integration/test_search_index.py
index aca97d41..e66c625b 100644
--- a/tests/integration/test_search_index.py
+++ b/tests/integration/test_search_index.py
@@ -114,11 +114,11 @@ def test_search_index_from_existing_url(redis_url, index):
     assert index2.schema == index.schema
 
 
-def test_search_index_from_existing_complex(client):
+def test_search_index_from_existing_complex(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": [
@@ -144,29 +144,32 @@ def test_search_index_from_existing_complex(client):
         ],
     }
     index = SearchIndex.from_dict(schema, redis_client=client)
-    index.create(overwrite=True)
+    index.create(overwrite=True, drop=True)
 
     try:
-        index2 = SearchIndex.from_existing(index.name, redis_client=client)
-    except Exception as e:
-        pytest.skip(str(e))
-
-    # Verify index metadata matches
-    assert index2.schema.index.name == index.schema.index.name
-    assert index2.schema.index.prefix == index.schema.index.prefix
-    assert index2.schema.index.storage_type == index.schema.index.storage_type
-
-    # Verify non-vector fields are present
-    for field_name in ["user", "credit_score", "job", "age"]:
-        assert field_name in index2.schema.fields
-        assert (
-            index2.schema.fields[field_name].type
-            == index.schema.fields[field_name].type
-        )
-
-    # Vector field may not be present on older Redis versions
-    if "user_embedding" in index2.schema.fields:
-        assert index2.schema.fields["user_embedding"].type == "vector"
+        try:
+            index2 = SearchIndex.from_existing(index.name, redis_client=client)
+        except Exception as e:
+            pytest.skip(str(e))
+
+        # Verify index metadata matches
+        assert index2.schema.index.name == index.schema.index.name
+        assert index2.schema.index.prefix == index.schema.index.prefix
+        assert index2.schema.index.storage_type == index.schema.index.storage_type
+
+        # Verify non-vector fields are present
+        for field_name in ["user", "credit_score", "job", "age"]:
+            assert field_name in index2.schema.fields
+            assert (
+                index2.schema.fields[field_name].type
+                == index.schema.fields[field_name].type
+            )
+
+        # Vector field may not be present on older Redis versions
+        if "user_embedding" in index2.schema.fields:
+            assert index2.schema.fields["user_embedding"].type == "vector"
+    finally:
+        index.delete(drop=True)
 
 
 def test_search_index_from_existing_multiple_prefixes(client, redis_test_name):
@@ -510,7 +513,7 @@ def test_search_index_that_owns_client_disconnect(index_schema, redis_url):
     assert index.client is None
 
 
-def test_search_index_no_proactive_module_validation(redis_url):
+def test_search_index_no_proactive_module_validation(redis_url, redis_test_name):
     """
     Updated test for issue #370: SearchIndex should not validate modules proactively.
     Operations should fail naturally if modules are missing.
@@ -522,7 +525,7 @@ def test_search_index_no_proactive_module_validation(redis_url):
         # Create index - validation should only set lib name, not check modules
         index = SearchIndex(
             schema=IndexSchema.from_dict(
-                {"index": {"name": "my_index"}, "fields": fields}
+                {"index": {"name": redis_test_name("my_index")}, "fields": fields}
             ),
             redis_client=client,
         )
@@ -534,8 +537,11 @@ def test_search_index_no_proactive_module_validation(redis_url):
         # The actual operation (create) will succeed if modules are present
         index.create(overwrite=True, drop=True)
 
-        # Verify index was created successfully (modules are present in test env)
-        assert index.exists()
+        try:
+            # Verify index was created successfully (modules are present in test env)
+            assert index.exists()
+        finally:
+            index.delete(drop=True)
 
 
 def test_batch_search(index):
diff --git a/tests/integration/test_search_results.py b/tests/integration/test_search_results.py
index f75d83b9..a9e9e9a4 100644
--- a/tests/integration/test_search_results.py
+++ b/tests/integration/test_search_results.py
@@ -16,7 +16,7 @@ def filter_query():
 
 
 @pytest.fixture
-def index(sample_data, redis_url, worker_id):
+def index(sample_data, redis_url, redis_test_name):
     fields_spec = [
         {"name": "credit_score", "type": "tag"},
         {"name": "user", "type": "tag"},
@@ -36,8 +36,8 @@ def index(sample_data, redis_url, worker_id):
 
     json_schema = {
         "index": {
-            "name": f"user_index_json_{worker_id}",
-            "prefix": f"users_json_{worker_id}",
+            "name": redis_test_name("user_index_json"),
+            "prefix": redis_test_name("users_json"),
             "storage_type": "json",
         },
         "fields": fields_spec,
@@ -46,8 +46,9 @@ def index(sample_data, redis_url, worker_id):
     # construct a search index from the schema
     index = SearchIndex.from_dict(json_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
     index.load(sample_data)
diff --git a/tests/integration/test_svs_integration.py b/tests/integration/test_svs_integration.py
index aae68797..5d2ac820 100644
--- a/tests/integration/test_svs_integration.py
+++ b/tests/integration/test_svs_integration.py
@@ -38,13 +38,13 @@
 
 
 @pytest.fixture
-def svs_schema_lvq(worker_id):
+def svs_schema_lvq(redis_test_name):
     """Create SVS-VAMANA schema with LVQ compression."""
     return IndexSchema.from_dict(
         {
             "index": {
-                "name": f"svs_lvq_{worker_id}",
-                "prefix": f"svs_lvq_{worker_id}",
+                "name": redis_test_name("svs_lvq"),
+                "prefix": redis_test_name("svs_lvq"),
             },
             "fields": [
                 {"name": "id", "type": "tag"},
@@ -69,13 +69,13 @@ def svs_schema_lvq(worker_id):
 
 
 @pytest.fixture
-def svs_schema_leanvec(worker_id):
+def svs_schema_leanvec(redis_test_name):
     """Create SVS-VAMANA schema with LeanVec compression and dimensionality reduction."""
     return IndexSchema.from_dict(
         {
             "index": {
-                "name": f"svs_leanvec_{worker_id}",
-                "prefix": f"svs_leanvec_{worker_id}",
+                "name": redis_test_name("svs_leanvec"),
+                "prefix": redis_test_name("svs_leanvec"),
             },
             "fields": [
                 {"name": "id", "type": "tag"},
@@ -104,7 +104,7 @@ def svs_schema_leanvec(worker_id):
 def svs_index_lvq(svs_schema_lvq, client):
     """Create SVS-VAMANA index with LVQ compression."""
     index = SearchIndex(schema=svs_schema_lvq, redis_client=client)
-    index.create(overwrite=True)
+    index.create(overwrite=True, drop=True)
     yield index
     index.delete(drop=True)
 
@@ -113,7 +113,7 @@ def svs_index_lvq(svs_schema_lvq, client):
 def svs_index_leanvec(svs_schema_leanvec, client):
     """Create SVS-VAMANA index with LeanVec compression."""
     index = SearchIndex(schema=svs_schema_leanvec, redis_client=client)
-    index.create(overwrite=True)
+    index.create(overwrite=True, drop=True)
     yield index
     index.delete(drop=True)
 
@@ -164,7 +164,7 @@ def test_create_svs_index_leanvec(self, svs_index_leanvec):
         info = svs_index_leanvec.info()
         assert info["num_docs"] == 0
 
-    def test_create_svs_with_compression_advisor(self, client, worker_id):
+    def test_create_svs_with_compression_advisor(self, client, redis_test_name):
         """Test creating SVS-VAMANA index using CompressionAdvisor."""
         dims = 768
         config = CompressionAdvisor.recommend(dims=dims, priority="balanced")
@@ -172,8 +172,8 @@ def test_create_svs_with_compression_advisor(self, client, worker_id):
         schema = IndexSchema.from_dict(
             {
                 "index": {
-                    "name": f"svs_advisor_{worker_id}",
-                    "prefix": f"svs_advisor_{worker_id}",
+                    "name": redis_test_name("svs_advisor"),
+                    "prefix": redis_test_name("svs_advisor"),
                 },
                 "fields": [
                     {"name": "id", "type": "tag"},
@@ -191,7 +191,7 @@ def test_create_svs_with_compression_advisor(self, client, worker_id):
         )
 
         index = SearchIndex(schema=schema, redis_client=client)
-        index.create(overwrite=True)
+        index.create(overwrite=True, drop=True)
 
         try:
             assert index.exists()
@@ -381,13 +381,15 @@ class TestSVSCompressionTypes:
             ("LVQ8", 384, "float32"),
         ],
     )
-    def test_lvq_compression_types(self, client, worker_id, compression, dims, dtype):
+    def test_lvq_compression_types(
+        self, client, redis_test_name, compression, dims, dtype
+    ):
         """Test various LVQ compression types."""
         schema = IndexSchema.from_dict(
             {
                 "index": {
-                    "name": f"svs_{compression.lower()}_{worker_id}",
-                    "prefix": f"svs_{compression.lower()}_{worker_id}",
+                    "name": redis_test_name(f"svs_{compression.lower()}"),
+                    "prefix": redis_test_name(f"svs_{compression.lower()}"),
                 },
                 "fields": [
                     {"name": "id", "type": "tag"},
@@ -407,7 +409,7 @@ def test_lvq_compression_types(self, client, worker_id, compression, dims, dtype
         )
 
         index = SearchIndex(schema=schema, redis_client=client)
-        index.create(overwrite=True)
+        index.create(overwrite=True, drop=True)
 
         try:
             # Load data
@@ -441,14 +443,14 @@ def test_lvq_compression_types(self, client, worker_id, compression, dims, dtype
         ],
     )
     def test_leanvec_compression_types(
-        self, client, worker_id, compression, dims, reduce, dtype
+        self, client, redis_test_name, compression, dims, reduce, dtype
     ):
         """Test various LeanVec compression types with dimensionality reduction."""
         schema = IndexSchema.from_dict(
             {
                 "index": {
-                    "name": f"svs_{compression.lower()}_{reduce}_{worker_id}",
-                    "prefix": f"svs_{compression.lower()}_{reduce}_{worker_id}",
+                    "name": redis_test_name(f"svs_{compression.lower()}_{reduce}"),
+                    "prefix": redis_test_name(f"svs_{compression.lower()}_{reduce}"),
                 },
                 "fields": [
                     {"name": "id", "type": "tag"},
@@ -469,7 +471,7 @@ def test_leanvec_compression_types(
         )
 
         index = SearchIndex(schema=schema, redis_client=client)
-        index.create(overwrite=True)
+        index.create(overwrite=True, drop=True)
 
         try:
             # Load data