Method test_custom_runner_class_single_task

tests/test_base.py:1488–1493

(self)

class TestRunnerClass(BaseTestCase):
    def test_custom_runner_class_single_task(self):
        task = self.tiger.delay(simple_task, runner_class=MyRunnerClass)
        Worker(self.tiger).run(once=True)
        assert self.conn.get("task_id") == task.id
        self.conn.delete("task_id")
        self._ensure_queues()

    def test_custom_runner_class_batch_task(self):
        self.tiger.delay(batch_task, args=[1], runner_class=MyRunnerClass)