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

Code before:
class TestValidThreadRunner:
    def test_thread_run(self, fan_out_fan_in, catalog):
        catalog.add_feed_dict({"A": 42})
        result = ThreadRunner().run(fan_out_fan_in, catalog)
        assert "Z" in result
        assert result["Z"] == (42, 42, 42)

    def test_thread_run_with_plugin_manager(self, fan_out_fan_in, catalog):
        catalog.add_feed_dict({"A": 42})
        result = ThreadRunner().run(
            fan_out_fan_in, catalog, hook_manager=_create_hook_manager()
        )
        assert "Z" in result
        assert result["Z"] == (42, 42, 42)

    def test_memory_dataset_input(self, fan_out_fan_in):
        catalog = DataCatalog({"A": MemoryDataset("42")})
        result = ThreadRunner().run(fan_out_fan_in, catalog)
        assert "Z" in result
        assert result["Z"] == ("42", "42", "42")

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


class TestMaxWorkers:
    @pytest.mark.parametrize(
        "user_specified_number, expected_number",
        [
            # The pipeline only needs 3 threads, no need for more
            (6, 3),
            (None, 3),
        ],
    )
    def test_specified_max_workers(
        self,
        mocker,
        fan_out_fan_in,
        catalog,
        user_specified_number,
        expected_number,
    ):  # noqa: PLR0913
        """
        We initialize the runner with max_workers=4.
        `fan_out_fan_in` pipeline needs 3 threads.
        A pool with 3 workers should be used.

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

Code after:
class TestValidThreadRunner:
    def test_thread_run(self, fan_out_fan_in, catalog):
        catalog.add_feed_dict({"A": 42})
        result = ThreadRunner().run(fan_out_fan_in, catalog)
        assert "Z" in result
        assert result["Z"] == (42, 42, 42)

    def test_thread_run_with_plugin_manager(self, fan_out_fan_in, catalog):
        catalog.add_feed_dict({"A": 42})
        result = ThreadRunner().run(
            fan_out_fan_in, catalog, hook_manager=_create_hook_manager()
        )
        assert "Z" in result
        assert result["Z"] == (42, 42, 42)

    def test_memory_dataset_input(self, fan_out_fan_in):
        catalog = DataCatalog({"A": MemoryDataset("42")})
        result = ThreadRunner().run(fan_out_fan_in, catalog)
        assert "Z" in result
        assert result["Z"] == ("42", "42", "42")

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


class TestMaxWorkers:
    @pytest.mark.parametrize(
        "user_specified_number, expected_number",
        [
            # The pipeline only needs 3 threads, no need for more
            (6, 3),
            (None, 3),
        ],
    )
    def test_specified_max_workers(
        self,
        mocker,
        fan_out_fan_in,
        catalog,
        user_specified_number,
        expected_number,
    ):  # noqa: PLR0913
        """
        We initialize the runner with max_workers=4.
        `fan_out_fan_in` pipeline needs 3 threads.
        A pool with 3 workers should be used.
        """
        executor_cls_mock = mocker.patch(
            "kedro.runner.thread_runner.ThreadPoolExecutor",