Use worker ID in search index names

Author: abrookins

tests/conftest.py

@@ -10,23 +10,32 @@
 from redisvl.utils.vectorize import HFTextVectorizer
 
 
+@pytest.fixture(scope="session")
+def worker_id(request):
+    """
+    Get the worker ID for the current test.
+
+    In pytest-xdist, the config has "workerid" in workerinput.
+    This fixture abstracts that logic to provide a consistent worker_id
+    across all tests.
+    """
+    workerinput = getattr(request.config, "workerinput", {})
+    return workerinput.get("workerid", "master")
+
+
 @pytest.fixture(autouse=True)
 def set_tokenizers_parallelism():
     """Disable tokenizers parallelism in tests to avoid deadlocks"""
     os.environ["TOKENIZERS_PARALLELISM"] = "false"
 
 
 @pytest.fixture(scope="session", autouse=True)
-def redis_container(request):
+def redis_container(worker_id):
     """
     If using xdist, create a unique Compose project for each xdist worker by
     setting COMPOSE_PROJECT_NAME. That prevents collisions on container/volume
     names.
     """
-    # In xdist, the config has "workerid" in workerinput
-    workerinput = getattr(request.config, "workerinput", {})
-    worker_id = workerinput.get("workerid", "master")
-
     # Set the Compose project name so containers do not clash across workers
     os.environ["COMPOSE_PROJECT_NAME"] = f"redis_test_{worker_id}"
     os.environ.setdefault("REDIS_IMAGE", "redis/redis-stack-server:latest")
