(self, r)
| 797 | |
| 798 | @pytest.mark.onlynoncluster |
| 799 | def test_pattern_message_handler(self, r): |
| 800 | p = r.pubsub(ignore_subscribe_messages=True) |
| 801 | p.psubscribe(**{"f*": self.message_handler}) |
| 802 | assert wait_for_message(p) is None |
| 803 | assert r.publish("foo", "test message") == 1 |
| 804 | assert wait_for_message(p) is None |
| 805 | assert self.message == make_message( |
| 806 | "pmessage", "foo", "test message", pattern="f*" |
| 807 | ) |
| 808 | |
| 809 | @pytest.mark.onlynoncluster |
| 810 | def test_binary_pattern_message_handler_with_subscription(self, r): |
nothing calls this directly
no test coverage detected