Repository: celery/celery
File: t/unit/utils/test_functional.py (Python)

Code before:
        g = regen(build_generator())
        assert bool(g)
        assert g[0] == 1

    def test_nonzero__empty_iter(self):
        assert not regen(iter([]))

    def test_deque(self):
        original_list = [42]
        d = collections.deque(original_list)
        # Confirm that concretising a `regen()` instance repeatedly for an
        # equality check always returns the original list
        g = regen(d)
        assert g == original_list
        assert g == original_list

    def test_repr(self):
        def die():
            raise AssertionError("Generator died")
            yield None

        # Confirm that `regen()` instances are not concretised when represented
        g = regen(die())
        assert "..." in repr(g)

    @pytest.mark.xfail(raises=StopIteration, reason="XXX NEEDS AN ISSUE")
    def test_partial_reconcretisation(self):
        class WeirdIterator():
            def __init__(self, iter_):
                self.iter_ = iter_
                self._errored = False

            def __iter__(self):
                yield from self.iter_
                if not self._errored:
                    try:
                        # This should stop the regen instance from marking
                        # itself as being done
                        #raise AssertionError("Iterator errored")
                        pass
                    finally:
                        self._errored = True

        original_list = list(range(42))
        g = regen(WeirdIterator(original_list))
        iter_g = iter(g)
        for e in original_list:
            assert e == next(iter_g)
        with pytest.raises(AssertionError):
            next(iter_g)
        assert getattr(g, "_regen__done") is False

Review comment (bug):
This test is for an edge case where a repeatable iterator could raise an exception at some point before completion which would result in `regen()` stashing the elements it saw but never marking itself as done. Then the next time we yield elements from an iterator derived from the `regen()` instance, we'd yield the stashed elements, then repeat them from the underlying iterator if it's capable of giving them to us again.

IDK if this is something we can even bother handling. If the underlying iterator explodes, we could choose to mark the regen instance as `__done` and just give up on attempting to get more elements? If we want to go down that route, I can roll a change into this PR since it'd be pretty simple and still thematically relevant. If we think we need to do something more involved or optimistic, I'll make a follow up issue for discussion.

Code after:
        g = regen(build_generator())
        assert bool(g)
        assert g[0] == 1

    def test_nonzero__empty_iter(self):
        assert not regen(iter([]))

    def test_deque(self):
        original_list = [42]
        d = collections.deque(original_list)
        # Confirm that concretising a `regen()` instance repeatedly for an
        # equality check always returns the original list
        g = regen(d)
        assert g == original_list
        assert g == original_list

    def test_repr(self):
        def die():
            raise AssertionError("Generator died")
            yield None

        # Confirm that `regen()` instances are not concretised when represented
        g = regen(die())
        assert "..." in repr(g)

    def test_partial_reconcretisation(self):
        class WeirdIterator():
            def __init__(self, iter_):
                self.iter_ = iter_
                self._errored = False

            def __iter__(self):
                yield from self.iter_
                if not self._errored:
                    try:
                        # This should stop the regen instance from marking
                        # itself as being done
                        raise AssertionError("Iterator errored")
                    finally:
                        self._errored = True

        original_list = list(range(42))
        g = regen(WeirdIterator(original_list))
        iter_g = iter(g)
        for e in original_list:
            assert e == next(iter_g)
        with pytest.raises(AssertionError, match="Iterator errored"):
            next(iter_g)
        # The following checks are for the known "misbehaviour"
        assert getattr(g, "_regen__done") is False
        # If the `regen()` instance doesn't think it's done then it'll dupe the