diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 491597d..c275ca9 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -46,3 +46,53 @@ jobs: run: | make test + + redis-py-compat: + name: redis-py ${{ matrix.redis-py }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + redis-py: ["6.3.0", "7.4.0", "8.1.0"] + + steps: + - name: Check out repository + uses: actions/checkout@v6 + + - name: Install Python + uses: actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Install uv + uses: astral-sh/setup-uv@v6 + with: + version: ${{ env.UV_VERSION }} + enable-cache: true + python-version: "3.12" # sets UV_PYTHON + cache-dependency-glob: | + pyproject.toml + uv.lock + + - name: Install dependencies + run: | + uv sync + + # Without this, a resolution surprise leaves the job exercising the + # locked redis-py while still reporting green, and the axis rots. + - name: Confirm the redis-py overlay took effect + run: > + uv run --with 'redis==${{ matrix.redis-py }}' + python -c "import redis, sys; + print('redis-py', redis.__version__); + sys.exit(0 if redis.__version__ == '${{ matrix.redis-py }}' else 1)" + + # Reply shapes depend on the redis-py version and the negotiated RESP + # protocol, so the protocol-marked tests run against each supported + # redis-py. Selecting by marker rather than by filename means a new + # protocol-sensitive test is picked up here without editing this job. + # 'uv run --with' layers an ephemeral overlay and never writes uv.lock. + - name: Run reply-shape and protocol tests + run: > + uv run --with 'redis==${{ matrix.redis-py }}' + python -m pytest -m protocol -q diff --git a/docs/concepts/index.md b/docs/concepts/index.md index 118a4fd..e64291b 100644 --- a/docs/concepts/index.md +++ b/docs/concepts/index.md @@ -60,7 +60,7 @@ Foundational reading for sql-redis. Each page explains a single design choice or --- - What `QueryResult.rows` actually contains, why it varies with the command, scoring, and client decoding. + What `QueryResult.rows` actually contains, why it varies with the command, scoring, client decoding and the wire protocol. - :material-test-tube:{ .lg .middle } **[Testing philosophy](testing-philosophy.md)** diff --git a/docs/concepts/result-shape.md b/docs/concepts/result-shape.md index 246fb81..9fd133e 100644 --- a/docs/concepts/result-shape.md +++ b/docs/concepts/result-shape.md @@ -4,14 +4,16 @@ Every `Executor.execute()` call returns a `QueryResult` with two attributes: `ro ## What `count` means -The integer in `count` is what Redis returned at position 0 of the reply. Its meaning depends on which command ran (see [FT.SEARCH vs FT.AGGREGATE](search-vs-aggregate.md)): +`count` is whatever Redis reported for the query: position 0 of a RESP2 array reply, or `total_results` in a RESP3 map reply. Its meaning depends on which command ran (see [FT.SEARCH vs FT.AGGREGATE](search-vs-aggregate.md)): -- **`FT.SEARCH`** (no aggregation): `count` is the **total number of matching documents** in the index, regardless of `LIMIT`. So `count` can be much larger than `len(rows)`. This is useful for pagination. -- **`FT.AGGREGATE`** (any aggregation, GROUP BY, computed field, date function): `count` is the **number of rows in the reply**, which is what you got back. After `LIMIT`, the two are equal. +- **`FT.SEARCH`** (no aggregation): `count` is the **total number of matching documents**, regardless of `LIMIT`. So `count` can be much larger than `len(rows)`, which is what makes it useful for pagination. Both protocols report the same figure. +- **`FT.AGGREGATE`** (any aggregation, GROUP BY, computed field, date function): `count` is **not the number of rows you got back**, and it is the one value that differs by protocol. With a `GROUP BY`, both protocols report the number of groups before `LIMIT`. Without one, a computed-field projection leaves RESP2 reporting `1` (a placeholder the [Redis docs](https://redis.io/docs/latest/commands/ft.aggregate/) call "not a valid value") while RESP3 reports the real figure. Use `len(rows)` if you want the rows you received. + +Neither aggregate value can be derived from the other without changing what `protocol=2` callers already get, so the library reports what Redis sent. ## What `rows` looks like -A row is always a `dict`, but the *keys* and *values* depend on three orthogonal factors. +A row is always a `dict`, but the *keys* and *values* depend on four orthogonal factors. ### Factor 1: client decoding @@ -50,12 +52,13 @@ If your SELECT contains `score()`, the underlying `FT.SEARCH` runs with `WITHSCO ```python # SELECT title, score() AS relevance FROM products WHERE fulltext(...) -{b"title": b"gaming laptop", b"relevance": "0.5"} +{b"title": b"gaming laptop", b"relevance": b"0.5"} # RESP2 +{b"title": b"gaming laptop", b"relevance": 0.5} # RESP3 ``` The score column name is whatever alias you wrote (`score() AS relevance` produces `relevance`). If you used `score()` without `AS`, the library falls back to a stable internal name. -The score's value type is `str` (or `bytes` depending on decoding); if you need a `float`, convert at the call site. +The score's value type depends on the wire protocol: `str` (or `bytes`, depending on decoding) on RESP2, and `float` on RESP3, where Redis sends it as a double. `float()` is the protocol-safe conversion at the call site and works on both. ### Factor 4: `RETURN 0` @@ -63,11 +66,17 @@ If your SELECT is just `score()` with no document fields, the underlying command ```python # SELECT score() AS s FROM products WHERE fulltext(title, 'laptop') -{b"s": "0.5"} +{b"s": b"0.5"} ``` This is mostly useful when you want a relevance-only ranking without paying to ship the document body back. +## A note on the wire protocol + +Redis replies to `FT.SEARCH` and `FT.AGGREGATE` with a flat array on RESP2 and a map on RESP3, and `sql_redis` reads the raw reply, so the protocol your client negotiated decides which shape arrives. The library parses both and the rows come out the same, but two values differ: the score's type (Factor 3) and the aggregate `count` described above. + +Worth knowing because redis-py 8 defaults to RESP3 where earlier versions defaulted to RESP2, so an upgrade can change both without you passing `protocol` at all. + ## Score-column collision avoidance What happens if your document has a field literally named `__score`? The library detects the collision and renames the score column. The deterministic fallback is `__score_`. So: @@ -81,7 +90,7 @@ You will not normally see this; it is a defensive measure for the unusual case. It would be tempting to normalise everything: always strings, always `float` scores, always uniform shape. The library does not, for three reasons. -1. **Bytes is the right default for Redis.** Some fields legitimately contain binary data. Force-decoding to UTF-8 would corrupt those. +1. **Bytes is the right default for Redis.** Some fields legitimately contain binary data. Force-decoding to UTF-8 would corrupt those. The same argument covers the RESP3 score: stringifying a double cannot reproduce the text Redis would have sent on RESP2, so normalising would invent a third form rather than deliver parity. 2. **The shape difference between `FT.SEARCH` and `FT.AGGREGATE` is intrinsic to the Redis commands.** Hiding it would lie about what is happening. 3. **One-call-site logic is cheap; library-wide policy is expensive.** A user who wants a uniform shape can write a 5-line decoder once, customised to their data. The library shipping that decoder for everyone is the wrong place for the policy. diff --git a/docs/concepts/search-vs-aggregate.md b/docs/concepts/search-vs-aggregate.md index eb26477..ee8e82d 100644 --- a/docs/concepts/search-vs-aggregate.md +++ b/docs/concepts/search-vs-aggregate.md @@ -39,6 +39,6 @@ These constraints are not bugs to fix; they are the cost of the abstraction. The Two practical consequences: 1. **The result-row shape changes.** `FT.SEARCH` returns rows with field-value pairs straight from the indexed documents, possibly with a score column. `FT.AGGREGATE` returns rows of computed fields, group keys, and reduced values; the original document fields are present only if the SQL asks for them. See [Result shape](result-shape.md). -2. **`LIMIT` semantics differ subtly.** Both commands honour `LIMIT`, but the `count` returned by `FT.AGGREGATE` reflects the post-pipeline row count, while `FT.SEARCH`'s `count` is the total match count regardless of the limit. The library exposes both as `QueryResult.count`; the meaning depends on which path ran. +2. **`LIMIT` semantics differ subtly.** Both commands honour `LIMIT`, but `FT.SEARCH`'s `count` is the total match count regardless of the limit, while `FT.AGGREGATE`'s is neither the match count nor the row count, and is the one value that differs between RESP2 and RESP3. The library exposes both as `QueryResult.count`; see [Result shape](result-shape.md#what-count-means) before relying on it. If you need to know which command was issued for a given SQL, call `Translator.translate(sql)` directly and inspect `TranslatedQuery.command`. diff --git a/docs/for-ais-only/BUILD_AND_TEST.md b/docs/for-ais-only/BUILD_AND_TEST.md index 638eded..ba2512f 100644 --- a/docs/for-ais-only/BUILD_AND_TEST.md +++ b/docs/for-ais-only/BUILD_AND_TEST.md @@ -62,7 +62,11 @@ uv run --group docs mkdocs build --strict ## CI gates (target state) -- `make check` (lint + tests) on every PR. +- `make check` (lint + tests) on every PR, across Python 3.9 to 3.13. +- `pytest -m protocol` against redis-py 6.3.0, 7.4.0 and 8.1.0 (the + `redis-py-compat` job). Mark any test whose result depends on the RESP + protocol or the redis-py version with `@pytest.mark.protocol` so it runs + there. - `make docs-build` with `-W` on every PR. - 100% coverage gate. @@ -76,7 +80,18 @@ uv run pytest tests/test_parser.py tests/test_translator.py -x -vv These do not need Redis and run in well under a second. -When changing executor logic: +When changing reply parsing: + +``` +uv run pytest tests/test_reply_shapes.py tests/test_nil_field_array.py -x -vv +``` + +These feed canned replies through the parser with no server, so they are the +fastest signal on a parsing change. Follow with +`tests/test_protocol_matrix.py`, which asserts the same behaviour against a +real server on both protocols. + +When changing other executor logic: ``` uv run pytest tests/test_executor.py tests/test_sql_queries.py -x -vv diff --git a/docs/for-ais-only/FAILURE_MODES.md b/docs/for-ais-only/FAILURE_MODES.md index 3cac618..f4cd74c 100644 --- a/docs/for-ais-only/FAILURE_MODES.md +++ b/docs/for-ais-only/FAILURE_MODES.md @@ -13,6 +13,33 @@ not "fix" this in `executor.py` by force-decoding; that breaks users who deliberately want bytes (binary fields, vectors). See [Result shape](../concepts/result-shape.md) for the full story. +## Reply shape depends on the RESP protocol, not on library configuration + +`sql_redis` sends FT.* commands with `client.execute_command` and reads the raw +reply. redis-py registers no `FT.SEARCH` / `FT.AGGREGATE` response callback on +the base client. `Search.search()` calls `execute_command` and then its own +`_parse_results`, so the executor receives the wire shape verbatim: a flat +array on RESP2, a map on RESP3. + +`_parse_reply` therefore dispatches on the shape of the reply +(`isinstance(raw_result, dict)`), and `_fold_map_to_array` folds a map back +into the array shape so one parser handles both. Do not "simplify" this into a +single unconditional path, and do not replace the shape check with a protocol +check on the connection: sniffing `connection_kwargs["protocol"]` is wrong for +cluster and sentinel clients, and it cannot distinguish redis-py 8's four +callback modes (`protocol=2`, protocol unset, `protocol=3`, +`legacy_responses=False`). + +Routing through `client.ft(index)` does not help. On redis-py 7.4 and earlier +`_parse_results` returns the raw map unchanged under RESP3, and on redis-py 8 +the `protocol=3` callback table has no `FT.AGGREGATE` entry at all. + +Structural map keys (`total_results`, `results`, `id`, `score`, +`extra_attributes`) arrive as `bytes` when the client has +`decode_responses=False`, which is why `_map_get` probes both key types. Reading +only `str` keys fails silently with zero rows rather than raising; that is +exactly the bug redis-py shipped in 8.0.0 (redis-py#4107). + ## Stopwords are silently stripped `WHERE title = 'bank of america'` becomes `@title:"bank america"` because diff --git a/docs/for-ais-only/REPOSITORY_MAP.md b/docs/for-ais-only/REPOSITORY_MAP.md index cbae43e..504e7f7 100644 --- a/docs/for-ais-only/REPOSITORY_MAP.md +++ b/docs/for-ais-only/REPOSITORY_MAP.md @@ -69,7 +69,7 @@ tests/ | `IS NULL`, `IS NOT NULL`, `exists()` | `parser.py`, `query_builder.py` | | Parameter substitution | `executor.py::_substitute_params` | | FT.SEARCH vs FT.AGGREGATE branching | `translator.py::translate_parsed` (the `use_aggregate` boolean) | -| Result-row parsing | `executor.py::Executor.execute` (the four parsing branches: WITHSCORES + RETURN 0, WITHSCORES, plain SEARCH, AGGREGATE) | +| Result-row parsing | `executor.py::_parse_reply` (dispatches by reply shape), `_parse_array_reply` (the four branches: WITHSCORES + RETURN 0, WITHSCORES, plain SEARCH, AGGREGATE), `_fold_map_to_array` (folds a RESP3 map into the array shape) | ## What to read before changing X @@ -79,9 +79,10 @@ tests/ - **Analyzer/query-builder changes.** Read `analyzer.py` and the matching branch in `query_builder.py`. The two are tightly coupled: a new field type in the analyzer needs a corresponding emitter in the builder. -- **Executor changes.** The sync and async executors share `_ScoreParseMixin` - and the result-parsing branches. If you change the response shape, both - need updates. +- **Executor changes.** The sync and async executors share the module-level + reply parsers, so a parsing change lands once. Their `execute()` bodies are + still near-identical above the `_parse_reply` call, so a change there needs + applying twice. - **Schema registry changes.** The two registry classes do not share code, but they share semantics. Lazy loading, invalidation, and load coalescing must be consistent across sync and async. diff --git a/docs/for-ais-only/index.md b/docs/for-ais-only/index.md index c25fc29..b8d1372 100644 --- a/docs/for-ais-only/index.md +++ b/docs/for-ais-only/index.md @@ -26,7 +26,7 @@ Use this to find the right starting point fast. | Change schema loading semantics | [Schema-aware translation](../concepts/schema-aware-translation.md), then `schema.py` | | Change async behavior | [Async invariants](../concepts/async-invariants.md), then `executor.py::AsyncExecutor` and `schema.py::AsyncSchemaRegistry` | | Change the FT.SEARCH/FT.AGGREGATE branching | [FT.SEARCH vs FT.AGGREGATE](../concepts/search-vs-aggregate.md), then `translator.py::translate_parsed` | -| Change the result-row shape | [Result shape](../concepts/result-shape.md), then `executor.py` (`Executor.execute` parsing branches) | +| Change the result-row shape | [Result shape](../concepts/result-shape.md), then `executor.py::_parse_reply` and the functions below it | | Run tests | [Build and test](BUILD_AND_TEST.md) | | Diagnose "this looks broken" | [Failure modes](FAILURE_MODES.md) before assuming a bug | diff --git a/pyproject.toml b/pyproject.toml index db0a99e..f3803df 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -82,6 +82,9 @@ exclude = ''' [tool.pytest.ini_options] log_cli = true asyncio_mode = "auto" +markers = [ + "protocol: reply shapes depend on the RESP protocol and the redis-py version (run by the redis-py-compat CI job)", +] [tool.mypy] warn_unused_configs = true diff --git a/sql_redis/executor.py b/sql_redis/executor.py index a9549ba..65e52a3 100644 --- a/sql_redis/executor.py +++ b/sql_redis/executor.py @@ -9,7 +9,7 @@ import redis from sql_redis.schema import AsyncSchemaRegistry, SchemaRegistry -from sql_redis.translator import Translator +from sql_redis.translator import TranslatedQuery, Translator if TYPE_CHECKING: import redis.asyncio as async_redis @@ -107,6 +107,35 @@ def _substitute_params(sql: str, params: dict[str, Any]) -> str: return "".join(result) +def _map_get(mapping: dict, key: str, default: Any = None) -> Any: + """Look up a RESP3 map key, tolerating bytes keys. + + RESP3 replies arrive as maps whose *structural* keys (``total_results``, + ``results``, ``id``, ``score``, ``extra_attributes``) are ``str`` when the + client decodes responses and ``bytes`` when it does not. Document *field* + keys never go through this helper: they reach the caller exactly as + received, so a ``decode_responses=False`` client still gets bytes keys. + """ + if key in mapping: + return mapping[key] + return mapping.get(key.encode(), default) + + +def _field_array(fields: Any) -> list: + """Return a document's fields as a flat ``[field, value, ...]`` array. + + Accepts the RESP3 map form and the RESP2 flat form, mirroring + ``_parse_hybrid_reply``. A nil or unrecognized value becomes an empty + array, which is the same tolerance the array parser already applies to a + nil field-array (issue #38). + """ + if isinstance(fields, dict): + return [item for pair in fields.items() for item in pair] + if isinstance(fields, (list, tuple)): + return list(fields) + return [] + + @dataclass class QueryResult: """Result of executing a SQL query.""" @@ -115,98 +144,248 @@ class QueryResult: count: int -class _ScoreParseMixin: - """Shared helpers for score-related response parsing.""" +def _is_unknown_command(error_msg: str) -> bool: + """Return True when a ResponseError means the command is unsupported.""" + lowered = error_msg.lower() + return "unknown command" in lowered or "unknown subcommand" in lowered - @staticmethod - def _is_unknown_command(error_msg: str) -> bool: - """Return True when a ResponseError means the command is unsupported.""" - lowered = error_msg.lower() - return "unknown command" in lowered or "unknown subcommand" in lowered - @staticmethod - def _parse_hybrid_reply(raw_result) -> tuple[Any, list[dict]]: - """Parse an FT.HYBRID reply into (count, rows). +def _inject_vector_param(cmd: list[str | bytes], vector_param: bytes) -> None: + """Replace the vector PARAMS value with the actual bytes, in place. - FT.HYBRID does not use the FT.AGGREGATE array shape. The reply is a map - ``{total_results: N, results: [...], warnings: [...], ...}`` that arrives - either as a dict (redis-py 8.x / RESP3) or as a flat list - (``[total_results, N, results, [...], ...]``) on RESP2. Each result row - is likewise a dict or a flat ``[field, val, ...]`` list. Keys/values may - be bytes or str depending on the client's decode_responses setting. - """ - if isinstance(raw_result, dict): - reply = raw_result - else: - reply = dict(zip(raw_result[::2], raw_result[1::2])) - - def _field(name: str): - if name in reply: - return reply[name] - return reply.get(name.encode()) - - count = _field("total_results") or 0 - results = _field("results") or [] - rows = [ - dict(row) if isinstance(row, dict) else dict(zip(row[::2], row[1::2])) - for row in results - ] + Only the ``$vector`` token in the PARAMS value position (the one + preceded by the param name ``vector``) is replaced. Query-side + references to ``$vector`` (FT.SEARCH KNN expressions, FT.HYBRID VSIM) + must stay as parameter references so Redis resolves them from PARAMS. + """ + for i, arg in enumerate(cmd): + if arg == "$vector" and i > 0 and cmd[i - 1] == "vector": + cmd[i] = vector_param + + +def _has_return_0(args: list[str]) -> bool: + """Return True when the args contain 'RETURN 0' (no document fields).""" + try: + idx = args.index("RETURN") + return args[idx + 1] == "0" + except (ValueError, IndexError): + return False + + +def _resolve_score_alias( + score_alias: str | None, + args: list[str], + first_row_fields: set[str] | None = None, +) -> str: + """Determine a stable score column name that won't collide with + document fields. The alias is resolved once and reused for every + row so all rows share the same column name. + + When a RETURN clause is present, the returned field names are used + for collision detection. When RETURN is absent (SELECT *), the + caller should pass ``first_row_fields`` — the union of all field + names across all result rows — so we can detect collisions even + when different documents have different field sets.""" + alias = score_alias or "__score" + # Extract RETURN field names from args to detect collision + try: + idx = args.index("RETURN") + count = int(args[idx + 1]) + return_fields = set(args[idx + 2 : idx + 2 + count]) + except (ValueError, IndexError): + # Normalize bytes keys to str so collision detection works + # regardless of decode_responses setting. + raw = first_row_fields or set() + return_fields = {k.decode() if isinstance(k, bytes) else k for k in raw} + while alias in return_fields: + alias = f"__score_{alias}" + return alias + + +def _parse_hybrid_reply(raw_result) -> tuple[Any, list[dict]]: + """Parse an FT.HYBRID reply into (count, rows). + + FT.HYBRID does not use the FT.AGGREGATE array shape. The reply is a map + ``{total_results: N, results: [...], warnings: [...], ...}`` that arrives + either as a dict (RESP3) or as a flat list + (``[total_results, N, results, [...], ...]``) on RESP2. Each result row is + likewise a dict of *fields* or a flat ``[field, val, ...]`` list, with no + ``extra_attributes`` nesting: unlike FT.SEARCH, a hybrid row's own keys are + the field names. Keys and values may be bytes or str depending on the + client's decode_responses setting. + """ + if isinstance(raw_result, dict): + reply = raw_result + else: + reply = dict(zip(raw_result[::2], raw_result[1::2])) + + count = _map_get(reply, "total_results") or 0 + results = _map_get(reply, "results") or [] + rows = [ + dict(row) if isinstance(row, dict) else dict(zip(row[::2], row[1::2])) + for row in results + ] + return count, rows + + +@dataclass(frozen=True) +class _ReplyLayout: + """How one reply is laid out, derived once per query. + + The array parser strides over the reply, so it and the RESP3 fold have to + agree on which slots each document occupies. Deriving that here and passing + it to both is what keeps them from drifting apart. + """ + + is_search: bool + with_scores: bool + has_fields: bool + score_alias: str | None + args: list[str] + + @classmethod + def of(cls, translated: TranslatedQuery) -> "_ReplyLayout": + return cls( + is_search=translated.command == "FT.SEARCH", + # The explicit score_alias signal, rather than scanning args for + # the literal token "WITHSCORES", which could false-positive if a + # returned field happened to be named "WITHSCORES". + with_scores=translated.score_alias is not None, + # RETURN 0 suppresses document fields (like NOCONTENT), so each + # document arrives without a field array. + has_fields=not _has_return_0(translated.args), + score_alias=translated.score_alias, + args=translated.args, + ) + + +def _fold_map_to_array(reply: dict, layout: _ReplyLayout) -> list: + """Fold a RESP3 FT.SEARCH / FT.AGGREGATE map into the RESP2 array shape. + + A RESP3 reply is a map:: + + {total_results: N, attributes: [...], format: ..., warning: [...], + results: [{id: key, score: 0.5, extra_attributes: {field: val}, + values: [...]}, ...]} + + The output is only meaningful as input to ``_parse_array_reply`` under the + same ``layout``: it emits the document key, the score when WITHSCORES was + requested, and the field array unless RETURN 0 suppressed document fields, + which is exactly the slot layout that parser will stride over. ``id`` lands + in the position the parser discards; ``values``, ``attributes``, ``format`` + and ``warning`` are not read. + + Values are moved, not converted, so a RESP3 score stays a float and + document field keys keep whatever type the client returned. + """ + response: list = [_map_get(reply, "total_results") or 0] + results = _map_get(reply, "results") or [] + + for item in results: + fields = _map_get(item, "extra_attributes") if isinstance(item, dict) else None + if not layout.is_search: + # FT.AGGREGATE rows carry only extra_attributes: no id, no score. + response.append(_field_array(fields)) + continue + response.append(_map_get(item, "id", "") if isinstance(item, dict) else "") + if layout.with_scores: + response.append(_map_get(item, "score") if isinstance(item, dict) else None) + if layout.has_fields: + response.append(_field_array(fields)) + return response + + +def _parse_array_reply( + raw_result: list, layout: _ReplyLayout +) -> tuple[Any, list[dict]]: + """Parse the flat-array FT.SEARCH / FT.AGGREGATE reply into rows. + + This is the RESP2 wire shape, and the shape ``_fold_map_to_array`` folds a + RESP3 map into. + """ + count = raw_result[0] if raw_result else 0 + rows: list[dict] = [] + + if not layout.is_search: + # FT.AGGREGATE format: [count, [fields1], [fields2], ...] + for row_data in raw_result[1:]: + row_data = row_data or [] + rows.append(dict(zip(row_data[::2], row_data[1::2]))) return count, rows - @staticmethod - def _inject_vector_param(cmd: list[str | bytes], vector_param: bytes) -> None: - """Replace the vector PARAMS value with the actual bytes, in place. + if layout.with_scores and not layout.has_fields: + # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...] + # Stride of 2: key, score (no field array) + score_alias = _resolve_score_alias(layout.score_alias, layout.args) + for i in range(1, len(raw_result) - 1, 2): + rows.append({score_alias: raw_result[i + 1]}) + elif layout.with_scores: + # WITHSCORES: [count, key1, score1, [fields1], key2, score2, ...] + # Stride of 3: key, score, field_list + # First pass: collect all field names across all rows so the alias + # avoids collisions with any document field, not just the first row's. + all_field_names: set[str] = set() + parsed_rows: list[tuple[dict, Any]] = [] + for i in range(1, len(raw_result) - 2, 3): + score = raw_result[i + 1] + # A nil field-array (e.g. doc expired mid-query) becomes an empty + # field set, keeping the row's score instead of crashing. + row_data = raw_result[i + 2] or [] + row = dict(zip(row_data[::2], row_data[1::2])) + all_field_names.update(row.keys()) + parsed_rows.append((row, score)) + resolved_alias = _resolve_score_alias( + layout.score_alias, layout.args, first_row_fields=all_field_names + ) + for row, score in parsed_rows: + row[resolved_alias] = score + rows.append(row) + else: + # Standard format: [count, key1, [fields1], key2, [fields2], ...] + for i in range(2, len(raw_result), 2): + row_data = raw_result[i] or [] + rows.append(dict(zip(row_data[::2], row_data[1::2]))) + + return count, rows + + +def _parse_reply(raw_result: Any, translated: TranslatedQuery) -> QueryResult: + """Turn a raw FT.* reply into a QueryResult, RESP2 or RESP3. + + Dispatch is by reply shape rather than by the protocol the connection + negotiated; see ``docs/for-ais-only/FAILURE_MODES.md`` for why. A reply + that is neither shape is rejected rather than parsed into empty rows, + because reading a map that is not a result set (a cluster node-keyed reply, + FT.PROFILE) or an array that is not one (a WITHCURSOR pair) would otherwise + silently return nothing. + + FT.HYBRID keeps its own parser: its RESP3 rows are flat field maps, not the + ``{id, extra_attributes}`` documents FT.SEARCH returns. + """ + if translated.command == "FT.HYBRID": + count, rows = _parse_hybrid_reply(raw_result) + return QueryResult(rows=rows, count=count) + + layout = _ReplyLayout.of(translated) + if isinstance(raw_result, dict): + if _map_get(raw_result, "results") is None: + raise ValueError( + f"Unrecognized {translated.command} reply: a map without a " + f"'results' key. Got keys {sorted(map(str, raw_result))}." + ) + raw_result = _fold_map_to_array(raw_result, layout) + elif not isinstance(raw_result, list): + raise ValueError( + f"Unrecognized {translated.command} reply of type " + f"{type(raw_result).__name__}; expected an array (RESP2) or a map " + "(RESP3)." + ) - Only the ``$vector`` token in the PARAMS value position (the one - preceded by the param name ``vector``) is replaced. Query-side - references to ``$vector`` (FT.SEARCH KNN expressions, FT.HYBRID VSIM) - must stay as parameter references so Redis resolves them from PARAMS. - """ - for i, arg in enumerate(cmd): - if arg == "$vector" and i > 0 and cmd[i - 1] == "vector": - cmd[i] = vector_param + count, rows = _parse_array_reply(raw_result, layout) + return QueryResult(rows=rows, count=count) - @staticmethod - def _has_return_0(args: list[str]) -> bool: - """Return True when the args contain 'RETURN 0' (no document fields).""" - try: - idx = args.index("RETURN") - return args[idx + 1] == "0" - except (ValueError, IndexError): - return False - - @staticmethod - def _resolve_score_alias( - score_alias: str | None, - args: list[str], - first_row_fields: set[str] | None = None, - ) -> str: - """Determine a stable score column name that won't collide with - document fields. The alias is resolved once and reused for every - row so all rows share the same column name. - - When a RETURN clause is present, the returned field names are used - for collision detection. When RETURN is absent (SELECT *), the - caller should pass ``first_row_fields`` — the union of all field - names across all result rows — so we can detect collisions even - when different documents have different field sets.""" - alias = score_alias or "__score" - # Extract RETURN field names from args to detect collision - try: - idx = args.index("RETURN") - count = int(args[idx + 1]) - return_fields = set(args[idx + 2 : idx + 2 + count]) - except (ValueError, IndexError): - # Normalize bytes keys to str so collision detection works - # regardless of decode_responses setting. - raw = first_row_fields or set() - return_fields = {k.decode() if isinstance(k, bytes) else k for k in raw} - while alias in return_fields: - alias = f"__score_{alias}" - return alias - - -class Executor(_ScoreParseMixin): + +class Executor: """Executes SQL queries against Redis.""" def __init__(self, client: redis.Redis, schema_registry: SchemaRegistry) -> None: @@ -239,16 +418,14 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: # Replace the $vector PARAMS value with actual bytes (query/VSIM # references to $vector stay as parameter references). if vector_param: - self._inject_vector_param(cmd, vector_param) + _inject_vector_param(cmd, vector_param) # Execute command try: raw_result = self._client.execute_command(*cmd) except redis.ResponseError as e: error_msg = str(e) - if translated.command == "FT.HYBRID" and self._is_unknown_command( - error_msg - ): + if translated.command == "FT.HYBRID" and _is_unknown_command(error_msg): raise redis.ResponseError( f"{error_msg}. hybrid_vector_search() translates to FT.HYBRID, " "which requires Redis 8.4+ (RediSearch with hybrid search) " @@ -271,77 +448,10 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: ) from e raise - # Parse result based on command type - # FT.SEARCH/FT.AGGREGATE replies are arrays with the count first; - # FT.HYBRID replies are maps (dict) and set count during parsing below. - count = raw_result[0] if isinstance(raw_result, list) and raw_result else 0 - rows: list[dict] = [] - - if translated.command == "FT.HYBRID": - count, rows = self._parse_hybrid_reply(raw_result) - elif translated.command == "FT.SEARCH": - # Use the explicit score_alias signal rather than scanning args - # for the literal token "WITHSCORES", which could false-positive - # if a returned field happened to be named "WITHSCORES". - with_scores = translated.score_alias is not None - # RETURN 0 suppresses document fields (like NOCONTENT); - # with WITHSCORES the reply is [count, id, score, id, score, ...] - no_content = self._has_return_0(translated.args) - - # Pre-resolve score alias; may be deferred for SELECT * - score_alias: str | None = None - - if with_scores and no_content: - # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...] - # Stride of 2: key, score (no field array) - score_alias = self._resolve_score_alias( - translated.score_alias, translated.args - ) - for i in range(1, len(raw_result) - 1, 2): - score = raw_result[i + 1] - row = {score_alias: score} - rows.append(row) - elif with_scores: - # WITHSCORES format: [count, key1, score1, [fields1], key2, score2, [fields2], ...] - # Stride of 3: key, score, field_list - # First pass: collect all field names across all rows so the - # alias avoids collisions with any document field, not just - # the first row's fields. - all_field_names: set[str] = set() - parsed_rows: list[tuple[dict, Any]] = [] - for i in range(1, len(raw_result) - 2, 3): - score = raw_result[i + 1] - # A nil field-array (e.g. doc expired mid-query) becomes an - # empty field set, keeping the row's score instead of crashing. - row_data = raw_result[i + 2] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - all_field_names.update(row.keys()) - parsed_rows.append((row, score)) - resolved_alias = self._resolve_score_alias( - translated.score_alias, - translated.args, - first_row_fields=all_field_names, - ) - for row, score in parsed_rows: - row[resolved_alias] = score - rows.append(row) - else: - # Standard format: [count, key1, [fields1], key2, [fields2], ...] - for i in range(2, len(raw_result), 2): - row_data = raw_result[i] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - else: - # FT.AGGREGATE format: [count, [fields1], [fields2], ...] - for row_data in raw_result[1:]: - row_data = row_data or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - - return QueryResult(rows=rows, count=count) + return _parse_reply(raw_result, translated) -class AsyncExecutor(_ScoreParseMixin): +class AsyncExecutor: """Async version of Executor for use with redis.asyncio clients.""" def __init__( @@ -388,16 +498,14 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: # Replace the $vector PARAMS value with actual bytes (query/VSIM # references to $vector stay as parameter references). if vector_param: - self._inject_vector_param(cmd, vector_param) + _inject_vector_param(cmd, vector_param) # Execute command asynchronously try: raw_result = await self._client.execute_command(*cmd) except redis.ResponseError as e: error_msg = str(e) - if translated.command == "FT.HYBRID" and self._is_unknown_command( - error_msg - ): + if translated.command == "FT.HYBRID" and _is_unknown_command(error_msg): raise redis.ResponseError( f"{error_msg}. hybrid_vector_search() translates to FT.HYBRID, " "which requires Redis 8.4+ (RediSearch with hybrid search) " @@ -420,65 +528,7 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult: ) from e raise - # Parse result based on command type - # FT.SEARCH/FT.AGGREGATE replies are arrays with the count first; - # FT.HYBRID replies are maps (dict) and set count during parsing below. - count = raw_result[0] if isinstance(raw_result, list) and raw_result else 0 - rows: list[dict] = [] - - if translated.command == "FT.HYBRID": - count, rows = self._parse_hybrid_reply(raw_result) - elif translated.command == "FT.SEARCH": - with_scores = translated.score_alias is not None - no_content = self._has_return_0(translated.args) - - score_alias: str | None = None - - if with_scores and no_content: - # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...] - score_alias = self._resolve_score_alias( - translated.score_alias, translated.args - ) - for i in range(1, len(raw_result) - 1, 2): - score = raw_result[i + 1] - row = {score_alias: score} - rows.append(row) - elif with_scores: - # WITHSCORES format: [count, key1, score1, [fields1], ...] - # First pass: collect all field names across all rows so the - # alias avoids collisions with any document field. - all_field_names: set[str] = set() - parsed_rows: list[tuple[dict, Any]] = [] - for i in range(1, len(raw_result) - 2, 3): - score = raw_result[i + 1] - # A nil field-array (e.g. doc expired mid-query) becomes an - # empty field set, keeping the row's score instead of crashing. - row_data = raw_result[i + 2] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - all_field_names.update(row.keys()) - parsed_rows.append((row, score)) - resolved_alias = self._resolve_score_alias( - translated.score_alias, - translated.args, - first_row_fields=all_field_names, - ) - for row, score in parsed_rows: - row[resolved_alias] = score - rows.append(row) - else: - # Standard format: [count, key1, [fields1], key2, [fields2], ...] - for i in range(2, len(raw_result), 2): - row_data = raw_result[i] or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - else: - # FT.AGGREGATE format: [count, [fields1], [fields2], ...] - for row_data in raw_result[1:]: - row_data = row_data or [] - row = dict(zip(row_data[::2], row_data[1::2])) - rows.append(row) - - return QueryResult(rows=rows, count=count) + return _parse_reply(raw_result, translated) def create_executor( diff --git a/sql_redis/schema.py b/sql_redis/schema.py index 1d89f31..14402c5 100644 --- a/sql_redis/schema.py +++ b/sql_redis/schema.py @@ -20,7 +20,7 @@ def _extract_attributes(info: Any) -> list: """Pull the ``attributes`` section out of an FT.INFO reply. Handles both reply shapes: the RESP2 flat list - (``[..., 'attributes', [...], ...]``) and the redis-py 8.x / RESP3 map + (``[..., 'attributes', [...], ...]``) and the RESP3 map (``{b'attributes': [...], ...}``), whose keys may be bytes or str. """ if isinstance(info, dict): @@ -66,7 +66,7 @@ def _parse_schema_from_info(info: Any) -> dict[str, str]: This is a pure function with no I/O operations, shared by both the sync and async schema registries. It accepts both the RESP2 list reply and the - redis-py 8.x / RESP3 map reply (see ``_extract_attributes``). + RESP3 map reply (see ``_extract_attributes``). Args: info: The raw response from the FT.INFO command (list or dict). diff --git a/tests/conftest.py b/tests/conftest.py index c0e7524..d20414f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,11 +1,15 @@ """Pytest configuration and fixtures.""" import struct +from unittest.mock import AsyncMock, MagicMock import pytest import redis from testcontainers.redis import RedisContainer +from sql_redis.executor import AsyncExecutor, Executor +from sql_redis.translator import TranslatedQuery + @pytest.fixture(scope="module") def redis_container(): @@ -358,3 +362,59 @@ def items_data(redis_client: redis.Redis, items_index: str): binary_client.close() return items_index + + +def make_stub_executor(translated: TranslatedQuery, raw_result) -> Executor: + """Build an Executor whose translation and Redis reply are stubbed. + + Feeds a crafted reply straight through the parser, so every parse branch + can be exercised deterministically without a live Redis server. Pass an + exception as ``raw_result`` to exercise the error paths. Exposed as the + ``stub_executor`` fixture. + """ + executor = Executor.__new__(Executor) + executor._client = MagicMock() + if isinstance(raw_result, BaseException): + executor._client.execute_command.side_effect = raw_result + else: + executor._client.execute_command.return_value = raw_result + executor._schema_registry = MagicMock() + executor._translator = MagicMock() + executor._translator.translate.return_value = translated + return executor + + +def make_stub_async_executor(translated: TranslatedQuery, raw_result) -> AsyncExecutor: + """Build an AsyncExecutor whose translation and Redis reply are stubbed. + + The async seam differs from the sync one: ``AsyncExecutor.execute`` parses + first, awaits ``ensure_schema`` when the query names an index, and only then + translates the pre-parsed result. Pass an exception as ``raw_result`` to + exercise the error paths. Exposed as the ``stub_async_executor`` fixture. + """ + executor = AsyncExecutor.__new__(AsyncExecutor) + executor._client = MagicMock() + if isinstance(raw_result, BaseException): + executor._client.execute_command = AsyncMock(side_effect=raw_result) + else: + executor._client.execute_command = AsyncMock(return_value=raw_result) + executor._schema_registry = MagicMock() + executor._schema_registry.ensure_schema = AsyncMock() + executor._translator = MagicMock() + parsed = MagicMock() + parsed.index = None # skip ensure_schema() + executor._translator.parse.return_value = parsed + executor._translator.translate_parsed.return_value = translated + return executor + + +@pytest.fixture +def stub_executor(): + """Factory fixture returning :func:`make_stub_executor`.""" + return make_stub_executor + + +@pytest.fixture +def stub_async_executor(): + """Factory fixture returning :func:`make_stub_async_executor`.""" + return make_stub_async_executor diff --git a/tests/test_ft_hybrid.py b/tests/test_ft_hybrid.py index 46b3a1d..ca3db45 100644 --- a/tests/test_ft_hybrid.py +++ b/tests/test_ft_hybrid.py @@ -48,6 +48,8 @@ from sql_redis.schema import SchemaRegistry from sql_redis.translator import Translator +pytestmark = pytest.mark.protocol + def float_vector_to_bytes(vector: list[float]) -> bytes: """Convert a list of floats to binary format for Redis vector storage.""" diff --git a/tests/test_nil_field_array.py b/tests/test_nil_field_array.py index ddce90d..5e3b9da 100644 --- a/tests/test_nil_field_array.py +++ b/tests/test_nil_field_array.py @@ -9,42 +9,16 @@ These tests feed crafted replies (with a nil field-array) straight through the parser by mocking the translator and client, so every parse branch is exercised -deterministically without a live Redis server. +deterministically without a live Redis server. The ``stub_executor`` / +``stub_async_executor`` factory fixtures live in ``conftest.py``; the RESP3 +equivalents of these cases are in ``test_reply_shapes.py``. """ -from unittest.mock import AsyncMock, MagicMock - import pytest -from sql_redis.executor import AsyncExecutor, Executor from sql_redis.translator import TranslatedQuery - -def _make_sync_executor(translated: TranslatedQuery, raw_result): - """Build an Executor whose translation and Redis reply are stubbed.""" - executor = Executor.__new__(Executor) - executor._client = MagicMock() - executor._client.execute_command.return_value = raw_result - executor._schema_registry = MagicMock() - executor._translator = MagicMock() - executor._translator.translate.return_value = translated - return executor - - -def _make_async_executor(translated: TranslatedQuery, raw_result): - """Build an AsyncExecutor whose translation and Redis reply are stubbed.""" - executor = AsyncExecutor.__new__(AsyncExecutor) - executor._client = MagicMock() - executor._client.execute_command = AsyncMock(return_value=raw_result) - executor._schema_registry = MagicMock() - executor._schema_registry.ensure_schema = AsyncMock() - executor._translator = MagicMock() - # AsyncExecutor.execute parses first, then translates the parsed result. - parsed = MagicMock() - parsed.index = None # skip ensure_schema() - executor._translator.parse.return_value = parsed - executor._translator.translate_parsed.return_value = translated - return executor +pytestmark = pytest.mark.protocol class TestStandardSearchNilFields: @@ -57,19 +31,19 @@ def _translated(self) -> TranslatedQuery: query_string="*", ) - def test_sync_tolerates_nil_field_array(self): + def test_sync_tolerates_nil_field_array(self, stub_executor): # Second document's field-array came back nil. raw_result = [2, "product:1", ["title", "Laptop"], "product:2", None] - executor = _make_sync_executor(self._translated(), raw_result) + executor = stub_executor(self._translated(), raw_result) result = executor.execute("SELECT * FROM products") assert result.count == 2 assert result.rows == [{"title": "Laptop"}, {}] - async def test_async_tolerates_nil_field_array(self): + async def test_async_tolerates_nil_field_array(self, stub_async_executor): raw_result = [2, "product:1", ["title", "Laptop"], "product:2", None] - executor = _make_async_executor(self._translated(), raw_result) + executor = stub_async_executor(self._translated(), raw_result) result = await executor.execute("SELECT * FROM products") @@ -88,7 +62,7 @@ def _translated(self) -> TranslatedQuery: score_alias="score", ) - def test_sync_keeps_score_when_fields_nil(self): + def test_sync_keeps_score_when_fields_nil(self, stub_executor): raw_result = [ 2, "product:1", @@ -98,7 +72,7 @@ def test_sync_keeps_score_when_fields_nil(self): "0.9", None, ] - executor = _make_sync_executor(self._translated(), raw_result) + executor = stub_executor(self._translated(), raw_result) result = executor.execute("SELECT * FROM products") @@ -108,7 +82,7 @@ def test_sync_keeps_score_when_fields_nil(self): {"score": "0.9"}, ] - async def test_async_keeps_score_when_fields_nil(self): + async def test_async_keeps_score_when_fields_nil(self, stub_async_executor): raw_result = [ 2, "product:1", @@ -118,7 +92,7 @@ async def test_async_keeps_score_when_fields_nil(self): "0.9", None, ] - executor = _make_async_executor(self._translated(), raw_result) + executor = stub_async_executor(self._translated(), raw_result) result = await executor.execute("SELECT * FROM products") @@ -139,18 +113,18 @@ def _translated(self) -> TranslatedQuery: query_string="*", ) - def test_sync_tolerates_nil_row(self): + def test_sync_tolerates_nil_row(self, stub_executor): raw_result = [2, ["category", "books"], None] - executor = _make_sync_executor(self._translated(), raw_result) + executor = stub_executor(self._translated(), raw_result) result = executor.execute("SELECT category FROM products GROUP BY category") assert result.count == 2 assert result.rows == [{"category": "books"}, {}] - async def test_async_tolerates_nil_row(self): + async def test_async_tolerates_nil_row(self, stub_async_executor): raw_result = [2, ["category", "books"], None] - executor = _make_async_executor(self._translated(), raw_result) + executor = stub_async_executor(self._translated(), raw_result) result = await executor.execute( "SELECT category FROM products GROUP BY category" diff --git a/tests/test_protocol_matrix.py b/tests/test_protocol_matrix.py new file mode 100644 index 0000000..a52a1df --- /dev/null +++ b/tests/test_protocol_matrix.py @@ -0,0 +1,306 @@ +"""End-to-end tests across RESP protocols and redis-py client modes. + +``sql_redis`` reads raw FT.* replies, so the shape it parses is decided by the +negotiated wire protocol rather than by anything the library configures. Four +client modes matter, not two: + +* ``protocol=2`` - flat arrays, the historical behaviour +* ``protocol`` unset - RESP2 on redis-py 6.x/7.x, RESP3 from redis-py 8 +* ``protocol=3`` - RESP3 maps on every version +* ``legacy_responses=False`` - RESP3, redis-py 8 only + +These tests assert against a real server, because reply shapes are the subject +and a mock would encode the assumption under test. ``test_reply_shapes.py`` +pins the exact map literals; this module proves those are the shapes Redis +actually sends and that the rows come out identical either way. + +The fixtures here are deliberately local. Parametrizing the shared +``redis_client`` fixture in ``conftest.py`` would multiply thirteen other +modules, and three of them assert raw positional replies against that client. +""" + +import pytest +import redis +import redis.asyncio as async_redis + +from sql_redis import create_async_executor, create_executor + +pytestmark = pytest.mark.protocol + +INDEX = "protocol_matrix" +PREFIX = "pm:" + +ROWS = [ + ("pm:1", {"title": "gaming laptop", "category": "electronics", "price": 1499}), + ("pm:2", {"title": "laptop stand", "category": "office", "price": 45}), + ("pm:3", {"title": "sql cookbook", "category": "books", "price": 39}), + ("pm:4", {"title": "redis in action", "category": "books", "price": 32}), +] + +PLAIN_SQL = ( + f"SELECT title, price FROM {INDEX} WHERE category = 'books' ORDER BY price ASC" +) +AGGREGATE_SQL = f"SELECT category, COUNT(*) AS cnt, SUM(price) AS total FROM {INDEX} GROUP BY category" +SCORE_SQL = ( + f"SELECT title, score() AS relevance FROM {INDEX} WHERE fulltext(title, 'laptop')" +) +LIMITED_SQL = f"SELECT title FROM {INDEX} ORDER BY price ASC LIMIT 2" +COMPUTED_SQL = f"SELECT price * 2 AS double_price FROM {INDEX}" + +_REDIS_PY_MAJOR = int(redis.__version__.split(".")[0]) + +CLIENT_MODES = [ + pytest.param({"protocol": 2}, id="protocol-2"), + pytest.param({}, id="default"), + pytest.param({"protocol": 3}, id="protocol-3"), + pytest.param( + {"legacy_responses": False}, + id="new-response-format", + marks=pytest.mark.skipif( + _REDIS_PY_MAJOR < 8, reason="legacy_responses requires redis-py 8" + ), + ), +] + + +@pytest.fixture(scope="module") +def endpoint(redis_container) -> tuple[str, int]: + """Host and port of the shared test container.""" + return ( + redis_container.get_container_host_ip(), + int(redis_container.get_exposed_port(6379)), + ) + + +@pytest.fixture(scope="module") +def matrix_index(endpoint) -> str: + """Create the index and load its documents over a plain RESP2 client. + + Setup deliberately does not depend on the protocol under test, so the data + is materialized once no matter how many client modes are exercised. + """ + host, port = endpoint + client = redis.Redis(host=host, port=port, protocol=2, decode_responses=True) + try: + client.execute_command("FT.DROPINDEX", INDEX, "DD") + except redis.ResponseError: + pass + client.execute_command( + "FT.CREATE", + INDEX, + "ON", + "HASH", + "PREFIX", + "1", + PREFIX, + "SCHEMA", + "title", + "TEXT", + "SORTABLE", + "category", + "TAG", + "SORTABLE", + "price", + "NUMERIC", + "SORTABLE", + ) + for key, row in ROWS: + client.hset(key, mapping=row) + yield INDEX + try: + client.execute_command("FT.DROPINDEX", INDEX, "DD") + except redis.ResponseError: + pass + client.close() + + +def _client(endpoint, **kwargs) -> redis.Redis: + host, port = endpoint + return redis.Redis(host=host, port=port, decode_responses=True, **kwargs) + + +def _rows(endpoint, sql, **kwargs) -> tuple[list[dict], int]: + client = _client(endpoint, **kwargs) + try: + result = create_executor(client).execute(sql) + return result.rows, result.count + finally: + client.close() + + +class TestReplyShapeCanary: + """The protocols really do send different shapes. + + Without this, the redis-py compatibility CI job could report green having + never produced a map at all. + """ + + @pytest.mark.parametrize("protocol,expected", [(2, list), (3, dict)]) + @pytest.mark.parametrize("command", ["FT.SEARCH", "FT.AGGREGATE"]) + def test_raw_reply_type_follows_protocol( + self, endpoint, matrix_index, protocol, expected, command + ): + client = _client(endpoint, protocol=protocol) + try: + if command == "FT.SEARCH": + raw = client.execute_command( + command, matrix_index, "*", "LIMIT", "0", "2" + ) + else: + raw = client.execute_command( + command, matrix_index, "*", "GROUPBY", "1", "@category" + ) + finally: + client.close() + + assert isinstance(raw, expected) + + def test_unset_protocol_follows_redis_py_default(self, endpoint, matrix_index): + """redis-py 8 defaults to RESP3; 6.x and 7.x default to RESP2. + + This is the claim docs/concepts/result-shape.md makes about an upgrade + changing the reply shape without the caller passing ``protocol``. + """ + client = _client(endpoint) + try: + raw = client.execute_command( + "FT.SEARCH", matrix_index, "*", "LIMIT", "0", "2" + ) + finally: + client.close() + + assert isinstance(raw, dict if _REDIS_PY_MAJOR >= 8 else list) + + +class TestRowsMatchAcrossClientModes: + """Every client mode yields the rows the protocol=2 baseline yields.""" + + @pytest.mark.parametrize("client_kwargs", CLIENT_MODES) + def test_search_rows_match(self, endpoint, matrix_index, client_kwargs): + baseline, baseline_count = _rows(endpoint, PLAIN_SQL, protocol=2) + rows, count = _rows(endpoint, PLAIN_SQL, **client_kwargs) + + assert ( + rows + == baseline + == [ + {"title": "redis in action", "price": "32"}, + {"title": "sql cookbook", "price": "39"}, + ] + ) + assert count == baseline_count == 2 + + @pytest.mark.parametrize("client_kwargs", CLIENT_MODES) + def test_aggregate_rows_match(self, endpoint, matrix_index, client_kwargs): + baseline, baseline_count = _rows(endpoint, AGGREGATE_SQL, protocol=2) + rows, count = _rows(endpoint, AGGREGATE_SQL, **client_kwargs) + + # RediSearch does not guarantee group order without a SORTBY, so + # compare by group key rather than positionally. + by_category = {row["category"]: row for row in rows} + assert by_category == {row["category"]: row for row in baseline} + assert sorted(by_category) == ["books", "electronics", "office"] + assert by_category["books"] == {"category": "books", "cnt": "2", "total": "71"} + assert count == baseline_count == 3 + + @pytest.mark.parametrize("client_kwargs", CLIENT_MODES) + def test_count_is_the_total_not_the_row_count( + self, endpoint, matrix_index, client_kwargs + ): + """LIMIT reduces the rows, not the count callers paginate on.""" + rows, count = _rows(endpoint, LIMITED_SQL, **client_kwargs) + + assert count == 4 + assert [row["title"] for row in rows] == ["redis in action", "sql cookbook"] + + +class TestScoreValueType: + """The score's type differs by protocol, and that is accepted.""" + + def test_score_type_diverges_by_protocol(self, endpoint, matrix_index): + resp2_rows, _ = _rows(endpoint, SCORE_SQL, protocol=2) + resp3_rows, _ = _rows(endpoint, SCORE_SQL, protocol=3) + + assert isinstance(resp2_rows[0]["relevance"], str) + assert isinstance(resp3_rows[0]["relevance"], float) + # float() is the protocol-safe conversion at the call site. + assert [float(r["relevance"]) for r in resp2_rows] == pytest.approx( + [float(r["relevance"]) for r in resp3_rows] + ) + assert {r["title"] for r in resp2_rows} == {r["title"] for r in resp3_rows} + + +class TestAggregateCountDivergesByProtocol: + """``count`` is protocol-dependent for an FT.AGGREGATE without GROUPBY. + + A computed-field projection produces a LOAD/APPLY-only pipeline, where + RESP2 leaves the leading integer at the placeholder the Redis docs call + "not a valid value" while RESP3 reports the real figure in + ``total_results``. Neither can be derived from the other without changing + the protocol=2 output callers already depend on, so both are surfaced as + Redis sent them. Documented in docs/concepts/result-shape.md. + """ + + def test_placeholder_on_resp2_real_total_on_resp3(self, endpoint, matrix_index): + resp2_rows, resp2_count = _rows(endpoint, COMPUTED_SQL, protocol=2) + resp3_rows, resp3_count = _rows(endpoint, COMPUTED_SQL, protocol=3) + + assert len(resp2_rows) == len(resp3_rows) == 4 + assert resp2_count == 1 + assert resp3_count == 4 + + def test_rows_still_match(self, endpoint, matrix_index): + """Only the count differs. The rows are identical.""" + resp2_rows, _ = _rows(endpoint, COMPUTED_SQL, protocol=2) + resp3_rows, _ = _rows(endpoint, COMPUTED_SQL, protocol=3) + + assert sorted(r["double_price"] for r in resp2_rows) == sorted( + r["double_price"] for r in resp3_rows + ) + + +class TestDecodeResponsesOff: + """A bytes-keyed RESP3 map is the shape that broke redis-py itself.""" + + def test_bytes_keys_still_yield_rows(self, endpoint, matrix_index): + host, port = endpoint + client = redis.Redis(host=host, port=port, protocol=3) + try: + result = create_executor(client).execute(PLAIN_SQL) + finally: + client.close() + + assert result.count == 2 + assert result.rows == [ + {b"title": b"redis in action", b"price": b"32"}, + {b"title": b"sql cookbook", b"price": b"39"}, + ] + + +class TestAsyncAcrossProtocols: + """The async executor has its own parse call site.""" + + @pytest.mark.parametrize("protocol", [2, 3]) + async def test_async_search_and_aggregate(self, endpoint, matrix_index, protocol): + host, port = endpoint + client = async_redis.Redis( + host=host, port=port, protocol=protocol, decode_responses=True + ) + try: + executor = await create_async_executor(client) + search = await executor.execute(PLAIN_SQL) + aggregate = await executor.execute(AGGREGATE_SQL) + finally: + await client.aclose() + + assert search.count == 2 + assert search.rows == [ + {"title": "redis in action", "price": "32"}, + {"title": "sql cookbook", "price": "39"}, + ] + assert aggregate.count == 3 + assert {row["category"] for row in aggregate.rows} == { + "books", + "electronics", + "office", + } diff --git a/tests/test_reply_shapes.py b/tests/test_reply_shapes.py new file mode 100644 index 0000000..94097c6 --- /dev/null +++ b/tests/test_reply_shapes.py @@ -0,0 +1,566 @@ +"""Reply-shape tests for RESP2 arrays and RESP3 maps. + +The reply shape follows the protocol the client negotiated: a flat array on +RESP2, a map on RESP3. See ``docs/for-ais-only/FAILURE_MODES.md`` for why the +executor sees the raw shape at all. + +Every canned reply below was captured from a real Redis 8.4.6 server rather +than hand-written, with the producing cell named in a comment. Reply shapes are +the whole subject here, so an invented literal would only encode the assumption +under test. Replies marked "derived" are a captured reply with one documented +mutation, to reach a case the server does not produce on demand. + +``test_protocol_matrix.py`` proves these are the shapes Redis really sends; the +nil-field-array cases for the array shape live in ``test_nil_field_array.py``. +""" + +import pytest +import redis + +pytestmark = pytest.mark.protocol + +from sql_redis.translator import TranslatedQuery + +# Captured replies. +# +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT title, price FROM verify_products WHERE category = 'books' +# ORDER BY price ASC LIMIT 10 +RESP3_SEARCH = { + "attributes": [], + "format": "STRING", + "results": [ + { + "id": "vp:5", + "extra_attributes": {"price": "32", "title": "redis in action"}, + "values": [], + }, + { + "id": "vp:4", + "extra_attributes": {"price": "39", "title": "sql cookbook"}, + "values": [], + }, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=False, redis:8.4 +# Same query. Structural keys are bytes; this is the shape that made redis-py's +# own Result.from_resp3 return zero rows (redis-py#4107). +RESP3_SEARCH_BYTES = { + b"attributes": [], + b"format": b"STRING", + b"results": [ + { + b"id": b"vp:5", + b"extra_attributes": {b"price": b"32", b"title": b"redis in action"}, + b"values": [], + }, + { + b"id": b"vp:4", + b"extra_attributes": {b"price": b"39", b"title": b"sql cookbook"}, + b"values": [], + }, + ], + b"total_results": 2, + b"warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT title, score() AS relevance FROM verify_products +# WHERE fulltext(title, 'laptop') +RESP3_SEARCH_WITHSCORES = { + "attributes": [], + "format": "STRING", + "results": [ + { + "id": "vp:1", + "score": 0.714285696039395, + "extra_attributes": {"title": "gaming laptop"}, + "values": [], + }, + { + "id": "vp:3", + "score": 0.714285696039395, + "extra_attributes": {"title": "laptop stand"}, + "values": [], + }, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT score() AS relevance FROM verify_products WHERE fulltext(title, 'laptop') +# RETURN 0 suppresses document fields, so results carry no extra_attributes. +RESP3_SEARCH_SCORE_ONLY = { + "attributes": [], + "format": "STRING", + "results": [ + {"id": "vp:1", "score": 0.714285696039395, "values": []}, + {"id": "vp:3", "score": 0.714285696039395, "values": []}, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT category, COUNT(*) AS cnt, SUM(price) AS total, AVG(rating) AS avg_rating +# FROM verify_products GROUP BY category +# Aggregate rows carry only extra_attributes: no id, no score. +RESP3_AGGREGATE = { + "attributes": [], + "format": "STRING", + "results": [ + { + "extra_attributes": { + "category": "books", + "cnt": "2", + "total": "71", + "avg_rating": "4.7", + }, + "values": [], + }, + { + "extra_attributes": { + "category": "office", + "cnt": "2", + "total": "67", + "avg_rating": "3.95", + }, + "values": [], + }, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT title FROM verify_products WHERE category = 'nonexistent' +RESP3_SEARCH_EMPTY = { + "attributes": [], + "format": "STRING", + "results": [], + "total_results": 0, + "warning": [], +} + +# Derived from RESP3_SEARCH: extra_attributes dropped from the second result, +# the RESP3 analogue of the nil field-array of issue #38 (a document expiring +# between id selection and field materialization). +RESP3_SEARCH_MISSING_FIELDS = { + "attributes": [], + "format": "STRING", + "results": [ + { + "id": "vp:5", + "extra_attributes": {"price": "32", "title": "redis in action"}, + "values": [], + }, + {"id": "vp:4", "values": []}, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=3, decode_responses=True, redis:8.4 +# SELECT title FROM cnt_idx ORDER BY price ASC LIMIT 2, over 5 matching documents. +# total_results is the total match count, not the number of rows returned, so +# this is the capture that catches count being derived from len(results). +RESP3_SEARCH_LIMITED = { + "attributes": [], + "format": "STRING", + "results": [ + {"id": "ci:0", "extra_attributes": {"title": "book 0"}, "values": []}, + {"id": "ci:1", "extra_attributes": {"title": "book 1"}, "values": []}, + ], + "total_results": 5, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=2, decode_responses=True, redis:8.4 +# Same query. RESP2 puts the same total at position 0. +RESP2_SEARCH_LIMITED = [ + 5, + "ci:0", + ["title", "book 0"], + "ci:1", + ["title", "book 1"], +] + +# Cells: redis-py 8.1.0, protocol=2 and protocol=3, decode_responses=True, +# redis:8.4. SELECT price * 2 AS dbl FROM cnt_idx, which translates to +# FT.AGGREGATE ... APPLY with no GROUPBY. RESP2 sends the placeholder 1 that +# the Redis docs disclaim as "not a valid value"; RESP3 sends the real count. +# See TestAggregateCountDivergesByProtocol. +RESP2_AGGREGATE_COMPUTED = [ + 1, + ["price", "10", "dbl", "20"], + ["price", "11", "dbl", "22"], +] + +RESP3_AGGREGATE_COMPUTED = { + "attributes": [], + "format": "STRING", + "results": [ + {"extra_attributes": {"price": "10", "dbl": "20"}, "values": []}, + {"extra_attributes": {"price": "11", "dbl": "22"}, "values": []}, + ], + "total_results": 2, + "warning": [], +} + +# Cell: redis-py 8.1.0, protocol=2, decode_responses=True, redis:8.4 +# SELECT score() AS relevance FROM verify_products WHERE fulltext(title, 'laptop') +# The one array branch no other test covers. +RESP2_SEARCH_SCORE_ONLY = [ + 2, + "vp:1", + "0.714285696039395", + "vp:3", + "0.714285696039395", +] + + +# Translations matching those captures. +def _search() -> TranslatedQuery: + return TranslatedQuery( + command="FT.SEARCH", + index="verify_products", + query_string="@category:{books}", + args=["RETURN", "2", "title", "price", "LIMIT", "0", "10", "DIALECT", "2"], + ) + + +def _search_withscores() -> TranslatedQuery: + return TranslatedQuery( + command="FT.SEARCH", + index="verify_products", + query_string="@title:laptop", + args=["RETURN", "1", "title", "WITHSCORES", "SCORER", "BM25", "DIALECT", "2"], + score_alias="relevance", + ) + + +def _search_score_only() -> TranslatedQuery: + return TranslatedQuery( + command="FT.SEARCH", + index="verify_products", + query_string="@title:laptop", + args=["RETURN", "0", "WITHSCORES", "SCORER", "BM25", "DIALECT", "2"], + score_alias="relevance", + ) + + +def _search_limited() -> TranslatedQuery: + return TranslatedQuery( + command="FT.SEARCH", + index="cnt_idx", + query_string="*", + args=["RETURN", "1", "title", "LIMIT", "0", "2", "DIALECT", "2"], + ) + + +def _aggregate_computed() -> TranslatedQuery: + return TranslatedQuery( + command="FT.AGGREGATE", + index="cnt_idx", + query_string="*", + args=["APPLY", "@price * 2", "AS", "dbl", "DIALECT", "2"], + ) + + +def _aggregate() -> TranslatedQuery: + return TranslatedQuery( + command="FT.AGGREGATE", + index="verify_products", + query_string="*", + args=["GROUPBY", "1", "@category", "REDUCE", "COUNT", "0", "AS", "cnt"], + ) + + +@pytest.fixture(params=["sync", "async"]) +def execute_reply(request, stub_executor, stub_async_executor): + """Run one SQL query through either executor against a canned reply. + + Parametrized so each test is written once and asserted for both the sync + and async executors, whose parse paths are separate call sites. + """ + flavour = request.param + + async def _run(translated: TranslatedQuery, raw_result): + sql = f"SELECT * FROM {translated.index}" + if flavour == "sync": + return stub_executor(translated, raw_result).execute(sql) + return await stub_async_executor(translated, raw_result).execute(sql) + + return _run + + +class TestSearchResp3Map: + """A RESP3 FT.SEARCH map parses to the same rows as the RESP2 array.""" + + async def test_standard_search_map_reply_yields_field_rows(self, execute_reply): + result = await execute_reply(_search(), RESP3_SEARCH) + + assert result.count == 2 + assert result.rows == [ + {"price": "32", "title": "redis in action"}, + {"price": "39", "title": "sql cookbook"}, + ] + + async def test_structural_keys_may_be_bytes(self, execute_reply): + """A decode_responses=False client sends bytes structural keys. + + Document field keys stay bytes, matching the RESP2 path; looking up + ``extra_attributes`` as str alone would yield silently empty rows. + """ + result = await execute_reply(_search(), RESP3_SEARCH_BYTES) + + assert result.count == 2 + assert result.rows == [ + {b"price": b"32", b"title": b"redis in action"}, + {b"price": b"39", b"title": b"sql cookbook"}, + ] + + async def test_only_document_fields_become_columns(self, execute_reply): + """A new RESP3 result key must not become a column. + + Asserted as an allowlist rather than a denylist of known bookkeeping + keys, so a key RediSearch adds later fails here instead of silently + appearing in every row. + """ + result = await execute_reply(_search(), RESP3_SEARCH) + + for row in result.rows: + assert set(row) == {"price", "title"} + + async def test_count_is_the_total_not_the_row_count(self, execute_reply): + """``count`` comes from ``total_results``, which LIMIT does not reduce. + + Callers paginate on this (docs/concepts/result-shape.md), so deriving + it from the number of returned results would be silently wrong. + """ + result = await execute_reply(_search_limited(), RESP3_SEARCH_LIMITED) + + assert result.count == 5 + assert len(result.rows) == 2 + + async def test_missing_extra_attributes_yields_empty_row(self, execute_reply): + result = await execute_reply(_search(), RESP3_SEARCH_MISSING_FIELDS) + + assert result.count == 2 + assert result.rows == [{"price": "32", "title": "redis in action"}, {}] + + async def test_flat_field_array_is_tolerated(self, execute_reply): + """Defensive: a result whose fields arrive already flat still parses. + + No redis-py version sends this for FT.SEARCH, but ``_parse_hybrid_reply`` + accepts both forms for the same conceptual field, and a total + ``_field_array`` is what keeps an unexpected shape from silently + becoming an empty row. Derived from RESP3_SEARCH. + """ + reply = { + "attributes": [], + "format": "STRING", + "results": [{"id": "vp:5", "extra_attributes": ["title", "flat"]}], + "total_results": 1, + "warning": [], + } + + result = await execute_reply(_search(), reply) + + assert result.rows == [{"title": "flat"}] + + async def test_empty_results_yields_no_rows(self, execute_reply): + result = await execute_reply(_search(), RESP3_SEARCH_EMPTY) + + assert result.count == 0 + assert result.rows == [] + + +class TestSearchWithScoresResp3Map: + """The score comes from the result's own ``score`` key, not a stride.""" + + async def test_withscores_map_reply_adds_score_column(self, execute_reply): + result = await execute_reply(_search_withscores(), RESP3_SEARCH_WITHSCORES) + + assert result.count == 2 + assert result.rows == [ + {"title": "gaming laptop", "relevance": 0.714285696039395}, + {"title": "laptop stand", "relevance": 0.714285696039395}, + ] + + async def test_score_stays_a_float(self, execute_reply): + """RESP3 scores are doubles and are passed through, not stringified. + + RESP2 delivers the score as text, so the type differs by protocol. The + library does not normalise it: ``float()`` is the protocol-safe + conversion at the call site. See docs/concepts/result-shape.md. + """ + result = await execute_reply(_search_withscores(), RESP3_SEARCH_WITHSCORES) + + assert isinstance(result.rows[0]["relevance"], float) + + async def test_score_alias_collision_is_resolved(self, execute_reply): + """An alias colliding with a document field is renamed, as on RESP2.""" + translated = _search_withscores() + translated.score_alias = "title" + + result = await execute_reply(translated, RESP3_SEARCH_WITHSCORES) + + assert result.rows[0]["title"] == "gaming laptop" + assert result.rows[0]["__score_title"] == 0.714285696039395 + + async def test_withscores_return0_map_reply_yields_score_only_rows( + self, execute_reply + ): + result = await execute_reply(_search_score_only(), RESP3_SEARCH_SCORE_ONLY) + + assert result.count == 2 + assert result.rows == [ + {"relevance": 0.714285696039395}, + {"relevance": 0.714285696039395}, + ] + + +class TestAggregateResp3Map: + """A RESP3 FT.AGGREGATE map parses to rows of reduced values.""" + + async def test_aggregate_map_reply_yields_rows(self, execute_reply): + result = await execute_reply(_aggregate(), RESP3_AGGREGATE) + + assert result.count == 2 + assert result.rows == [ + {"category": "books", "cnt": "2", "total": "71", "avg_rating": "4.7"}, + {"category": "office", "cnt": "2", "total": "67", "avg_rating": "3.95"}, + ] + + async def test_missing_extra_attributes_yields_empty_row(self, execute_reply): + """The aggregate branch reaches the field flattener separately.""" + reply = { + "attributes": [], + "format": "STRING", + "results": [{"extra_attributes": {"category": "books"}}, {"values": []}], + "total_results": 2, + "warning": [], + } + + result = await execute_reply(_aggregate(), reply) + + assert result.rows == [{"category": "books"}, {}] + + +class TestAggregateCountDivergesByProtocol: + """For FT.AGGREGATE, ``count`` is protocol-dependent, and that is accepted. + + An aggregate pipeline with no GROUPBY (a computed field, a date function, a + geo_distance projection) leaves RESP2's leading integer as the placeholder + the Redis docs call "not a valid value", while RESP3's ``total_results`` + reports the real figure. Measured on Redis 8.4.6: the same SQL over 5 + documents gives count=1 on RESP2 and count=5 on RESP3. + + Neither value can be turned into the other without changing the protocol=2 + behaviour callers already depend on, so both are surfaced as Redis sent + them. This test exists to stop someone "fixing" the difference by deriving + count from len(rows), which would silently change RESP2 output. + """ + + async def test_resp2_reports_the_placeholder(self, execute_reply): + result = await execute_reply(_aggregate_computed(), RESP2_AGGREGATE_COMPUTED) + + assert result.count == 1 + assert len(result.rows) == 2 + + async def test_resp3_reports_the_real_total(self, execute_reply): + result = await execute_reply(_aggregate_computed(), RESP3_AGGREGATE_COMPUTED) + + assert result.count == 2 + assert len(result.rows) == 2 + + +class TestResp2ArrayUnchanged: + """The array path still parses the shape it always did.""" + + async def test_count_is_the_total_not_the_row_count(self, execute_reply): + result = await execute_reply(_search_limited(), RESP2_SEARCH_LIMITED) + + assert result.count == 5 + assert len(result.rows) == 2 + + async def test_withscores_return0_array_reply(self, execute_reply): + result = await execute_reply(_search_score_only(), RESP2_SEARCH_SCORE_ONLY) + + assert result.count == 2 + assert result.rows == [ + {"relevance": "0.714285696039395"}, + {"relevance": "0.714285696039395"}, + ] + + +class TestUnrecognizedReplies: + """A reply that is neither shape is rejected, not parsed into empty rows. + + A map that is not a result set (a cluster node-keyed reply, FT.PROFILE) or + an array that is not one (a WITHCURSOR pair) would otherwise fold to zero + rows and look like an empty result set. That silent-empty failure is the + one redis-py shipped in 8.0.0 (redis-py#4107). + """ + + async def test_map_without_results_raises(self, execute_reply): + node_keyed = {"127.0.0.1:6379": {"total_results": 2, "results": []}} + + with pytest.raises(ValueError, match="without a 'results' key"): + await execute_reply(_search(), node_keyed) + + async def test_profile_shaped_map_raises(self, execute_reply): + with pytest.raises(ValueError, match="without a 'results' key"): + await execute_reply(_search(), {"Results": {}, "Profile": {}}) + + async def test_non_array_non_map_reply_raises(self, execute_reply): + with pytest.raises(ValueError, match="Unrecognized FT.SEARCH reply"): + await execute_reply(_search(), "OK") + + async def test_empty_results_list_is_not_rejected(self, execute_reply): + """An empty result set is legitimate and must still parse.""" + result = await execute_reply(_search(), RESP3_SEARCH_EMPTY) + + assert result.count == 0 + assert result.rows == [] + + +class TestErrorRewrapping: + """Both executors re-wrap ResponseError with an actionable hint.""" + + async def test_ismissing_hint_is_added(self, execute_reply): + translated = TranslatedQuery( + command="FT.SEARCH", + index="verify_products", + query_string="ismissing(@email)", + args=["DIALECT", "2"], + ) + error = redis.ResponseError("Unknown function") + + with pytest.raises(redis.ResponseError, match="ismissing\\(\\) requires"): + await execute_reply(translated, error) + + async def test_unrelated_response_error_propagates_unchanged(self, execute_reply): + """Only the two known-cause errors are re-wrapped.""" + error = redis.ResponseError("no such index") + + with pytest.raises(redis.ResponseError, match="^no such index$"): + await execute_reply(_search(), error) + + async def test_hybrid_version_hint_is_added(self, execute_reply): + translated = TranslatedQuery( + command="FT.HYBRID", + index="verify_items", + query_string="", + args=["SEARCH", "@description:smartphone"], + is_hybrid=True, + ) + error = redis.ResponseError("unknown command 'FT.HYBRID'") + + with pytest.raises(redis.ResponseError, match="8.4"): + await execute_reply(translated, error) diff --git a/tests/test_schema_registry.py b/tests/test_schema_registry.py index 91683cc..2274d52 100644 --- a/tests/test_schema_registry.py +++ b/tests/test_schema_registry.py @@ -7,6 +7,8 @@ from sql_redis.schema import SchemaRegistry, _parse_schema_from_info +pytestmark = pytest.mark.protocol + def _create_test_indexes(redis_client: redis.Redis) -> list[str]: """Helper to create test indexes."""