diff --git a/.github/workflows/auto-release.yml b/.github/workflows/auto-release.yml index 6b89c2590..8490367ab 100644 --- a/.github/workflows/auto-release.yml +++ b/.github/workflows/auto-release.yml @@ -15,7 +15,7 @@ concurrency: cancel-in-progress: false env: - PYTHON_VERSION: "3.11" + PYTHON_VERSION: "3.12" UV_VERSION: "0.13.1" AUTO_VERSION: "11.3.6" RELEASE_BOT_NAME: "applied-ai-releases[bot]" diff --git a/.github/workflows/test-fork-pr.yml b/.github/workflows/test-fork-pr.yml index 715102084..4fff7ef94 100644 --- a/.github/workflows/test-fork-pr.yml +++ b/.github/workflows/test-fork-pr.yml @@ -14,7 +14,7 @@ permissions: pull-requests: read env: - PYTHON_VERSION: "3.11" + PYTHON_VERSION: "3.12" UV_VERSION: "0.13.1" jobs: diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 778323835..d48da204d 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -13,7 +13,7 @@ on: workflow_dispatch: env: - PYTHON_VERSION: "3.11" + PYTHON_VERSION: "3.12" UV_VERSION: "0.13.1" jobs: diff --git a/.readthedocs.yaml b/.readthedocs.yaml index e526ec65f..c5e59c278 100644 --- a/.readthedocs.yaml +++ b/.readthedocs.yaml @@ -7,7 +7,7 @@ version: 2 build: os: ubuntu-24.04 tools: - python: "3.11" + python: "3.12" jobs: # 1️⃣ Install uv once before we create the virtual-env diff --git a/docs/user_guide/how_to_guides/mcp.md b/docs/user_guide/how_to_guides/mcp.md index 6afd2282d..9b37fd598 100644 --- a/docs/user_guide/how_to_guides/mcp.md +++ b/docs/user_guide/how_to_guides/mcp.md @@ -195,6 +195,7 @@ indexes: - `runtime.default_embed_text_field` is only required when the server should generate embeddings during upsert - `vectorizer` is required for query embedding and server-side embedding, but optional for fulltext-only configs - `runtime.max_result_window` caps deep paging by limiting the maximum `offset + limit` +- `runtime.startup_timeout_seconds` bounds three startup steps for the binding: connecting to Redis together with the first `INFO`, which share one deadline, then reading the index, then building the vectorizer - `schema_overrides` is only for patching incomplete field attrs discovered from Redis ### Fulltext-Only Config diff --git a/docs/user_guide/how_to_guides/migrate-indexes.md b/docs/user_guide/how_to_guides/migrate-indexes.md index f9cd2df52..1a4e2a64c 100644 --- a/docs/user_guide/how_to_guides/migrate-indexes.md +++ b/docs/user_guide/how_to_guides/migrate-indexes.md @@ -887,7 +887,10 @@ backup is written *before* the index is dropped, so a crash at any point leaves a complete backup on disk. In multi-worker mode, dump and quantize are fused (each worker reads, backs up, and converts its shard in one pass *after* the index drop). A crash during this fused phase may leave partial -backup shards. Re-running detects and resumes from partial state. +backup shards. Re-running detects and resumes from partial state. With +`--async` (`AsyncMigrationExecutor`), the first worker to fail cancels the +others; the sync executor lets them finish their shards. Either way, no shard +is still being written when the migration reports the failure. ``` ### Rollback diff --git a/redisvl/extensions/message_history/schema.py b/redisvl/extensions/message_history/schema.py index 22aa22143..9c80587ec 100644 --- a/redisvl/extensions/message_history/schema.py +++ b/redisvl/extensions/message_history/schema.py @@ -1,5 +1,5 @@ import warnings -from enum import Enum +from enum import StrEnum from typing import Any from uuid import uuid4 @@ -20,7 +20,7 @@ from redisvl.utils.utils import current_timestamp -class ChatRole(str, Enum): +class ChatRole(StrEnum): """Enumeration of valid roles for a chat message.""" USER = "user" diff --git a/redisvl/mcp/errors.py b/redisvl/mcp/errors.py index aae49576b..ef7790765 100644 --- a/redisvl/mcp/errors.py +++ b/redisvl/mcp/errors.py @@ -1,5 +1,5 @@ import asyncio -from enum import Enum +from enum import StrEnum from typing import Any from pydantic import ValidationError @@ -8,7 +8,7 @@ from redisvl.exceptions import RedisSearchError -class MCPErrorCode(str, Enum): +class MCPErrorCode(StrEnum): """Stable internal error codes exposed by the MCP framework.""" INVALID_REQUEST = "invalid_request" diff --git a/redisvl/mcp/server.py b/redisvl/mcp/server.py index 4568482ce..66130920c 100644 --- a/redisvl/mcp/server.py +++ b/redisvl/mcp/server.py @@ -979,13 +979,20 @@ async def _connect_redis_client(self, timeout: int) -> Any: if self.config is None: raise RuntimeError("MCP server config not loaded") - client = await asyncio.wait_for( - RedisConnectionFactory._get_aredis_connection( - redis_url=self.config.server.redis_url - ), - timeout=timeout, - ) - await asyncio.wait_for(client.info("server"), timeout=timeout) + # Connecting and the first INFO share one deadline, so a slow server + # fails startup within ``timeout`` rather than up to twice that. + client: Any = None + try: + async with asyncio.timeout(timeout): + client = await RedisConnectionFactory._get_aredis_connection( + redis_url=self.config.server.redis_url + ) + await client.info("server") + except BaseException: + # The caller only takes ownership of a client that answered INFO. + if client is not None: + await client.aclose() + raise return client async def _load_effective_schema( diff --git a/redisvl/mcp/tools/list_indexes.py b/redisvl/mcp/tools/list_indexes.py index a53609383..5a4dc4324 100644 --- a/redisvl/mcp/tools/list_indexes.py +++ b/redisvl/mcp/tools/list_indexes.py @@ -27,7 +27,7 @@ def _binding_fields(binding_runtime: BindingRuntime) -> list[dict[str, str]]: embed_source = binding_runtime.binding.runtime.default_embed_text_field fields: list[dict[str, str]] = [] for field in binding_runtime.schema.fields.values(): - field_type = str(getattr(field.type, "value", field.type)) + field_type = str(field.type) if field_type.lower() == "vector": continue if field.name == embed_source: diff --git a/redisvl/mcp/tools/search.py b/redisvl/mcp/tools/search.py index 38ab4076a..556521b08 100644 --- a/redisvl/mcp/tools/search.py +++ b/redisvl/mcp/tools/search.py @@ -40,7 +40,7 @@ def _build_filter_hint( that the model never sees the knob. """ filter_fields = [ - f"{field.name}({getattr(field.type, 'value', field.type)})" + f"{field.name}({field.type})" for field in schema.fields.values() if field.type in _DSL_FILTER_FIELD_TYPES and field.name not in exclude ] diff --git a/redisvl/mcp/tools/upsert.py b/redisvl/mcp/tools/upsert.py index f079e923c..a34041d02 100644 --- a/redisvl/mcp/tools/upsert.py +++ b/redisvl/mcp/tools/upsert.py @@ -184,8 +184,7 @@ async def _embed_many(vectorizer: Any, contents: list[str]) -> list[list[float]] def _vector_dtype(rt: Any) -> str: """Resolve the binding's vector field datatype as a lowercase string.""" field = rt.binding.get_vector_field(rt.schema) - datatype = getattr(field.attrs.datatype, "value", field.attrs.datatype) - return str(datatype).lower() + return str(field.attrs.datatype).lower() def _validation_schema_for_record( diff --git a/redisvl/migration/async_executor.py b/redisvl/migration/async_executor.py index 24237656a..a5fc19776 100644 --- a/redisvl/migration/async_executor.py +++ b/redisvl/migration/async_executor.py @@ -564,8 +564,9 @@ async def apply( Async counterpart of :meth:`MigrationExecutor.apply`. Uses ``await`` for Redis I/O so the event loop remains responsive during - large quantization jobs. Multi-worker quantization uses - ``asyncio.gather`` with independent connections. + large quantization jobs. Multi-worker quantization runs in an + ``asyncio.TaskGroup`` with independent connections, so one worker's + failure cancels the rest. Args: plan: The migration plan to apply (from diff --git a/redisvl/migration/quantize.py b/redisvl/migration/quantize.py index 02de574ed..a35b83cff 100644 --- a/redisvl/migration/quantize.py +++ b/redisvl/migration/quantize.py @@ -4,7 +4,7 @@ the per-key HGET loop with batched pipeline operations. Also provides multi-worker orchestration for parallel quantization -using ThreadPoolExecutor (sync) or asyncio.gather (async). +using ThreadPoolExecutor (sync) or asyncio.TaskGroup (async). """ import hashlib @@ -538,9 +538,19 @@ async def async_multi_worker_quantize( progress_callback: Optional[Callable[[str, int, int], None]] = None, worker_backup_paths: Optional[List[str]] = None, ) -> MultiWorkerResult: - """Orchestrate async multi-worker quantization via asyncio.gather. + """Orchestrate async multi-worker quantization via asyncio.TaskGroup. Each worker gets its own async Redis connection and backup file shard. + If a worker fails, its siblings are cancelled, and each closes its + connection, before the failure is raised, so no worker is still writing + once the caller sees the error. A cancelled worker's shard records only + whole batches, so a later run resumes it like any interrupted shard. + + Raises: + Exception: The failing worker's own exception, not an + ``ExceptionGroup``. If several workers fail before the + cancellation reaches them, the first is raised and the rest + are logged. """ import asyncio @@ -562,22 +572,34 @@ async def async_multi_worker_quantize( f"({len(worker_backup_paths)} != {actual_workers})" ) - coroutines = [ - _async_worker_quantize( - worker_id=i, - redis_url=redis_url, - keys=slices[i], - datatype_changes=datatype_changes, - backup_path=worker_backup_paths[i], - index_name=index_name, - batch_size=batch_size, - progress_callback=progress_callback, - ) - for i in range(actual_workers) - ] - - results = await asyncio.gather(*coroutines) - worker_results = list(results) + failure: Exception | None = None + try: + async with asyncio.TaskGroup() as group: + tasks = [ + group.create_task( + _async_worker_quantize( + worker_id=i, + redis_url=redis_url, + keys=slices[i], + datatype_changes=datatype_changes, + backup_path=worker_backup_paths[i], + index_name=index_name, + batch_size=batch_size, + progress_callback=progress_callback, + ) + ) + for i in range(actual_workers) + ] + except ExceptionGroup as group_error: + failure = group_error.exceptions[0] + for other in group_error.exceptions[1:]: + logger.error("Another quantization worker also failed", exc_info=other) + if failure is not None: + # Raised outside the except block so the worker's traceback and cause + # reach the caller unchanged, as they did under asyncio.gather. + raise failure + + worker_results = [task.result() for task in tasks] total_docs = sum(r["docs"] for r in worker_results) return MultiWorkerResult( diff --git a/redisvl/query/filter.py b/redisvl/query/filter.py index 616ef45ac..6411bd567 100644 --- a/redisvl/query/filter.py +++ b/redisvl/query/filter.py @@ -2,7 +2,7 @@ import math import numbers import re -from enum import Enum +from enum import Enum, StrEnum from functools import wraps from typing import Any, Callable @@ -12,7 +12,7 @@ # mypy: disable-error-code="override" -class Inclusive(str, Enum): +class Inclusive(StrEnum): """Enum for valid inclusive options""" BOTH = "both" diff --git a/redisvl/query/query.py b/redisvl/query/query.py index e740d9855..ce5c001b0 100644 --- a/redisvl/query/query.py +++ b/redisvl/query/query.py @@ -1,4 +1,4 @@ -from enum import Enum +from enum import StrEnum from typing import Any, Self from redis.commands.search.query import Query as RedisQuery @@ -429,7 +429,7 @@ class BaseVectorQuery: _normalize_vector_distance: bool = False -class HybridPolicy(str, Enum): +class HybridPolicy(StrEnum): """Enum for valid hybrid policy options in vector queries.""" BATCHES = "BATCHES" @@ -592,7 +592,7 @@ def _build_query_string(self) -> str: # Add hybrid policy parameters if specified if self._hybrid_policy: - knn_query += f" HYBRID_POLICY {self._hybrid_policy.value}" + knn_query += f" HYBRID_POLICY {self._hybrid_policy}" # Add batch size if specified and using BATCHES policy if self._hybrid_policy == HybridPolicy.BATCHES and self._batch_size: diff --git a/redisvl/schema/fields.py b/redisvl/schema/fields.py index d80f26394..236f4aad2 100644 --- a/redisvl/schema/fields.py +++ b/redisvl/schema/fields.py @@ -34,7 +34,7 @@ - Vector Search: https://redis.io/docs/interact/search-and-query/advanced-concepts/vectors/ """ -from enum import Enum +from enum import StrEnum from typing import Any, Literal from pydantic import BaseModel, Field, field_validator, model_validator @@ -57,7 +57,7 @@ } -class FieldTypes(str, Enum): +class FieldTypes(StrEnum): TAG = "tag" TEXT = "text" NUMERIC = "numeric" @@ -65,13 +65,13 @@ class FieldTypes(str, Enum): VECTOR = "vector" -class VectorDistanceMetric(str, Enum): +class VectorDistanceMetric(StrEnum): COSINE = "COSINE" L2 = "L2" IP = "IP" -class VectorDataType(str, Enum): +class VectorDataType(StrEnum): BFLOAT16 = "BFLOAT16" FLOAT16 = "FLOAT16" FLOAT32 = "FLOAT32" @@ -80,13 +80,13 @@ class VectorDataType(str, Enum): UINT8 = "UINT8" -class VectorIndexAlgorithm(str, Enum): +class VectorIndexAlgorithm(StrEnum): FLAT = "FLAT" HNSW = "HNSW" SVS_VAMANA = "SVS-VAMANA" -class CompressionType(str, Enum): +class CompressionType(StrEnum): """Vector compression types for SVS-VAMANA algorithm""" LVQ4 = "LVQ4" @@ -328,7 +328,7 @@ def validate_svs_params(self): if not self.compression.value.startswith("LeanVec"): raise ValueError( f"reduce parameter is only supported with LeanVec compression types. " - f"Got compression={self.compression.value}. " + f"Got compression={self.compression}. " f"Either use LeanVec4x8/LeanVec8x8 or remove the reduce parameter." ) diff --git a/redisvl/schema/validation.py b/redisvl/schema/validation.py index 3bd7cf9e3..d629129c9 100644 --- a/redisvl/schema/validation.py +++ b/redisvl/schema/validation.py @@ -174,22 +174,18 @@ def _validate_vector(cls, value): raise ValueError( f"Vector field '{fname}' must have {dims} dimensions, got {len(value)}" ) - # Validate data types - datatype_str = str(datatype).upper() - # Integer-based datatypes - if datatype_str in ("INT8", "UINT8"): - # Check range for INT8 - if datatype_str == "INT8": - if any(v < -128 or v > 127 for v in value): - raise ValueError( - f"Vector field '{fname}' contains values outside the INT8 range (-128 to 127)" - ) - # Check range for UINT8 - elif datatype_str == "UINT8": - if any(v < 0 or v > 255 for v in value): - raise ValueError( - f"Vector field '{fname}' contains values outside the UINT8 range (0 to 255)" - ) + # Check range for INT8 + if datatype == VectorDataType.INT8: + if any(v < -128 or v > 127 for v in value): + raise ValueError( + f"Vector field '{fname}' contains values outside the INT8 range (-128 to 127)" + ) + # Check range for UINT8 + elif datatype == VectorDataType.UINT8: + if any(v < 0 or v > 255 for v in value): + raise ValueError( + f"Vector field '{fname}' contains values outside the UINT8 range (0 to 255)" + ) return value return _validate_vector diff --git a/tests/integration/test_index_migrator_e2e_recovery.py b/tests/integration/test_index_migrator_e2e_recovery.py index 878d072c0..8749af3ca 100644 --- a/tests/integration/test_index_migrator_e2e_recovery.py +++ b/tests/integration/test_index_migrator_e2e_recovery.py @@ -536,9 +536,6 @@ def test_json_same_width_datatype_change_is_schema_only_end_to_end( live_index = SearchIndex.from_existing(index_name, redis_url=redis_url) vector_field = live_index.schema.fields.get("embedding") assert vector_field is not None - datatype = getattr( - vector_field.attrs.datatype, "value", vector_field.attrs.datatype - ) - assert str(datatype).lower() == "bfloat16" + assert str(vector_field.attrs.datatype).lower() == "bfloat16" finally: _cleanup(redis_url, client, index_name, prefix) diff --git a/tests/integration/test_message_history.py b/tests/integration/test_message_history.py index 09a95f1c3..df8157682 100644 --- a/tests/integration/test_message_history.py +++ b/tests/integration/test_message_history.py @@ -5,7 +5,11 @@ from redis.exceptions import ConnectionError, NoPermissionError from redisvl.extensions.constants import ID_FIELD_NAME -from redisvl.extensions.message_history import MessageHistory, SemanticMessageHistory +from redisvl.extensions.message_history import ( + ChatRole, + MessageHistory, + SemanticMessageHistory, +) from tests.conftest import skip_if_no_redis_search @@ -306,6 +310,25 @@ def test_standard_get_raw(standard_history): assert raw[1]["content"] == "first response" +def test_standard_chat_role_members_round_trip(standard_history, client): + standard_history.add_messages( + [ + {"role": ChatRole.USER, "content": "prompt"}, + {"role": ChatRole.TOOL, "content": "tool output", "tool_call_id": "t1"}, + {"role": ChatRole.ASSISTANT, "content": "response"}, + ] + ) + + raw = standard_history.get_recent(raw=True) + stored = [ + client.hget(standard_history._index.key(m[ID_FIELD_NAME]), "role") for m in raw + ] + assert stored == [b"user", b"tool", b"assistant"] + + by_role = standard_history.get_recent(role=[ChatRole.USER, ChatRole.TOOL]) + assert [m["role"] for m in by_role] == ["user", "tool"] + + def test_standard_drop(standard_history): standard_history.store("first prompt", "first response") standard_history.store("second prompt", "second response") diff --git a/tests/unit/test_mcp/test_server.py b/tests/unit/test_mcp/test_server.py index 394ffe656..e9918c8ab 100644 --- a/tests/unit/test_mcp/test_server.py +++ b/tests/unit/test_mcp/test_server.py @@ -478,3 +478,52 @@ async def test_shutdown_after_stop_is_idempotent(): await server.shutdown() assert server._lifecycle_state.name == "STOPPED" + + +class _SlowInfoClient: + def __init__(self, info_delay: float): + self.info_delay = info_delay + self.aclose_calls = 0 + + async def info(self, section): + await asyncio.sleep(self.info_delay) + return {"redis_version": "8.0.0"} + + async def aclose(self): + self.aclose_calls += 1 + + +def _patch_slow_connect(monkeypatch, client, connect_delay: float): + async def slow_connect(redis_url=None, **kwargs): + await asyncio.sleep(connect_delay) + return client + + monkeypatch.setattr( + "redisvl.mcp.server.RedisConnectionFactory._get_aredis_connection", + slow_connect, + ) + + +@pytest.mark.asyncio +async def test_connect_redis_client_shares_one_deadline(monkeypatch): + # Each step fits the timeout alone, but not both together. + client = _SlowInfoClient(info_delay=0.15) + _patch_slow_connect(monkeypatch, client, connect_delay=0.15) + server = RedisVLMCPServer(_dummy_settings()) + server.config = _startup_config() + + with pytest.raises(TimeoutError): + await server._connect_redis_client(0.2) + + assert client.aclose_calls == 1 + + +@pytest.mark.asyncio +async def test_connect_redis_client_returns_client_within_deadline(monkeypatch): + client = _SlowInfoClient(info_delay=0.05) + _patch_slow_connect(monkeypatch, client, connect_delay=0.05) + server = RedisVLMCPServer(_dummy_settings()) + server.config = _startup_config() + + assert await server._connect_redis_client(1) is client + assert client.aclose_calls == 0 diff --git a/tests/unit/test_multi_worker_quantize.py b/tests/unit/test_multi_worker_quantize.py index 8a3808b30..0a68ba6ee 100644 --- a/tests/unit/test_multi_worker_quantize.py +++ b/tests/unit/test_multi_worker_quantize.py @@ -9,10 +9,13 @@ - Progress aggregation """ +import asyncio +import logging import struct from unittest.mock import MagicMock, patch import pytest +from redis.exceptions import ConnectionError as RedisConnectionError def _make_float32_vector(dims: int = 4, seed: float = 0.0) -> bytes: @@ -364,3 +367,171 @@ async def test_async_worker_loads_existing_backup(self, tmp_path): fields=dt_changes, batch_size=2, ) + + +class _FakeAsyncRedis: + """Hash store shared by every worker; ``on_execute`` may fail or stall.""" + + def __init__(self, store, on_execute): + self.store = store + self.on_execute = on_execute + self.closed = False + + def pipeline(self, transaction=False): + return _FakePipeline(self) + + async def aclose(self): + self.closed = True + + +class _FakePipeline: + def __init__(self, client): + self.client = client + self.ops = [] + + def hget(self, key, field): + self.ops.append(("hget", key, field)) + + def hset(self, key, mapping): + self.ops.append(("hset", key, mapping)) + + async def execute(self): + results = [] + for op, key, arg in self.ops: + if op == "hget": + results.append(self.client.store.get(key, {}).get(arg)) + else: + self.client.store.setdefault(key, {}).update(arg) + results.append(len(arg)) + # Runs after the commands land, like a reply lost after Redis applied them. + await self.client.on_execute(self.ops) + return results + + +class _WorkerFailure: + """Worker 0 fails once workers 1 and 2 are stalled inside their second batch. + + Keys are ``doc:0``..``doc:11``: three workers of four keys each, in + batches of two. + """ + + def __init__(self, op): + self.op = op + self.stalled = 0 + self.siblings_stalled = asyncio.Event() + self.cancelled = [] + + async def on_execute(self, ops): + op, key, _ = ops[0] + if op != self.op: + return + doc = int(key.split(":")[1]) + worker = doc // 4 + if worker == 0: + await self.siblings_stalled.wait() + raise RedisConnectionError("worker 0 lost its connection") + if doc % 4 == 2: + self.stalled += 1 + if self.stalled == 2: + self.siblings_stalled.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + self.cancelled.append(worker) + raise + + +async def _healthy(ops): + return None + + +class TestAsyncMultiWorkerFailure: + """One worker's failure cancels its siblings, and the shards still resume.""" + + DATATYPE_CHANGES = { + "embedding": {"source": "float32", "target": "float16", "dims": 4} + } + + def _run(self, store, on_execute, tmp_path, clients): + from redisvl.migration.quantize import async_multi_worker_quantize + + def from_url(url): + client = _FakeAsyncRedis(store, on_execute) + clients.append(client) + return client + + async def run(): + with patch("redis.asyncio.from_url", from_url): + async with asyncio.timeout(5): + return await async_multi_worker_quantize( + redis_url="redis://localhost:6379", + keys=[f"doc:{i}" for i in range(12)], + datatype_changes=self.DATATYPE_CHANGES, + backup_dir=str(tmp_path), + index_name="idx", + num_workers=3, + batch_size=2, + ) + + return run() + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "op, dump_batches, quantized_batches", + [("hget", 1, 0), ("hset", 2, 1)], + ids=["during-dump", "during-quantize"], + ) + async def test_failure_cancels_siblings_and_backups_resume( + self, tmp_path, op, dump_batches, quantized_batches + ): + from redisvl.migration.backup import VectorBackup + from redisvl.migration.quantize import ( + build_worker_backup_paths, + convert_vectors, + ) + + originals = { + f"doc:{i}": {"embedding": _make_float32_vector(4, seed=float(i))} + for i in range(12) + } + store = {key: dict(fields) for key, fields in originals.items()} + scenario = _WorkerFailure(op) + clients = [] + + with pytest.raises(RedisConnectionError, match="worker 0"): + await self._run(store, scenario.on_execute, tmp_path, clients) + + # The siblings were cancelled before the error surfaced, not left running. + assert sorted(scenario.cancelled) == [1, 2] + assert len(clients) == 3 + assert all(client.closed for client in clients) + + # Each cancelled shard recorded only the batches it finished. + paths = build_worker_backup_paths(str(tmp_path), "idx", 3) + for path in paths[1:]: + header = VectorBackup.load(path).header + assert header.dump_completed_batches == dump_batches + assert header.quantize_completed_batches == quantized_batches + + result = await self._run(store, _healthy, tmp_path, []) + + assert result.total_docs_quantized == 12 + assert all(VectorBackup.load(p).header.phase == "completed" for p in paths) + assert store == convert_vectors(originals, self.DATATYPE_CHANGES) + + @pytest.mark.asyncio + async def test_simultaneous_failures_raise_the_first_not_a_group( + self, tmp_path, caplog + ): + store = {f"doc:{i}": {"embedding": _make_float32_vector(4)} for i in range(12)} + + async def every_worker_fails(ops): + raise RedisConnectionError(f"failed at {ops[0][1]}") + + with ( + caplog.at_level(logging.ERROR, logger="redisvl.migration.quantize"), + pytest.raises(RedisConnectionError, match="failed at doc:0"), + ): + await self._run(store, every_worker_fails, tmp_path, []) + + assert caplog.text.count("Another quantization worker also failed") == 2 diff --git a/tests/unit/test_str_enums.py b/tests/unit/test_str_enums.py new file mode 100644 index 000000000..1c92afbde --- /dev/null +++ b/tests/unit/test_str_enums.py @@ -0,0 +1,143 @@ +"""Pin the visible string form of RedisVL's string enums. + +These enums are ``StrEnum`` subclasses, so ``str()`` and f-strings give the +member's value. Serialisation (pydantic, YAML, the FT.CREATE arguments sent to +Redis) went through the value before the move from ``(str, Enum)`` and must +not change. +""" + +from enum import StrEnum + +import pytest +import yaml +from redis.connection import Encoder + +from redisvl.extensions.message_history.schema import ChatRole +from redisvl.mcp.errors import MCPErrorCode +from redisvl.query.filter import Inclusive, Num, Tag +from redisvl.query.query import HybridPolicy +from redisvl.schema import IndexSchema +from redisvl.schema.fields import ( + CompressionType, + FieldTypes, + VectorDataType, + VectorDistanceMetric, + VectorIndexAlgorithm, +) + +STR_ENUMS = [ + MCPErrorCode, + ChatRole, + FieldTypes, + VectorDistanceMetric, + VectorDataType, + VectorIndexAlgorithm, + CompressionType, + HybridPolicy, + Inclusive, +] + + +@pytest.mark.parametrize( + "member", [m for cls in STR_ENUMS for m in cls], ids=lambda m: repr(m) +) +def test_member_stringifies_to_its_value(member): + assert isinstance(member, StrEnum) + assert str(member) == f"{member}" == "%s" % member == member.value + + +def _schema(**index): + return IndexSchema.from_dict( + { + "index": {"name": "enums", "prefix": "enums", **index}, + "fields": [ + {"name": "tag", "type": "tag"}, + { + "name": "flat", + "type": "vector", + "attrs": { + "algorithm": "flat", + "dims": 4, + "datatype": "int8", + "distance_metric": "l2", + }, + }, + { + "name": "svs", + "type": "vector", + "attrs": { + "algorithm": "svs-vamana", + "dims": 8, + "datatype": "float16", + "distance_metric": "ip", + "compression": "LeanVec4x8", + "reduce": 4, + }, + }, + ], + } + ) + + +def test_ft_create_arguments_encode_member_values(): + encoder = Encoder("utf-8", "strict", False) + fields = _schema(storage_type="json").fields + + def encoded(name): + return [encoder.encode(a) for a in fields[name].as_redis_field().redis_args()] + + assert encoded("tag")[3] == b"TAG" + assert encoded("flat") == [ + *(b"$.flat", b"AS", b"flat", b"VECTOR", b"FLAT", b"6"), + *(b"TYPE", b"INT8", b"DIM", b"4", b"DISTANCE_METRIC", b"L2"), + ] + svs = encoded("svs") + assert svs[3:5] == [b"VECTOR", b"SVS-VAMANA"] + assert svs[svs.index(b"TYPE") + 1] == b"FLOAT16" + assert svs[svs.index(b"DISTANCE_METRIC") + 1] == b"IP" + assert svs[svs.index(b"COMPRESSION") + 1] == b"LeanVec4x8" + + +def test_schema_serialisation_uses_member_values(tmp_path): + schema = _schema() + + flat = schema.model_dump(mode="json")["fields"]["flat"] + assert flat["type"] == "vector" + assert flat["attrs"]["algorithm"] == "FLAT" + assert flat["attrs"]["datatype"] == "INT8" + assert flat["attrs"]["distance_metric"] == "L2" + svs_attrs = schema.fields["svs"].attrs + assert svs_attrs.model_dump(mode="json")["compression"] == "LeanVec4x8" + + as_dict = schema.to_dict() + schema.to_yaml(str(tmp_path / "schema.yaml")) + assert yaml.safe_load((tmp_path / "schema.yaml").read_text()) == as_dict + flat_attrs = next(f for f in as_dict["fields"] if f["name"] == "flat")["attrs"] + assert flat_attrs["algorithm"] == "flat" + assert flat_attrs["datatype"] == "int8" + assert flat_attrs["distance_metric"] == "l2" + + +def test_error_messages_name_the_value(): + with pytest.raises(ValueError, match=r"Got: INT8\. "): + IndexSchema.from_dict( + { + "index": {"name": "e", "prefix": "e"}, + "fields": [ + { + "name": "v", + "type": "vector", + "attrs": { + "algorithm": "svs-vamana", + "dims": 4, + "datatype": "int8", + }, + } + ], + } + ) + + +def test_query_strings_use_member_values(): + assert str(Tag("role") == [ChatRole.USER, ChatRole.TOOL]) == "@role:{user|tool}" + assert str(Num("n").between(1, 2, inclusive=Inclusive.LEFT)) == "@n:[1 (2]" diff --git a/tests/unit/test_validation.py b/tests/unit/test_validation.py index f34d0035e..fa12ba81d 100644 --- a/tests/unit/test_validation.py +++ b/tests/unit/test_validation.py @@ -529,6 +529,59 @@ def test_vector_field_validation( ) +class TestIntegerVectorRangeValidation: + """Range checks for INT8 and UINT8 vectors given as lists (JSON storage).""" + + @pytest.fixture + def int_vector_schema(self): + return IndexSchema.from_dict( + { + "index": {"name": "int-vectors", "storage_type": "json"}, + "fields": [ + { + "name": field_name, + "type": "vector", + "attrs": { + "algorithm": "flat", + "dims": 2, + "datatype": datatype, + "distance_metric": "l2", + }, + } + for field_name, datatype in (("i8", "int8"), ("u8", "uint8")) + ], + } + ) + + @pytest.mark.parametrize( + "field_name,value", + [ + ("i8", [-128, 127]), + ("u8", [0, 255]), + ], + ) + def test_in_range_values_accepted(self, int_vector_schema, field_name, value): + assert validate_object(int_vector_schema, {field_name: value}) == { + field_name: value + } + + @pytest.mark.parametrize( + "field_name,value,error_text", + [ + ("i8", [999, -999], "INT8 range"), + ("i8", [-129, 0], "INT8 range"), + ("i8", [0, 128], "INT8 range"), + ("u8", [-1, 0], "UINT8 range"), + ("u8", [0, 256], "UINT8 range"), + ], + ) + def test_out_of_range_values_rejected( + self, int_vector_schema, field_name, value, error_text + ): + with pytest.raises(ValueError, match=error_text): + validate_object(int_vector_schema, {field_name: value}) + + class TestNestedJsonValidation: """Tests for JSON path-based validation with nested structures."""