(self)
| 1030 | pytest.raises(Exception, task.execute) |
| 1031 | |
| 1032 | def test_tasks_from_queue(self): |
| 1033 | task0 = Task(self.tiger, simple_task) |
| 1034 | task1 = Task(self.tiger, exception_task) |
| 1035 | task2 = Task(self.tiger, simple_task, queue="other") |
| 1036 | |
| 1037 | task0.delay() |
| 1038 | task1.delay() |
| 1039 | task2.delay() |
| 1040 | |
| 1041 | n, tasks = Task.tasks_from_queue(self.tiger, "default", "queued") |
| 1042 | assert n == 2 |
| 1043 | assert task0.id == tasks[0].id |
| 1044 | assert task0.func == simple_task |
| 1045 | assert task0.func == tasks[0].func |
| 1046 | assert task0.serialized_func == "tests.tasks:simple_task" |
| 1047 | assert task0.serialized_func == tasks[0].serialized_func |
| 1048 | assert task0.state == tasks[0].state |
| 1049 | assert task0.state == "queued" |
| 1050 | assert task0.queue == tasks[0].queue |
| 1051 | assert task0.queue == "default" |
| 1052 | |
| 1053 | def test_tasks_from_queue_with_executions(self): |
| 1054 | self.tiger.delay(exception_task, retry=True) |
nothing calls this directly
no test coverage detected