fix: wrong-type list key is a TypeError; bytearray/memoryview stored as strings (#12)

redisdol/base.py

@@ -256,6 +256,24 @@ def __delitem__(self, k):
 from collections.abc import MutableSequence
 
 
+def _translate_list_response_error(e: ResponseError, what: str):
+    """The Python exception a failed list write maps to (to be raised ``from e``), or ``None``.
+
+    ``WRONGTYPE`` (the key holds a string, hash...) is a ``TypeError``; 'no such key' and
+    'index out of range' are an ``IndexError`` like a list's; anything else (READONLY,
+    OOM, ...) is not an index problem: ``None``, the caller re-raises it as it is. Only
+    the server's own words are read: a pipeline's message also quotes the command, and so
+    the key's name.
+    """
+    message = str(e).rsplit("caused error: ", 1)[-1]
+    if message.upper().startswith("WRONGTYPE"):
+        return TypeError(f"{what}: the key holds a Redis value that is not a list")
+    lowered = message.lower()
+    if "no such key" in lowered or "index out of range" in lowered:
+        return IndexError(f"{what} index out of range")
+    return None
+
+
 # TODO: Redis list type is more of a linked-list than a list. Better subclass deque?
 class RedisList(MutableSequence):
     """
@@ -324,8 +342,11 @@ def __setitem__(self, i, v):
         i = operator.index(i)
         try:
             self._source.lset(self._name, i, self._encode(v))
-        except ResponseError as e:  # redis: 'no such key' / 'index out of range'
-            raise IndexError("RedisList assignment index out of range") from e
+        except ResponseError as e:  # redis: 'no such key' / 'index out of range' / WRONGTYPE
+            translated = _translate_list_response_error(e, "RedisList assignment")
+            if translated is None:
+                raise
+            raise translated from e
 
     def __delitem__(self, i):
         """Delete the element at position ``i`` (atomically).
@@ -344,7 +365,10 @@ def __delitem__(self, i):
                 pipe.lrem(self._name, 1, sentinel)
                 pipe.execute()
         except ResponseError as e:
-            raise IndexError("RedisList assignment index out of range") from e
+            translated = _translate_list_response_error(e, "RedisList deletion")
+            if translated is None:
+                raise
+            raise translated from e
 
     def append(self, v):
         # rpush (not rpushx): appending to a not-yet-existing list creates it,
@@ -432,6 +456,8 @@ def __getitem__(self, k):
 
 class RedisPersister(RedisCollection, RedisBytesPersister):
     def __setitem__(self, k, v):
+        if isinstance(v, (bytearray, memoryview)):
+            v = bytes(v)  # bytes-like: a redis string, not a list of ints
         if isinstance(v, Iterable) and not isinstance(v, (str, bytes)):
             encode = self._list_codec("list_encode") or _identity
             values = [encode(x) for x in v]  # also consumes one-shot iterables

tests/test_wrapping_fakeredis.py

@@ -371,3 +371,85 @@ def __init__(self, store, tag="T"):
     assert s["k"] == b"v"
     cls2 = wrap_kvs(RedisPersister, wrapper=Tagged, obj_of_data=lambda v: v)
     assert store_from_source(FakeRedis(), cls2).tag == "T"
+
+
+# ---- wrong-type keys and bytes-like values (i2mint/redisdol#12) -------------------------
+
+
+def test_list_ops_on_a_string_key_are_a_wrong_type_not_an_index_error():
+    r = FakeRedis()
+    r.set("s", "text")
+    lst = RedisList(r, "s")
+    with pytest.raises(TypeError, match="not a list"):
+        lst[0] = "x"
+    with pytest.raises(TypeError, match="not a list"):
+        del lst[0]
+    assert r.get("s") == b"text"  # untouched
+
+
+def test_list_ops_out_of_range_or_missing_are_still_index_errors():
+    r = FakeRedis()
+    r.rpush("l", "a")
+    lst = RedisList(r, "l")
+    for op in (lambda: lst.__setitem__(5, "x"), lambda: lst.__delitem__(5)):
+        with pytest.raises(IndexError):
+            op()
+    missing = RedisList(r, "nope")
+    with pytest.raises(IndexError):
+        missing[0] = "x"
+    with pytest.raises(IndexError):
+        del missing[0]
+
+
+def test_other_redis_errors_are_not_disguised_as_index_errors():
+    from redis.exceptions import ResponseError
+
+    class Failing:
+        def lset(self, *a):
+            raise ResponseError("READONLY You can't write against a read only replica.")
+
+    with pytest.raises(ResponseError, match="READONLY"):
+        RedisList(Failing(), "l")[0] = "x"
+
+
+@pytest.mark.parametrize("make", [bytearray, memoryview])
+def test_bytes_like_values_are_stored_as_a_string_not_a_list_of_ints(make):
+    s = _store()
+    s["k"] = make(b"hello")
+    assert s["k"] == b"hello"
+    assert s._source.type("k") == b"string"
+
+
+def test_key_names_do_not_change_how_a_pipeline_error_is_classified():
+    from redis.exceptions import ResponseError
+
+    r = FakeRedis()
+    with pytest.raises(IndexError):  # a missing key that is *named* WRONGTYPE
+        del RedisList(r, "WRONGTYPE")[0]
+
+    class ReadOnlyPipe:  # a READONLY error on a key named like a list error
+        def pipeline(self, transaction=True):
+            class P:
+                def __enter__(s):
+                    return s
+
+                def __exit__(s, *a):
+                    return False
+
+                def lset(s, *a):
+                    pass
+
+                def lrem(s, *a):
+                    pass
+
+                def execute(s):
+                    raise ResponseError(
+                        "Command # 1 (LSET no such key 0 x) of pipeline caused error: "
+                        "READONLY You can't write against a read only replica."
+                    )
+
+            return P()
+
+    with pytest.raises(ResponseError, match="READONLY") as info:
+        del RedisList(ReadOnlyPipe(), "no such key")[0]
+    assert info.value.__cause__ is None  # not its own cause