(self, manager)
| 1780 | assert res.get(timeout=TIMEOUT) == [13, 14, 15, 16] |
| 1781 | |
| 1782 | def test_group_args_and_kwargs(self, manager): |
| 1783 | try: |
| 1784 | manager.app.backend.ensure_chords_allowed() |
| 1785 | except NotImplementedError as e: |
| 1786 | raise pytest.skip(e.args[0]) |
| 1787 | c = ( |
| 1788 | group(add.s(i) for i in range(4)) | |
| 1789 | add_to_all.s(8) |
| 1790 | ) |
| 1791 | res = c.apply_async(args=(4,), kwargs={"z": 1}) |
| 1792 | if manager.app.conf.result_backend.startswith('redis'): |
| 1793 | # for a simple chord like the one above, redis does not guarantee |
| 1794 | # the ordering of the results as a performance trade off. |
| 1795 | assert set(res.get(timeout=TIMEOUT)) == {13, 14, 15, 16} |
| 1796 | else: |
| 1797 | assert res.get(timeout=TIMEOUT) == [13, 14, 15, 16] |
| 1798 | |
| 1799 | def test_nested_group_chain(self, manager): |
| 1800 | try: |
nothing calls this directly
no test coverage detected