Implement XDEL (#153)

Fix #150

docs/about/changelog.md

@@ -11,6 +11,7 @@ description: Change log of all fakeredis releases
 ### 🧰 Bug Fixes
 
 - Fixed xadd timestamp (fixes #151) (#152)
+- Implement XDEL #153
 
 ### 🧰 Maintenance
 

fakeredis/_stream.py

@@ -42,6 +42,20 @@ def __init__(self):
         # ]
         self._values = list()
 
+    def delete(self, lst: List[str]) -> int:
+        """Delete items from stream
+
+        :param lst: list of IDs to delete, in the form of timestamp-sequence.
+        :returns: Number of items deleted
+        """
+        res = 0
+        for item in lst:
+            ind, found = self.find_index(item)
+            if found:
+                del self._values[ind]
+                res += 1
+        return res
+
     def add(self, fields: List, id_str: str = '*') -> Union[None, bytes]:
         assert len(fields) % 2 == 0
         if isinstance(id_str, bytes):
@@ -84,6 +98,8 @@ def gen():
         return gen()
 
     def find_index(self, id_str: str) -> Tuple[int, bool]:
+        if len(self._values) == 0:
+            return 0, False
         ts_seq = StreamRangeTest.parse_id(id_str)
         ind = bisect.bisect_left(list(map(lambda x: x[0], self._values)), ts_seq)
         return ind, self._values[ind][0] == ts_seq

fakeredis/commands_mixins/streams_mixin.py

@@ -110,3 +110,12 @@ def xread(self, *args):
             return self._xread(stream_start_id_list, count, False)
         else:
             return self._blocking(timeout, functools.partial(self._xread, stream_start_id_list, count))
+
+    @command(name="XDEL", fixed=(Key(XStream),), repeat=(bytes,), )
+    def xdel(self, key, *args):
+        if len(args) == 0:
+            raise SimpleError(msgs.WRONG_ARGS_MSG6.format('xdel'))
+        if key.value is None:
+            return 0
+        res = key.value.delete(args)
+        return res

pyproject.toml

@@ -8,7 +8,7 @@ name = "fakeredis"
 packages = [
     { include = "fakeredis" },
 ]
-version = "2.12.2"
+version = "2.13.0"
 description = "Fake implementation of redis API for testing purposes."
 readme = "README.md"
 keywords = ["redis", "RedisJson", ]

test/conftest.py

@@ -66,6 +66,7 @@ def _create_redis(request) -> Callable[[int], redis.Redis]:
     server_version = request.getfixturevalue('real_redis_version')
     if not cls_name.startswith('Fake') and not server_version:
         pytest.skip('Redis is not running')
+    server_version = server_version or '6'
     min_server = _marker_version_value(request, 'min_server')
     max_server = _marker_version_value(request, 'max_server')
     if Version(server_version) < min_server:

test/test_mixins/test_streams_commands.py

@@ -3,6 +3,7 @@
 import pytest
 import redis
 
+from fakeredis import _msgs as msgs
 from fakeredis._stream import XStream
 from test import testtools
 
@@ -42,6 +43,12 @@ def test_xstream(r: redis.Redis):
     lst = stream.irange((0, 2), (3, 0))
     assert len(lst) == 4
 
+    stream = XStream()
+    assert stream.delete(['1']) == 0
+    id_str = stream.add([0, 0, 1, 1, 2, 2, 3, 3])
+    assert stream.delete([id_str, ]) == 1
+    assert len(stream) == 0
+
 
 @pytest.mark.max_server('6.3')
 def test_xadd_redis6(r: redis.Redis):
@@ -261,3 +268,23 @@ def test_xread_bad_commands(r: redis.Redis):
     with pytest.raises(redis.ResponseError) as ex2:
         testtools.raw_command(r, 'xread', 'streams', 'foo', )
     print(ex2)
+
+
+def test_xdel(r: redis.Redis):
+    stream = "stream"
+
+    # deleting from an empty stream doesn't do anything
+    assert r.xdel(stream, 1) == 0
+
+    m1 = r.xadd(stream, {"foo": "bar"})
+    m2 = r.xadd(stream, {"foo": "bar"})
+    m3 = r.xadd(stream, {"foo": "bar"})
+
+    # xdel returns the number of deleted elements
+    assert r.xdel(stream, m1) == 1
+    assert r.xdel(stream, m2, m3) == 2
+
+    with pytest.raises(redis.ResponseError) as ex:
+        testtools.raw_command(r, 'XDEL', stream)
+    assert ex.value.args[0] == msgs.WRONG_ARGS_MSG6.format('xdel')[4:]
+    assert r.xdel('non-existing-key', '1-1') == 0

test/test_redis_asyncio.py

@@ -245,6 +245,21 @@ async def test_disconnect_server(req_aioredis2, fake_server):
     fake_server.connected = True
 
 
+async def test_xdel(req_aioredis2: redis.asyncio.Redis):
+    stream = "stream"
+
+    # deleting from an empty stream doesn't do anything
+    assert await req_aioredis2.xdel(stream, 1) == 0
+
+    m1 = await req_aioredis2.xadd(stream, {"foo": "bar"})
+    m2 = await req_aioredis2.xadd(stream, {"foo": "bar"})
+    m3 = await req_aioredis2.xadd(stream, {"foo": "bar"})
+
+    # xdel returns the number of deleted elements
+    assert await req_aioredis2.xdel(stream, m1) == 1
+    assert await req_aioredis2.xdel(stream, m2, m3) == 2
+
+
 @pytest.mark.fake
 async def test_from_url():
     r0 = aioredis.FakeRedis.from_url('redis://localhost?db=0')