Repository: celery/celery
File: t/integration/test_canvas.py (Python)

Code before:
        with subtests.test(msg="Errback is called after body group fails"):
            await_redis_echo({errback_msg, }, redis_key=redis_key)
        redis_connection.delete(redis_key)

    @pytest.mark.parametrize(
        "errback_task", [errback_old_style, errback_new_style, ],
    )
    def test_mutable_errback_called_by_chord_from_group(
        self, errback_task, manager, subtests
    ):
        if not manager.app.conf.result_backend.startswith("redis"):
            raise pytest.skip("Requires redis result backend.")
        redis_connection = get_redis_connection()

        errback = errback_task.s()
        fail_sig = fail.s()
        fail_sig_id = fail_sig.freeze().id
        child_sig = group(identity.si(42), fail_sig)

        chord_sig = chord((child_sig, ), identity.s())
        chord_sig.link_error(errback)
        expected_redis_key = chord_sig.body.freeze().id
        redis_connection.delete(expected_redis_key)
        with subtests.test(msg="Error propagates from header group"):
            res = chord_sig.delay()
            with pytest.raises(ExpectedException):
                res.get(timeout=TIMEOUT)
        with subtests.test(msg="Errback is called after header group fails"):
            await_redis_count(1, redis_key=expected_redis_key)

        chord_sig = chord((identity.si(42), ), child_sig)
        chord_sig.link_error(errback)
        expected_redis_key = fail_sig_id
        redis_connection.delete(expected_redis_key)
        with subtests.test(msg="Error propagates from body group"):
            res = chord_sig.delay()
            with pytest.raises(ExpectedException):
                res.get(timeout=TIMEOUT)
        with subtests.test(msg="Errback is called after body group fails"):
            await_redis_count(1, redis_key=expected_redis_key)
        redis_connection.delete(expected_redis_key)

    def test_immutable_errback_called_by_chord_from_group_fail_multiple(
        self, manager, subtests
    ):
        if not manager.app.conf.result_backend.startswith("redis"):
            raise pytest.skip("Requires redis result backend.")
        redis_connection = get_redis_connection()

        fail_task_count = 42
        redis_key = str(uuid.uuid4())

Review comment (suggestion):
This and the `res.get()` in the subtest below seem to sometimes time out and I can't work out why. It seems to only really occur when I run the full suite (redis backend) or run `tox -e ... -- -k test_canvas` to select the canvas integration tests.

I can't mark them as flaky since subtests don't play nicely with that decorator :(

Code after:
            await_redis_echo({errback_msg, }, redis_key=redis_key)
        redis_connection.delete(redis_key)

    @pytest.mark.parametrize(
        "errback_task", [errback_old_style, errback_new_style, ],
    )
    def test_mutable_errback_called_by_chord_from_group(
        self, errback_task, manager, subtests
    ):
        if not manager.app.conf.result_backend.startswith("redis"):
            raise pytest.skip("Requires redis result backend.")
        redis_connection = get_redis_connection()

        errback = errback_task.s()
        fail_sig = fail.s()
        fail_sig_id = fail_sig.freeze().id
        child_sig = group(identity.si(42), fail_sig)

        chord_sig = chord((child_sig, ), identity.s())
        chord_sig.link_error(errback)
        expected_redis_key = chord_sig.body.freeze().id
        redis_connection.delete(expected_redis_key)
        with subtests.test(msg="Error propagates from header group"):
            res = chord_sig.delay()
            with pytest.raises(ExpectedException):
                res.get(timeout=TIMEOUT)
        with subtests.test(msg="Errback is called after header group fails"):
            await_redis_count(1, redis_key=expected_redis_key)

        chord_sig = chord((identity.si(42), ), child_sig)
        chord_sig.link_error(errback)
        expected_redis_key = fail_sig_id
        redis_connection.delete(expected_redis_key)
        with subtests.test(msg="Error propagates from body group"):
            res = chord_sig.delay()
            with pytest.raises(ExpectedException):
                res.get(timeout=TIMEOUT)
        with subtests.test(msg="Errback is called after body group fails"):
            await_redis_count(1, redis_key=expected_redis_key)
        redis_connection.delete(expected_redis_key)

    def test_immutable_errback_called_by_chord_from_group_fail_multiple(
        self, manager, subtests
    ):
        if not manager.app.conf.result_backend.startswith("redis"):
            raise pytest.skip("Requires redis result backend.")
        redis_connection = get_redis_connection()

        fail_task_count = 42
        redis_key = str(uuid.uuid4())
        errback = redis_count.si(redis_key=redis_key)