Repository: kedro-org/kedro
File: tests/runner/test_parallel_runner.py (Python)

Code before:
        assert result["Z"] == (42, 42, 42)

    @pytest.mark.parametrize("is_async", [False, True])
    def test_parallel_run_with_plugin_manager(self, is_async, fan_out_fan_in, catalog):
        catalog.add_feed_dict({"A": 42})
        result = ParallelRunner(is_async=is_async).run(
            fan_out_fan_in, catalog, hook_manager=_create_hook_manager()
        )
        assert "Z" in result
        assert len(result["Z"]) == 3
        assert result["Z"] == (42, 42, 42)

    @pytest.mark.parametrize("is_async", [False, True])
    def test_memory_dataset_input(self, is_async, fan_out_fan_in):
        pipeline = modular_pipeline([fan_out_fan_in])
        catalog = DataCatalog({"A": MemoryDataset("42")})
        result = ParallelRunner(is_async=is_async).run(pipeline, catalog)
        assert "Z" in result
        assert len(result["Z"]) == 3
        assert result["Z"] == ("42", "42", "42")

    def test_log_not_using_async(self, fan_out_fan_in, catalog, caplog):
        catalog.add_feed_dict({"A": 42})
        ParallelRunner().run(fan_out_fan_in, catalog)
        assert (
            "Asynchronous mode is disabled for loading and saving data." in caplog.text
        )


class TestMaxWorkers:
    @pytest.mark.parametrize("is_async", [False, True])
    @pytest.mark.parametrize(
        "cpu_cores, user_specified_number, expected_number",
        [
            # The pipeline only needs 3 processes, no need for more
            (4, 6, 3),
            (4, None, 3),
            # We need 3 processes, but only 2 CPU cores available
            (2, None, 2),
            # Even though we have 1 CPU core, allow user to use more.
            (1, 2, 2),
        ],
    )
    def test_specified_max_workers_bellow_cpu_cores_count(
        self,
        is_async,
        mocker,
        fan_out_fan_in,
        catalog,
        cpu_cores,
        user_specified_number,

Review comment (suggestion):
```suggestion
            "Using synchronous mode for loading and saving data." in caplog.text
```

Code after:
        assert result["Z"] == (42, 42, 42)

    @pytest.mark.parametrize("is_async", [False, True])
    def test_parallel_run_with_plugin_manager(self, is_async, fan_out_fan_in, catalog):
        catalog.add_feed_dict({"A": 42})
        result = ParallelRunner(is_async=is_async).run(
            fan_out_fan_in, catalog, hook_manager=_create_hook_manager()
        )
        assert "Z" in result
        assert len(result["Z"]) == 3
        assert result["Z"] == (42, 42, 42)

    @pytest.mark.parametrize("is_async", [False, True])
    def test_memory_dataset_input(self, is_async, fan_out_fan_in):
        pipeline = modular_pipeline([fan_out_fan_in])
        catalog = DataCatalog({"A": MemoryDataset("42")})
        result = ParallelRunner(is_async=is_async).run(pipeline, catalog)
        assert "Z" in result
        assert len(result["Z"]) == 3
        assert result["Z"] == ("42", "42", "42")

    def test_log_not_using_async(self, fan_out_fan_in, catalog, caplog):
        catalog.add_feed_dict({"A": 42})
        ParallelRunner().run(fan_out_fan_in, catalog)
        assert "Using synchronous mode for loading and saving data." in caplog.text


class TestMaxWorkers:
    @pytest.mark.parametrize("is_async", [False, True])
    @pytest.mark.parametrize(
        "cpu_cores, user_specified_number, expected_number",
        [
            # The pipeline only needs 3 processes, no need for more
            (4, 6, 3),
            (4, None, 3),
            # We need 3 processes, but only 2 CPU cores available
            (2, None, 2),
            # Even though we have 1 CPU core, allow user to use more.
            (1, 2, 2),
        ],
    )
    def test_specified_max_workers_bellow_cpu_cores_count(
        self,
        is_async,
        mocker,
        fan_out_fan_in,
        catalog,
        cpu_cores,
        user_specified_number,
        expected_number,
    ):  # noqa: PLR0913