Method test_iterdeps

t/unit/tasks/test_result.py:196–215

(self)

        ]

    def test_iterdeps(self):
        x = self.app.AsyncResult('1')
        c = [EagerResult(str(i), i, states.SUCCESS) for i in range(3)]
        x._cache = {'status': states.SUCCESS, 'result': None, 'children': c}
        for child in c:
            child.backend = Mock()
            child.backend.get_children.return_value = []
        it = x.iterdeps()
        assert list(it) == [
            (None, x),
            (x, c[0]),
            (x, c[1]),
            (x, c[2]),
        ]
        x._cache = None
        x.ready = Mock()
        x.ready.return_value = False
        with pytest.raises(IncompleteStream):
            list(x.iterdeps())
        list(x.iterdeps(intermediate=True))

    def test_eq_not_implemented(self):
        assert self.app.AsyncResult('1') != object()