Function create_test_app

t/integration/test_native_delayed_delivery_binding.py:82–128

Create Celery app configured for native delayed delivery testing. Args: unique_id: Unique identifier to ensure queue/exchange names don't conflict Returns: Tuple of (app, exchange_name, queue_a_name, queue_b_name)

(unique_id)

def create_test_app(unique_id):
    """Create Celery app configured for native delayed delivery testing.

    Args:
        unique_id: Unique identifier to ensure queue/exchange names don't
                   conflict

    Returns:
        Tuple of (app, exchange_name, queue_a_name, queue_b_name)
    """
    broker_url = get_rabbitmq_url()

    # Get Redis backend URL from environment
    redis_host = os.environ.get("REDIS_HOST", "localhost")
    redis_port = os.environ.get("REDIS_PORT", "6379")
    backend_url = os.environ.get(
        "TEST_BACKEND", f"redis://{redis_host}:{redis_port}/0")

    app = Celery(
        "test_native_delayed_delivery_binding",
        broker=broker_url,
        backend=backend_url,
    )

    # Configure topic exchange with unique name
    exchange_name = f'celery.topic_{unique_id}'
    default_exchange = Exchange(exchange_name, type='topic')

    # Define task queues with queue-a first, queue-b second
    queue_a_name = f'queue-a_{unique_id}'
    queue_b_name = f'queue-b_{unique_id}'
    app.conf.task_queues = [
        Queue(queue_a_name, exchange=default_exchange,
              routing_key=queue_a_name,
              queue_arguments={'x-queue-type': 'quorum'}),
        Queue(queue_b_name, exchange=default_exchange,
              routing_key=queue_b_name,
              queue_arguments={'x-queue-type': 'quorum'}),
    ]

    # Recommended setting for using celery with quorum queues
    app.conf.broker_transport_options = {"confirm_publish": True}

    # Enable quorum queue detection to disable global QoS
    app.conf.worker_detect_quorum_queues = True

    return app, exchange_name, queue_a_name, queue_b_name


@pytest.mark.amqp