Method test_xgroup_create_mkstream

tests/test_commands.py:6550–6573

(self, r)

    @skip_if_server_version_lt("7.0.0")
    def test_xgroup_create_mkstream(self, r):
        # tests xgroup_create and xinfo_groups
        stream = "stream"
        group = "group"

        # an error is raised if a group is created on a stream that
        # doesn't already exist
        with pytest.raises(exceptions.ResponseError):
            r.xgroup_create(stream, group, 0)

        # however, with mkstream=True, the underlying stream is created
        # automatically
        assert r.xgroup_create(stream, group, 0, mkstream=True)
        expected = [
            {
                "name": group.encode(),
                "consumers": 0,
                "pending": 0,
                "last-delivered-id": b"0-0",
                "entries-read": None,
                "lag": 0,
            }
        ]
        assert r.xinfo_groups(stream) == expected

    @skip_if_server_version_lt("7.0.0")
    @skip_if_server_version_gte("8.2.2")