hub / github.com/celery/celery / get_active_redis_channels

Function get_active_redis_channels

t/integration/conftest.py:47–48 · view source on GitHub ↗

()

Source from the content-addressed store, hash-verified

45
46
47	def get_active_redis_channels():
48	return get_redis_connection().execute_command('PUBSUB CHANNELS')
49
50
51	def check_for_logs(caplog, message: str, max_wait: float = 1.0, interval: float = 0.1) -> bool:

test_redis_subscribed_channels_leakMethod · 0.85

test_ignoring_result_no_subscriptionsMethod · 0.85

test_asyncresult_forget_cancels_subscriptionMethod · 0.85

test_asyncresult_get_cancels_subscriptionMethod · 0.85

get_redis_connectionFunction · 0.90

no test coverage detected