Method test_revoke_task

huey/tests/test_api.py:202–246

(self)

        self.assertEqual(self.huey.execute(task, t20), 6)

    def test_revoke_task(self):
        state = {}
        @self.huey.task()
        def task_a(n):
            state[n] = n  # Modify some state for extra visibility.
            return n + 1

        self.assertFalse(task_a.is_revoked())
        task_a.revoke()
        self.assertTrue(task_a.is_revoked())

        r1, r2, r3 = task_a(1), task_a(2), task_a(3)
        self.assertEqual(len(self.huey), 3)

        # The result-wrapper will indicate the tasks are revoked.
        for r in (r1, r2, r3):
            self.assertTrue(r.is_revoked())
            self.assertTrue(self.huey.is_revoked(r))
            # We don't have the task class when using an ID so this doesn't
            # work.
            self.assertFalse(self.huey.is_revoked(r.task.id))

        # Task is discarded and not executed, due to being revoked.
        t1 = self.huey.dequeue()
        self.assertTrue(self.huey.execute(t1) is None)
        self.assertTrue(r1.get() is None)
        self.assertEqual(state, {})

        # Next task is also discarded.
        self.assertTrue(r2.is_revoked())
        self.assertTrue(self.execute_next() is None)
        self.assertTrue(r2.get() is None)
        self.assertEqual(state, {})

        # Restore task, we will see side-effects and results of execution.
        self.assertTrue(task_a.is_revoked())  # Still revoked.
        self.assertTrue(r3.is_revoked())
        self.assertTrue(task_a.restore())
        self.assertFalse(task_a.is_revoked())
        self.assertFalse(r3.is_revoked())

        t3 = self.huey.dequeue()
        self.assertEqual(self.huey.execute(t3), 4)
        self.assertEqual(r3.get(), 4)
        self.assertEqual(state, {3: 3})

    def test_revoke_task_instance(self):
        state = {}