Improve parsing of XINFO STREAM response (#3282)

Make the parsing of XINFO STREAM response more resilient, by
handling the case when first and last entries are None after XTRIM.

Improve the parsing of consumers related info.

redis/_parsers/helpers.py

@@ -275,17 +275,22 @@ def parse_xinfo_stream(response, **options):
         data = {str_if_bytes(k): v for k, v in response.items()}
     if not options.get("full", False):
         first = data.get("first-entry")
-        if first is not None:
+        if first is not None and first[0] is not None:
             data["first-entry"] = (first[0], pairs_to_dict(first[1]))
         last = data["last-entry"]
-        if last is not None:
+        if last is not None and last[0] is not None:
             data["last-entry"] = (last[0], pairs_to_dict(last[1]))
     else:
         data["entries"] = {_id: pairs_to_dict(entry) for _id, entry in data["entries"]}
-        if isinstance(data["groups"][0], list):
+        if len(data["groups"]) > 0 and isinstance(data["groups"][0], list):
             data["groups"] = [
                 pairs_to_dict(group, decode_keys=True) for group in data["groups"]
             ]
+            for g in data["groups"]:
+                if g["consumers"] and g["consumers"][0] is not None:
+                    g["consumers"] = [
+                        pairs_to_dict(c, decode_keys=True) for c in g["consumers"]
+                    ]
         else:
             data["groups"] = [
                 {str_if_bytes(k): v for k, v in group.items()}

tests/test_asyncio/test_commands.py

@@ -2919,6 +2919,31 @@ async def test_xinfo_stream(self, r: redis.Redis):
         assert info["first-entry"] == await get_stream_message(r, stream, m1)
         assert info["last-entry"] == await get_stream_message(r, stream, m2)
 
+        await r.xtrim(stream, 0)
+        info = await r.xinfo_stream(stream)
+        assert info["length"] == 0
+        assert info["first-entry"] is None
+        assert info["last-entry"] is None
+
+    @skip_if_server_version_lt("6.0.0")
+    async def test_xinfo_stream_full(self, r: redis.Redis):
+        stream = "stream"
+        group = "group"
+
+        await r.xadd(stream, {"foo": "bar"})
+        info = await r.xinfo_stream(stream, full=True)
+        assert info["length"] == 1
+        assert len(info["groups"]) == 0
+
+        await r.xgroup_create(stream, group, 0)
+        info = await r.xinfo_stream(stream, full=True)
+        assert info["length"] == 1
+
+        await r.xreadgroup(group, "consumer", streams={stream: ">"})
+        info = await r.xinfo_stream(stream, full=True)
+        consumer = info["groups"][0]["consumers"][0]
+        assert isinstance(consumer, dict)
+
     @skip_if_server_version_lt("5.0.0")
     async def test_xlen(self, r: redis.Redis):
         stream = "stream"

tests/test_commands.py

@@ -4452,14 +4452,23 @@ def test_xinfo_stream(self, r):
         assert info["entries-added"] == 2
         assert info["recorded-first-entry-id"] == m1
 
+        r.xtrim(stream, 0)
+        info = r.xinfo_stream(stream)
+        assert info["length"] == 0
+        assert info["first-entry"] is None
+        assert info["last-entry"] is None
+
     @skip_if_server_version_lt("6.0.0")
     def test_xinfo_stream_full(self, r):
         stream = "stream"
         group = "group"
         m1 = r.xadd(stream, {"foo": "bar"})
-        r.xgroup_create(stream, group, 0)
         info = r.xinfo_stream(stream, full=True)
+        assert info["length"] == 1
+        assert len(info["groups"]) == 0
 
+        r.xgroup_create(stream, group, 0)
+        info = r.xinfo_stream(stream, full=True)
         assert info["length"] == 1
         assert_resp_response_in(
             r,
@@ -4469,6 +4478,11 @@ def test_xinfo_stream_full(self, r):
         )
         assert len(info["groups"]) == 1
 
+        r.xreadgroup(group, "consumer", streams={stream: ">"})
+        info = r.xinfo_stream(stream, full=True)
+        consumer = info["groups"][0]["consumers"][0]
+        assert isinstance(consumer, dict)
+
     @skip_if_server_version_lt("5.0.0")
     def test_xlen(self, r):
         stream = "stream"