@@ -206,16 +215,17 @@ def pytest_collection_modifyitems(
 
 
 @pytest.fixture
-def flat_index(sample_data, redis_url):
+def flat_index(sample_data, redis_url, worker_id):
     """
     A fixture that uses the "flag" algorithm for its vector field.
     """
+
     # construct a search index from the schema
     index = SearchIndex.from_dict(
         {
             "index": {
-                "name": "user_index",
-                "prefix": "v1",
+                "name": f"user_index_{worker_id}",
+                "prefix": f"v1_{worker_id}",
                 "storage_type": "hash",
             },
             "fields": [
@@ -260,16 +270,17 @@ def hash_preprocess(item: dict) -> dict:
 
 
 @pytest.fixture
-async def async_flat_index(sample_data, redis_url):
+async def async_flat_index(sample_data, redis_url, worker_id):
     """
     A fixture that uses the "flag" algorithm for its vector field.
     """
+
     # construct a search index from the schema
     index = AsyncSearchIndex.from_dict(
         {
             "index": {
-                "name": "user_index",
-                "prefix": "v1",
+                "name": f"user_index_{worker_id}",
+                "prefix": f"v1_{worker_id}",
                 "storage_type": "hash",
             },
             "fields": [
@@ -314,15 +325,16 @@ def hash_preprocess(item: dict) -> dict:
 
 
 @pytest.fixture
-async def async_hnsw_index(sample_data, redis_url):
+async def async_hnsw_index(sample_data, redis_url, worker_id):
     """
     A fixture that uses the "hnsw" algorithm for its vector field.
     """
+
     index = AsyncSearchIndex.from_dict(
         {
             "index": {
-                "name": "user_index",
-                "prefix": "v1",
+                "name": f"user_index_{worker_id}",
+                "prefix": f"v1_{worker_id}",
                 "storage_type": "hash",
             },
             "fields": [
@@ -364,15 +376,16 @@ def hash_preprocess(item: dict) -> dict:
 
 
 @pytest.fixture
-def hnsw_index(sample_data, redis_url):
+def hnsw_index(sample_data, redis_url, worker_id):
     """
     A fixture that uses the "hnsw" algorithm for its vector field.
     """
+
     index = SearchIndex.from_dict(
         {
             "index": {
-                "name": "user_index",
-                "prefix": "v1",
+                "name": f"user_index_{worker_id}",
+                "prefix": f"v1_{worker_id}",
                 "storage_type": "hash",
             },
             "fields": [

tests/integration/test_aggregation.py

@@ -10,15 +10,11 @@
 
 
 @pytest.fixture
-def index(sample_data, redis_url, request):
-    # In xdist, the config has "workerid" in workerinput
-    workerinput = getattr(request.config, "workerinput", {})
-    worker_id = workerinput.get("workerid", "master")
-
+def index(sample_data, redis_url, worker_id):
     index = SearchIndex.from_dict(
         {
             "index": {
-                "name": "user_index",
+                "name": f"user_index_{worker_id}",
                 "prefix": f"v1_{worker_id}",
                 "storage_type": "hash",
             },

tests/integration/test_async_search_index.py

@@ -32,30 +32,24 @@ def async_index(index_schema, async_client):
 
 
 @pytest.fixture
-def async_index_from_dict(request):
-    # In xdist, the config has "workerid" in workerinput
-    workerinput = getattr(request.config, "workerinput", {})
-    worker_id = workerinput.get("workerid", "master")
+def async_index_from_dict(worker_id):
 
     return AsyncSearchIndex.from_dict(
-        {"index": {"name": "my_index", "prefix": f"rvl_{worker_id}"}, "fields": fields}
+        {
+            "index": {"name": f"my_index_{worker_id}", "prefix": f"rvl_{worker_id}"},
+            "fields": fields,
+        }
     )
 
 
 @pytest.fixture
-def async_index_from_yaml(request):
-    # In xdist, the config has "workerid" in workerinput
-    workerinput = getattr(request.config, "workerinput", {})
-    worker_id = workerinput.get("workerid", "master")
+def async_index_from_yaml(worker_id):
 
-    # Load the schema from YAML
-    schema = IndexSchema.from_yaml("schemas/test_json_schema.yaml")
-
-    # Modify the prefix to include the worker ID
-    schema.index.prefix = f"{schema.index.prefix}_{worker_id}"
-
-    # Create the AsyncSearchIndex with the modified schema
-    return AsyncSearchIndex(schema=schema)
+    index = AsyncSearchIndex.from_yaml("schemas/test_json_schema.yaml")
+    # Update the index name and prefix to include worker_id
+    index.schema.index.name = f"{index.schema.index.name}_{worker_id}"
+    index.schema.index.prefix = f"{index.schema.index.prefix}_{worker_id}"
+    return index
 
 
 def test_search_index_properties(index_schema, async_index):
@@ -73,7 +67,7 @@ def test_search_index_properties(index_schema, async_index):
 
 
 def test_search_index_from_yaml(async_index_from_yaml):
-    assert async_index_from_yaml.name == "json-test"
+    assert async_index_from_yaml.name.startswith("json-test")
     assert async_index_from_yaml.client is None
     assert async_index_from_yaml.prefix == "json"
     assert async_index_from_yaml.key_separator == ":"
@@ -82,7 +76,7 @@ def test_search_index_from_yaml(async_index_from_yaml):
 
 
 def test_search_index_from_dict(async_index_from_dict):
-    assert async_index_from_dict.name == "my_index"
+    assert async_index_from_dict.name.startswith("my_index")
     assert async_index_from_dict.client is None
     assert async_index_from_dict.prefix == "rvl"
     assert async_index_from_dict.key_separator == ":"

tests/integration/test_flow.py

@@ -42,7 +42,12 @@
 
 
 @pytest.mark.parametrize("schema", [hash_schema, json_schema])
-def test_simple(client, schema, sample_data):
+def test_simple(client, schema, sample_data, worker_id):
+    # Update schema with worker_id
+    schema = schema.copy()
+    schema["index"] = schema["index"].copy()
+    schema["index"]["name"] = f"{schema['index']['name']}_{worker_id}"
+    schema["index"]["prefix"] = f"{schema['index']['prefix']}_{worker_id}"
     index = SearchIndex.from_dict(schema, redis_client=client)
     # create the index
     index.create(overwrite=True, drop=True)

tests/integration/test_flow_async.py

@@ -46,7 +46,12 @@
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("schema", [hash_schema, json_schema])
-async def test_simple(async_client, schema, sample_data):
+async def test_simple(async_client, schema, sample_data, worker_id):
+    # Update schema with worker_id
+    schema = schema.copy()
+    schema["index"] = schema["index"].copy()
+    schema["index"]["name"] = f"{schema['index']['name']}_{worker_id}"
+    schema["index"]["prefix"] = f"{schema['index']['prefix']}_{worker_id}"
     index = AsyncSearchIndex.from_dict(schema, redis_client=async_client)
     # create the index
     await index.create(overwrite=True, drop=True)