(self)
| 378 | isinstance(chunks_subclass.from_dict(d).clone(), chunks_subclass) |
| 379 | |
| 380 | def test_chunks(self): |
| 381 | x = self.add.chunks(range(100), 10) |
| 382 | assert dict(chunks.from_dict(dict(x), app=self.app)) == dict(x) |
| 383 | |
| 384 | assert x.group() |
| 385 | assert len(x.group().tasks) == 10 |
| 386 | |
| 387 | x.group = Mock() |
| 388 | gr = x.group.return_value = Mock() |
| 389 | |
| 390 | x.apply_async() |
| 391 | gr.apply_async.assert_called_with((), {}, route_name=self.add.name) |
| 392 | gr.apply_async.reset_mock() |
| 393 | x() |
| 394 | gr.apply_async.assert_called_with((), {}, route_name=self.add.name) |
| 395 | |
| 396 | self.app.conf.task_always_eager = True |
| 397 | chunks.apply_chunks(app=self.app, **x['kwargs']) |
| 398 | |
| 399 | |
| 400 | class test_chain(CanvasCase): |
nothing calls this directly
no test coverage detected