Flaky test fixes (#328)

- Centralize HG vectorizers
- Use worker ID in search index names and keys

Author: abrookins

redisvl/extensions/cache/llm/schema.py

@@ -114,7 +114,6 @@ class SemanticCacheIndexSchema(IndexSchema):
 
     @classmethod
     def from_params(cls, name: str, prefix: str, vector_dims: int, dtype: str):
-
         return cls(
             index={"name": name, "prefix": prefix},  # type: ignore
             fields=[  # type: ignore

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")
@@ -80,6 +89,16 @@ def hf_vectorizer():
     )
 
 
+@pytest.fixture(scope="session")
+def hf_vectorizer_float16():
+    return HFTextVectorizer(dtype="float16")
+
+
+@pytest.fixture(scope="session")
+def hf_vectorizer_with_model():
+    return HFTextVectorizer("sentence-transformers/all-mpnet-base-v2")
+
+
 @pytest.fixture
 def sample_datetimes():
     return {
@@ -196,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": [
@@ -250,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": [
@@ -304,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": [
@@ -354,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,12 +10,12 @@
 
 
 @pytest.fixture
-def index(sample_data, redis_url):
+def index(sample_data, redis_url, worker_id):
     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_async_search_index.py

@@ -32,13 +32,24 @@ def async_index(index_schema, async_client):
 
 
 @pytest.fixture
-def async_index_from_dict():
-    return AsyncSearchIndex.from_dict({"index": {"name": "my_index"}, "fields": fields})
+def async_index_from_dict(worker_id):
+
+    return AsyncSearchIndex.from_dict(
+        {
+            "index": {"name": f"my_index_{worker_id}", "prefix": f"rvl_{worker_id}"},
+            "fields": fields,
+        }
+    )
 
 
 @pytest.fixture
-def async_index_from_yaml():
-    return AsyncSearchIndex.from_yaml("schemas/test_json_schema.yaml")
+def async_index_from_yaml(worker_id):
+
+    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):
@@ -56,18 +67,18 @@ 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.prefix.startswith("json_")
     assert async_index_from_yaml.key_separator == ":"
     assert async_index_from_yaml.storage_type == StorageType.JSON
     assert async_index_from_yaml.key("foo").startswith(async_index_from_yaml.prefix)
 
 
 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.prefix.startswith("rvl_")
     assert async_index_from_dict.key_separator == ":"
     assert async_index_from_dict.storage_type == StorageType.HASH
     assert len(async_index_from_dict.schema.fields) == len(fields)

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)