Method test_create_task_handler

t/unit/worker/test_consumer.py:157–184

(self, subtests)

        assert c._restore_prefetch_count_after_connection_restart(None) is None

    def test_create_task_handler(self, subtests):
        c = self.get_consumer()
        c.qos = MagicMock()
        c.qos.value = 1
        c._maximum_prefetch_restored = False

        sig = self.add.s(2, 2)
        message = self.task_message_from_sig(self.app, sig)

        def raise_exception():
            raise KeyError('Foo')

        def strategy(_, __, ack_log_error_promise, ___, ____):
            ack_log_error_promise()

        c.strategies[sig.task] = strategy
        c.call_soon = raise_exception
        on_task_received = c.create_task_handler()
        on_task_received(message)

        with subtests.test("initial prefetch count is never 0"):
            assert c.initial_prefetch_count != 0

        with subtests.test("initial prefetch count is 2"):
            assert c.initial_prefetch_count == 2

        with subtests.test("maximum prefetch is reached"):
            assert c._maximum_prefetch_restored is True

    def test_flush_events(self):
        c = self.get_consumer()