Repository: celery/celery
File: t/unit/worker/test_native_delayed_delivery.py (Python)

Code before:
        consumer_mock.app.conf.broker_native_delayed_delivery_queue_type = 'classic'
        consumer_mock.app.conf.broker_url = 'amqp://'
        consumer_mock.app.amqp.queues = {
            'celery': Queue('celery', exchange=Exchange('celery', type='fanout'))
        }

        delayed_delivery = DelayedDelivery(consumer_mock)

        delayed_delivery.start(consumer_mock)

        assert len(caplog.records) == 0

    def test_validate_broker_urls_empty(self):
        delayed_delivery = DelayedDelivery(Mock())

        with pytest.raises(ValueError, match="broker_url configuration is empty"):
            delayed_delivery._validate_broker_urls("")

        with pytest.raises(ValueError, match="broker_url configuration is empty"):
            delayed_delivery._validate_broker_urls(None)

    def test_validate_broker_urls_invalid(self):
        delayed_delivery = DelayedDelivery(Mock())

        with pytest.raises(ValueError, match="No valid broker URLs found in configuration"):
            delayed_delivery._validate_broker_urls("  ;  ;  ")

    def test_validate_broker_urls_valid(self):
        delayed_delivery = DelayedDelivery(Mock())

        urls = delayed_delivery._validate_broker_urls("amqp://localhost;amqp://remote")
        assert urls == {"amqp://localhost", "amqp://remote"}

    def test_validate_queue_type_empty(self):
        delayed_delivery = DelayedDelivery(Mock())

        with pytest.raises(ValueError, match="broker_native_delayed_delivery_queue_type is not configured"):
            delayed_delivery._validate_queue_type(None)

        with pytest.raises(ValueError, match="broker_native_delayed_delivery_queue_type is not configured"):
            delayed_delivery._validate_queue_type("")

    def test_validate_queue_type_invalid(self):
        delayed_delivery = DelayedDelivery(Mock())

        with pytest.raises(ValueError, match="Invalid queue type 'invalid'. Must be one of: classic, quorum"):
            delayed_delivery._validate_queue_type("invalid")

    def test_validate_queue_type_valid(self):
        delayed_delivery = DelayedDelivery(Mock())


Review comment (bug):
I think this test case for multiple empty broker URLs would begin failing with the hotfixed code.  Could we retain it in the parametrized tests?

Code after:
        consumer_mock.app.conf.broker_native_delayed_delivery_queue_type = 'classic'
        consumer_mock.app.conf.broker_url = 'amqp://'
        consumer_mock.app.amqp.queues = {
            'celery': Queue('celery', exchange=Exchange('celery', type='fanout'))
        }

        delayed_delivery = DelayedDelivery(consumer_mock)

        delayed_delivery.start(consumer_mock)

        assert len(caplog.records) == 0

    @pytest.mark.parametrize(
        "broker_urls, expected_result",
        [
            ("amqp://", {"amqp://"}),
            ("amqp://;redis://", {"amqp://", "redis://"}),
            (
                ["amqp://", "redis://", "sqs://"],
                {"amqp://", "redis://", "sqs://"},
            ),
        ],
    )
    def test_validate_broker_urls_valid(self, broker_urls, expected_result):
        delayed_delivery = DelayedDelivery(Mock())
        urls = delayed_delivery._validate_broker_urls(broker_urls)
        assert urls == expected_result

    @pytest.mark.parametrize(
        "broker_urls, exception_type, exception_match",
        [
            ("", ValueError, "broker_url configuration is empty"),
            (None, ValueError, "broker_url configuration is empty"),
            ([], ValueError, "broker_url configuration is empty"),
            (123, ValueError, "broker_url must be a string or list"),
            (["amqp://", 123, None, "amqp://"], ValueError, "All broker URLs must be strings"),
        ],
    )
    def test_validate_broker_urls_invalid(self, broker_urls, exception_type, exception_match):
        delayed_delivery = DelayedDelivery(Mock())
        with pytest.raises(exception_type, match=exception_match):
            delayed_delivery._validate_broker_urls(broker_urls)

    def test_validate_queue_type_empty(self):
        delayed_delivery = DelayedDelivery(Mock())

        with pytest.raises(ValueError, match="broker_native_delayed_delivery_queue_type is not configured"):
            delayed_delivery._validate_queue_type(None)

        with pytest.raises(ValueError, match="broker_native_delayed_delivery_queue_type is not configured"):
            delayed_delivery._validate_queue_type("")