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

Code before:
            ]),
            group([
                redis_echo.si('3', redis_key=redis_key),
                redis_echo.si('4', redis_key=redis_key),
                redis_echo.si('5', redis_key=redis_key)
            ]),
            group([
                redis_echo.si('6', redis_key=redis_key),
                redis_echo.si('7', redis_key=redis_key),
                redis_echo.si('8', redis_key=redis_key),
                redis_echo.si('9', redis_key=redis_key)
            ]),
            redis_echo.si('Done', redis_key='Done'),
        )

        with subtests.test(msg='Run the chain and wait for completion'):
            redis_connection.delete(redis_key, 'Done')
            c.delay().get(timeout=TIMEOUT)
            await_redis_list_message_length(1, redis_key='Done', timeout=10)

        with subtests.test(msg='All tasks are executed once'):
            actual = [sig.decode('utf-8') for sig in redis_connection.lrange(redis_key, 0, -1)]
            expected = [str(i) for i in range(1, 10)]
            with subtests.test(msg='All tasks are executed once'):
                assert sorted(actual) == sorted(expected)

        # Cleanup
        redis_connection.delete(redis_key, 'Done')

    def test_chaining_upgraded_chords_mixed_canvas_protocol_3(self, manager, subtests):
        """ This test is built to reproduce the github issue https://github.com/celery/celery/issues/8662

        The issue describes a canvas where a chain of groups are executed multiple times instead of once.
        This test is built to reproduce the issue and to verify that the issue is fixed.
        """
        try:
            manager.app.backend.ensure_chords_allowed()
        except NotImplementedError as e:
            raise pytest.skip(e.args[0])

        if not manager.app.conf.result_backend.startswith('redis'):
            raise pytest.skip('Requires redis result backend.')

        redis_connection = get_redis_connection()
        redis_key = 'echo_chamber'

        c = chain(
            group([
                redis_echo.si('1', redis_key=redis_key),
                redis_echo.si('2', redis_key=redis_key)
            ]),

Review comment (refactor):
I had to literally compare `test_chaining_upgraded_chords_mixed_canvas_protocol_2` and `test_chaining_upgraded_chords_mixed_canvas_protocol_3` to find that single difference as it is so small haha.

You can use a single test case with `@pytest.mark.parameterize` like this:
```python


    @pytest.mark.parametrize(
        "last_task",
        [
            redis_echo.si('Done', redis_key='Done'),
            group(redis_echo.si('Done', redis_key='Done')),
        ],
    )
    def test_chaining_upgraded_chords_mixed_canvas_protocol_123(self, manager, subtests, last_task):
        ...
```

This will also reduce duplicated code which is generally better.

Code after:
            raise pytest.skip(e.args[0])

        if not manager.app.conf.result_backend.startswith('redis'):
            raise pytest.skip('Requires redis result backend.')

        redis_connection = get_redis_connection()
        redis_key = 'echo_chamber'

        c = chain(
            group([
                redis_echo.si('1', redis_key=redis_key),
                redis_echo.si('2', redis_key=redis_key)
            ]),
            group([
                redis_echo.si('3', redis_key=redis_key),
                redis_echo.si('4', redis_key=redis_key),
                redis_echo.si('5', redis_key=redis_key)
            ]),
            group([
                redis_echo.si('6', redis_key=redis_key),
                redis_echo.si('7', redis_key=redis_key),
                redis_echo.si('8', redis_key=redis_key),
                redis_echo.si('9', redis_key=redis_key)
            ]),
            redis_echo.si('Done', redis_key='Done') if not group_last_task else
            group(redis_echo.si('Done', redis_key='Done')),
        )

        with subtests.test(msg='Run the chain and wait for completion'):
            redis_connection.delete(redis_key, 'Done')
            c.delay().get(timeout=TIMEOUT)
            await_redis_list_message_length(1, redis_key='Done', timeout=10)

        with subtests.test(msg='All tasks are executed once'):
            actual = [
                sig.decode('utf-8')
                for sig in redis_connection.lrange(redis_key, 0, -1)
            ]
            expected = [str(i) for i in range(1, 10)]
            with subtests.test(msg='All tasks are executed once'):
                assert sorted(actual) == sorted(expected)

        # Cleanup
        redis_connection.delete(redis_key, 'Done')


class test_result_set:

    @flaky
    def test_result_set(self, manager):
        assert_ping(manager)