Method test_complex_chain

t/integration/test_canvas.py:185–193

(self, manager)

    @flaky
    def test_complex_chain(self, manager):
        g = group(add.s(i) for i in range(4))
        c = (
            add.s(2, 2) | (
                add.s(4) | add_replaced.s(8) | add.s(16) | add.s(32)
            ) | g
        )
        res = c()
        assert res.get(timeout=TIMEOUT) == [64, 65, 66, 67]

    @pytest.mark.xfail(raises=TimeoutError, reason=class="st">"Task is timeout")
    def test_group_results_in_chain(self, manager):