Method test_push_handler

tests/test_pubsub.py:865–874

(self, r)

        self.message = ["my handler", message]

    def test_push_handler(self, r):
        if is_resp2_connection(r):
            return
        p = r.pubsub(push_handler_func=self.my_handler)
        p.subscribe("foo")
        assert wait_for_message(p) is None
        assert self.message == ["my handler", [b"subscribe", b"foo", 1]]
        assert r.publish("foo", "test message") == 1
        assert wait_for_message(p) is None
        assert self.message == ["my handler", [b"message", b"foo", b"test message"]]

    @skip_if_server_version_lt("7.0.0")
    def test_push_handler_sharded_pubsub(self, r):