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

Code before:
        job.acknowledge = Mock(name='ack')
        job.task.acks_late = True
        job.on_timeout(soft=False, timeout=1337)
        assert 'Hard time limit' in error.call_args[0][0]
        assert self.mytask.backend.get_status(job.id) == states.FAILURE
        job.acknowledge.assert_called_with()

        job = self.xRequest()
        job.acknowledge = Mock(name='ack')
        job.task.acks_late = False
        job.on_timeout(soft=False, timeout=1335)
        job.acknowledge.assert_not_called()

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

        job = self.xRequest()
        job.acknowledge = Mock(name='ack')
        job.task.acks_late = True
        job.task.acks_on_failure_or_timeout = True
        job.on_timeout(soft=False, timeout=1337)
        assert 'Hard time limit' in error.call_args[0][0]
        assert self.mytask.backend.get_status(job.id) == states.FAILURE
        job.acknowledge.assert_called_with()

        job = self.xRequest()
        job.acknowledge = Mock(name='ack')
        job.task.acks_late = True
        job.task.acks_on_failure_or_timeout = False
        job.on_timeout(soft=False, timeout=1337)
        assert 'Hard time limit' in error.call_args[0][0]
        assert self.mytask.backend.get_status(job.id) == states.FAILURE
        job.acknowledge.assert_not_called()

        job = self.xRequest()
        job.acknowledge = Mock(name='ack')
        job.task.acks_late = False
        job.task.acks_on_failure_or_timeout = True
        job.on_timeout(soft=False, timeout=1335)
        job.acknowledge.assert_not_called()

    def test_on_soft_timeout(self, patching):
        warn = patching('celery.worker.request.warn')

        job = self.xRequest()
        job.acknowledge = Mock(name='ack')
        job.task.acks_late = True
        job.on_timeout(soft=True, timeout=1337)
        assert 'Soft time limit' in warn.call_args[0][0]
        assert self.mytask.backend.get_status(job.id) == states.PENDING
        job.acknowledge.assert_not_called()

Review comment (suggestion):
```suggestion

    def test_from_message_invalid_kwargs(self):
```

Code after:
            job.on_failure(exc_info)

        job.on_failure(exc_info)
        assert not job.eventer.send.called

    def test_on_failure_should_terminate(self):
        from celery.worker import state
        original_should_terminate = state.should_terminate
        state.should_terminate = True
        job = self.xRequest()
        job.send_event = Mock(name='send_event')
        job.task.backend = Mock(name='backend')

        try:
            try:
                raise KeyError('foo')
            except KeyError:
                exc_info = ExceptionInfo()
                job.on_failure(exc_info)

            job.send_event.assert_not_called()
            job.task.backend.mark_as_failure.assert_not_called()
        finally:
            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')

        job = self.xRequest()
        job.acknowledge = Mock(name='ack')
        job.task.acks_late = True
        job.on_timeout(soft=False, timeout=1337)
        assert 'Hard time limit' in error.call_args[0][0]
        assert self.mytask.backend.get_status(job.id) == states.FAILURE
        job.acknowledge.assert_called_with()

        job = self.xRequest()
        job.acknowledge = Mock(name='ack')
        job.task.acks_late = False
        job.on_timeout(soft=False, timeout=1335)
        job.acknowledge.assert_not_called()

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