(self)
| 1234 | job._on_ack.assert_not_called() |
| 1235 | |
| 1236 | def test_execute_success_no_kwargs(self): |
| 1237 | |
| 1238 | @self.app.task # traverses coverage for decorator without parens |
| 1239 | def mytask_no_kwargs(i): |
| 1240 | return i ** i |
| 1241 | |
| 1242 | tid = uuid() |
| 1243 | job = self.xRequest( |
| 1244 | name=mytask_no_kwargs.name, |
| 1245 | id=tid, |
| 1246 | args=[4], |
| 1247 | kwargs={}, |
| 1248 | ) |
| 1249 | assert job.execute() == 256 |
| 1250 | meta = mytask_no_kwargs.backend.get_task_meta(tid) |
| 1251 | assert meta['result'] == 256 |
| 1252 | assert meta['status'] == states.SUCCESS |
| 1253 | |
| 1254 | def test_execute_ack(self): |
| 1255 | scratch = {'ACK': False} |
nothing calls this directly
no test coverage detected