Method test_xreadgroup

tests/test_commands.py:7129–7208

(self, r)

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

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

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

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

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

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

        r.xgroup_destroy(stream, group)

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

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

        # xreadgroup with noack does not have any items in the PEL
        r.xgroup_destroy(stream, group)
        r.xgroup_create(stream, group, "0")
        res = r.xreadgroup(group, consumer, streams={stream: ">"}, noack=True)
        empty_res = 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
            assert len(empty_res[0][1]) == 0