|
| 1 | +"""Integration tests for native delayed delivery queue binding. |
| 2 | +
|
| 3 | +Tests that verify queue bindings are created correctly for native delayed |
| 4 | +delivery, especially when some queues in task_queues fail to bind. |
| 5 | +""" |
| 6 | +import os |
| 7 | +import uuid |
| 8 | +from urllib.parse import quote |
| 9 | + |
| 10 | +import pytest |
| 11 | +import requests |
| 12 | +from kombu import Exchange, Queue |
| 13 | +from requests.auth import HTTPBasicAuth |
| 14 | + |
| 15 | +from celery import Celery |
| 16 | +from celery.contrib.testing.worker import start_worker |
| 17 | + |
| 18 | + |
| 19 | +def get_rabbitmq_credentials(): |
| 20 | + """Get RabbitMQ credentials from environment.""" |
| 21 | + user = os.environ.get("RABBITMQ_DEFAULT_USER", "guest") |
| 22 | + password = os.environ.get("RABBITMQ_DEFAULT_PASSWORD", "guest") |
| 23 | + return user, password |
| 24 | + |
| 25 | + |
| 26 | +def get_rabbitmq_url(): |
| 27 | + """Get RabbitMQ broker URL from environment.""" |
| 28 | + user, password = get_rabbitmq_credentials() |
| 29 | + return os.environ.get( |
| 30 | + "TEST_BROKER", f"pyamqp://{user}:{password}@localhost:5672//") |
| 31 | + |
| 32 | + |
| 33 | +def get_management_api_url(): |
| 34 | + """Get RabbitMQ Management API base URL.""" |
| 35 | + return "http://localhost:15672/api" |
| 36 | + |
| 37 | + |
| 38 | +def get_bindings_for_exchange(exchange_name, vhost='/'): |
| 39 | + """Fetch bindings where the given exchange is the source. |
| 40 | +
|
| 41 | + Args: |
| 42 | + exchange_name: Name of the exchange |
| 43 | + vhost: Virtual host (default '/') |
| 44 | +
|
| 45 | + Returns: |
| 46 | + List of binding dictionaries |
| 47 | + """ |
| 48 | + user, password = get_rabbitmq_credentials() |
| 49 | + vhost_encoded = quote(vhost, safe='') |
| 50 | + exchange_encoded = quote(exchange_name, safe='') |
| 51 | + api_url = ( |
| 52 | + f"{get_management_api_url()}/exchanges/{vhost_encoded}/" |
| 53 | + f"{exchange_encoded}/bindings/source" |
| 54 | + ) |
| 55 | + response = requests.get(api_url, auth=HTTPBasicAuth(user, password)) |
| 56 | + response.raise_for_status() |
| 57 | + return response.json() |
| 58 | + |
| 59 | + |
| 60 | +def get_bindings_for_queue(queue_name, vhost='/'): |
| 61 | + """Fetch bindings for a specific queue. |
| 62 | +
|
| 63 | + Args: |
| 64 | + queue_name: Name of the queue |
| 65 | + vhost: Virtual host (default '/') |
| 66 | +
|
| 67 | + Returns: |
| 68 | + List of binding dictionaries |
| 69 | + """ |
| 70 | + user, password = get_rabbitmq_credentials() |
| 71 | + vhost_encoded = quote(vhost, safe='') |
| 72 | + queue_encoded = quote(queue_name, safe='') |
| 73 | + api_url = ( |
| 74 | + f"{get_management_api_url()}/queues/{vhost_encoded}/{queue_encoded}/" |
| 75 | + "bindings" |
| 76 | + ) |
| 77 | + response = requests.get(api_url, auth=HTTPBasicAuth(user, password)) |
| 78 | + response.raise_for_status() |
| 79 | + return response.json() |
| 80 | + |
| 81 | + |
| 82 | +def create_test_app(unique_id): |
| 83 | + """Create Celery app configured for native delayed delivery testing. |
| 84 | +
|
| 85 | + Args: |
| 86 | + unique_id: Unique identifier to ensure queue/exchange names don't |
| 87 | + conflict |
| 88 | +
|
| 89 | + Returns: |
| 90 | + Tuple of (app, exchange_name, queue_a_name, queue_b_name) |
| 91 | + """ |
| 92 | + broker_url = get_rabbitmq_url() |
| 93 | + |
| 94 | + # Get Redis backend URL from environment |
| 95 | + redis_host = os.environ.get("REDIS_HOST", "localhost") |
| 96 | + redis_port = os.environ.get("REDIS_PORT", "6379") |
| 97 | + backend_url = os.environ.get( |
| 98 | + "TEST_BACKEND", f"redis://{redis_host}:{redis_port}/0") |
| 99 | + |
| 100 | + app = Celery( |
| 101 | + "test_native_delayed_delivery_binding", |
| 102 | + broker=broker_url, |
| 103 | + backend=backend_url, |
| 104 | + ) |
| 105 | + |
| 106 | + # Configure topic exchange with unique name |
| 107 | + exchange_name = f'celery.topic_{unique_id}' |
| 108 | + default_exchange = Exchange(exchange_name, type='topic') |
| 109 | + |
| 110 | + # Define task queues with queue-a first, queue-b second |
| 111 | + queue_a_name = f'queue-a_{unique_id}' |
| 112 | + queue_b_name = f'queue-b_{unique_id}' |
| 113 | + app.conf.task_queues = [ |
| 114 | + Queue(queue_a_name, exchange=default_exchange, |
| 115 | + routing_key=queue_a_name, |
| 116 | + queue_arguments={'x-queue-type': 'quorum'}), |
| 117 | + Queue(queue_b_name, exchange=default_exchange, |
| 118 | + routing_key=queue_b_name, |
| 119 | + queue_arguments={'x-queue-type': 'quorum'}), |
| 120 | + ] |
| 121 | + |
| 122 | + # Recommended setting for using celery with quorum queues |
| 123 | + app.conf.broker_transport_options = {"confirm_publish": True} |
| 124 | + |
| 125 | + # Enable quorum queue detection to disable global QoS |
| 126 | + app.conf.worker_detect_quorum_queues = True |
| 127 | + |
| 128 | + return app, exchange_name, queue_a_name, queue_b_name |
| 129 | + |
| 130 | + |
| 131 | +@pytest.mark.amqp |
| 132 | +@pytest.mark.timeout(90) |
| 133 | +def test_worker_binds_consumed_queue_despite_earlier_queue_failure(): |
| 134 | + """Test that queue binding continues even when earlier queues fail to bind. |
| 135 | +
|
| 136 | + This test reproduces the scenario from |
| 137 | + https://github.com/celery/celery/issues/9960 |
| 138 | + """ |
| 139 | + unique_id = uuid.uuid4().hex |
| 140 | + app, exchange_name, queue_a_name, queue_b_name = create_test_app(unique_id) |
| 141 | + |
| 142 | + # Set default queue to queue-b so the start_worker ping task is received |
| 143 | + # by our worker |
| 144 | + app.conf.task_default_queue = queue_b_name |
| 145 | + |
| 146 | + # Start worker that only consumes from queue-b |
| 147 | + # queue-a is NOT consumed, so it won't be declared by this worker |
| 148 | + with start_worker( |
| 149 | + app, |
| 150 | + queues=[queue_b_name], |
| 151 | + loglevel="INFO", |
| 152 | + perform_ping_check=True, |
| 153 | + shutdown_timeout=15, |
| 154 | + ): |
| 155 | + # Check celery_delayed_delivery → exchange bindings |
| 156 | + delayed_delivery_bindings = \ |
| 157 | + get_bindings_for_exchange('celery_delayed_delivery') |
| 158 | + queue_b_delayed_binding = [ |
| 159 | + b for b in delayed_delivery_bindings |
| 160 | + if b.get('destination') == exchange_name |
| 161 | + and b.get('routing_key') == f'#.{queue_b_name}' |
| 162 | + ] |
| 163 | + assert len(queue_b_delayed_binding) >= 1, ( |
| 164 | + f"Expected delayed delivery binding for {queue_b_name!r}, but " |
| 165 | + f"got bindings: {delayed_delivery_bindings!r}" |
| 166 | + ) |
| 167 | + |
| 168 | + # Check celery.topic → queue-b bindings |
| 169 | + # Should have bindings from the topic exchange to queue-b for both |
| 170 | + # immediate and delayed delivery |
| 171 | + queue_b_bindings = get_bindings_for_queue(queue_b_name) |
| 172 | + topic_to_queue_bindings = [ |
| 173 | + b for b in queue_b_bindings |
| 174 | + if b.get('source') == exchange_name |
| 175 | + ] |
| 176 | + topic_to_queue_routing_keys = { |
| 177 | + b.get('routing_key') for b in topic_to_queue_bindings |
| 178 | + } |
| 179 | + |
| 180 | + # Check the routing key for immediate delivery |
| 181 | + assert queue_b_name in topic_to_queue_routing_keys, ( |
| 182 | + f"Expected routing key {queue_b_name!r} in bindings, but got: " |
| 183 | + f"{topic_to_queue_bindings!r}" |
| 184 | + ) |
| 185 | + |
| 186 | + # Check the routing key for delayed delivery |
| 187 | + assert f"#.{queue_b_name}" in topic_to_queue_routing_keys, ( |
| 188 | + f"Expected at least one binding from {exchange_name!r} to " |
| 189 | + f"{queue_b_name!r}, but got: {topic_to_queue_bindings!r}" |
| 190 | + ) |
0 commit comments