Method test_from_message_invalid_kwargs

t/unit/worker/test_request.py:984–988

(self)

            state.should_terminate = original_should_terminate

    def test_from_message_invalid_kwargs(self):
        m = self.TaskMessage(self.mytask.name, args=(), kwargs='foo')
        req = Request(m, app=self.app)
        with pytest.raises(InvalidTaskError):
            raise req.execute().exception.exc

    def test_on_hard_timeout_acks_late(self, patching):
        error = patching('celery.worker.request.error')