Method test_xreadgroup

tests/test_asyncio/test_commands.py:5308–5389

(self, r: redis.Redis)

    @skip_if_server_version_lt("5.0.0")
    async def test_xreadgroup(self, r: redis.Redis):
        stream = "stream"
        group = "group"
        consumer = "consumer"
        m1 = await r.xadd(stream, {"foo": "bar"})
        m2 = await r.xadd(stream, {"bing": "baz"})
        await r.xgroup_create(stream, group, 0)

        strem_name = stream.encode()
        expected_entries = [
            await get_stream_message(r, stream, m1),
            await get_stream_message(r, stream, m2),
        ]

        # xread starting at 0 returns both messages
        res = await r.xreadgroup(group, consumer, streams={stream: ">"})
        assert_resp_response(
            r,
            res,
            [[strem_name, expected_entries]],
            {strem_name: [expected_entries]},
            {strem_name: expected_entries},
        )

        await r.xgroup_destroy(stream, group)
        await r.xgroup_create(stream, group, 0)

        expected_entries = [await get_stream_message(r, stream, m1)]

        # xread with count=1 returns only the first message
        res = await r.xreadgroup(group, consumer, streams={stream: ">"}, count=1)
        assert_resp_response(
            r,
            res,
            [[strem_name, expected_entries]],
            {strem_name: [expected_entries]},
            {strem_name: expected_entries},
        )

        await r.xgroup_destroy(stream, group)

        # create the group using $ as the last id meaning subsequent reads
        # will only find messages added after this
        await r.xgroup_create(stream, group, "$")

        # xread starting after the last message returns an empty message list
        res = await r.xreadgroup(group, consumer, streams={stream: ">"})
        assert_resp_response(r, res, [], {})

        # xreadgroup with noack does not have any items in the PEL
        await r.xgroup_destroy(stream, group)
        await r.xgroup_create(stream, group, "0")
        res = await r.xreadgroup(group, consumer, streams={stream: ">"}, noack=True)
        empty_res = await r.xreadgroup(group, consumer, streams={stream: "0"})
        shape = expected_response_shape(r)
        if shape == "legacy_resp2":
            assert len(res[0][1]) == 2
            # now there should be nothing pending