(self)
| 1119 | assert self.conn.get("task_id") == task.id |
| 1120 | |
| 1121 | def test_current_tasks(self): |
| 1122 | task1 = Task(self.tiger, verify_current_tasks) |
| 1123 | task1.delay() |
| 1124 | task2 = Task(self.tiger, verify_current_tasks) |
| 1125 | task2.delay() |
| 1126 | Worker(self.tiger).run(once=True) |
| 1127 | assert self.conn.lrange("task_ids", 0, -1) == [task1.id, task2.id] |
| 1128 | |
| 1129 | def test_current_task_eager(self): |
| 1130 | self.tiger.config["ALWAYS_EAGER"] = True |