| 87 | assert task["kwargs"] == {"a": 3, "b": 4} |
| 88 | |
| 89 | def test_simple_task(self): |
| 90 | self.tiger.delay(simple_task) |
| 91 | queues = self._ensure_queues(queued={"default": 1}) |
| 92 | task = queues["queued"]["default"][0] |
| 93 | assert task["func"] == "tests.tasks:simple_task" |
| 94 | |
| 95 | Worker(self.tiger).run(once=True) |
| 96 | self._ensure_queues(queued={"default": 0}) |
| 97 | assert not self.conn.exists("t:task:%s" % task["id"]) |
| 98 | |
| 99 | @pytest.mark.skipif(sys.version_info < (3, 3), reason="__qualname__ unavailable") |
| 100 | def test_staticmethod_task(self): |