Method test_revoked_expires_expired

t/unit/worker/test_request.py:669–679

(self)

        assert job._already_cancelled is True

    def test_revoked_expires_expired(self):
        job = self.get_request(self.mytask.s(1, f='x').set(
            expires=datetime.now(timezone.utc) - timedelta(days=1)
        ))
        with self.assert_signal_called(
                task_revoked, sender=job.task, request=job._context,
                terminated=False, expired=True, signum=None):
            job.revoked()
            assert job.id in revoked
            self.app.set_current()
            assert self.mytask.backend.get_status(job.id) == states.REVOKED

    def test_revoked_expires_not_expired(self):
        job = self.xRequest(