| 98 | |
| 99 | @pytest.mark.skipif(sys.version_info < (3, 3), reason="__qualname__ unavailable") |
| 100 | def test_staticmethod_task(self): |
| 101 | self.tiger.delay(StaticTask.task) |
| 102 | queues = self._ensure_queues(queued={"default": 1}) |
| 103 | task = queues["queued"]["default"][0] |
| 104 | assert task["func"] == "tests.tasks:StaticTask.task" |
| 105 | |
| 106 | Worker(self.tiger).run(once=True) |
| 107 | self._ensure_queues(queued={"default": 0}) |
| 108 | assert not self.conn.exists("t:task:%s" % task["id"]) |
| 109 | |
| 110 | def test_task_delay(self): |
| 111 | decorated_task.delay(1, 2, a=3, b=4) |