diff --git a/src/openjd/adaptor_runtime/adaptors/_base_adaptor.py b/src/openjd/adaptor_runtime/adaptors/_base_adaptor.py index 70912b39..35db17e2 100644 --- a/src/openjd/adaptor_runtime/adaptors/_base_adaptor.py +++ b/src/openjd/adaptor_runtime/adaptors/_base_adaptor.py @@ -13,6 +13,7 @@ from typing import Type from typing import TypeVar +from ..process._logging import _ADAPTOR_OUTPUT_LEVEL from .._utils._constants import _OPENJD_PROGRESS_STDOUT_PREFIX, _OPENJD_STATUS_STDOUT_PREFIX from .configuration import AdaptorConfiguration, ConfigurationManager from .configuration._configuration_manager import ( @@ -194,16 +195,20 @@ def update_status( if progress is not None: if math.isfinite(progress): - sys.stdout.write(f"{cls._OPENJD_PROGRESS_STDOUT_PREFIX}{progress}{os.linesep}") - sys.stdout.flush() + _logger.log( + _ADAPTOR_OUTPUT_LEVEL, + f"{cls._OPENJD_PROGRESS_STDOUT_PREFIX}{progress}", + ) else: _logger.warning( f"Attempted to set progress to something non-finite: {progress}. " "Ignoring progress update." ) if status_message is not None: - sys.stdout.write(f"{cls._OPENJD_STATUS_STDOUT_PREFIX}{status_message}{os.linesep}") - sys.stdout.flush() + _logger.log( + _ADAPTOR_OUTPUT_LEVEL, + f"{cls._OPENJD_STATUS_STDOUT_PREFIX}{status_message}", + ) @property def path_mapping_rules(self) -> list[PathMappingRule]: diff --git a/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/ProgressReportingAdaptor.json b/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/ProgressReportingAdaptor.json new file mode 100644 index 00000000..9e26dfee --- /dev/null +++ b/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/ProgressReportingAdaptor.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/__init__.py b/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/__init__.py new file mode 100644 index 00000000..95ccf203 --- /dev/null +++ b/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/__init__.py @@ -0,0 +1,5 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +from .adaptor import ProgressReportingAdaptor + +__all__ = ["ProgressReportingAdaptor"] diff --git a/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/__main__.py b/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/__main__.py new file mode 100644 index 00000000..e1eb0c38 --- /dev/null +++ b/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/__main__.py @@ -0,0 +1,18 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +import sys + +from openjd.adaptor_runtime import EntryPoint +from .adaptor import ProgressReportingAdaptor + + +def main(): + package_name = vars(sys.modules[__name__])["__package__"] + if not package_name: + raise RuntimeError(f"Must be run as a module. Do not run {__file__} directly") + + EntryPoint(ProgressReportingAdaptor).start() + + +if __name__ == "__main__": + main() diff --git a/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/adaptor.py b/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/adaptor.py new file mode 100644 index 00000000..61001524 --- /dev/null +++ b/test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/adaptor.py @@ -0,0 +1,19 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +from openjd.adaptor_runtime.adaptors import Adaptor, SemanticVersion + + +class ProgressReportingAdaptor(Adaptor): + """ + A minimal adaptor that calls update_status to report progress during on_run. + Used to test that progress messages propagate from the background adaptor + to the frontend process's stdout. + """ + + @property + def integration_data_interface_version(self) -> SemanticVersion: + return SemanticVersion(major=0, minor=1) + + def on_run(self, run_data: dict) -> None: + progress = run_data.get("progress", 50.0) + self.update_status(progress=progress) diff --git a/test/openjd/adaptor_runtime/integ/adaptors/test_integration_adaptor.py b/test/openjd/adaptor_runtime/integ/adaptors/test_integration_adaptor.py index b3c49bd8..30bf0136 100644 --- a/test/openjd/adaptor_runtime/integ/adaptors/test_integration_adaptor.py +++ b/test/openjd/adaptor_runtime/integ/adaptors/test_integration_adaptor.py @@ -17,7 +17,7 @@ class TestRun: _OPENJD_PROGRESS_STDOUT_PREFIX: str = "openjd_progress: " _OPENJD_STATUS_STDOUT_PREFIX: str = "openjd_status: " - def test_run(self, capsys) -> None: + def test_run(self, capsys, caplog) -> None: first_progress = 0.0 first_status_message = "Starting the printing of run_data" second_progress = 100.0 @@ -46,14 +46,15 @@ def integration_data_interface_version(self) -> SemanticVersion: adaptor = PrintAdaptor(init_data) # WHEN - adaptor._run(run_data) + with caplog.at_level(0): + adaptor._run(run_data) result = capsys.readouterr().out.strip() # THEN - assert f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{first_progress}" in result - assert f"{self._OPENJD_STATUS_STDOUT_PREFIX}{first_status_message}" in result - assert f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{second_progress}" in result - assert f"{self._OPENJD_STATUS_STDOUT_PREFIX}{second_status_message}" in result + assert f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{first_progress}" in caplog.text + assert f"{self._OPENJD_STATUS_STDOUT_PREFIX}{first_status_message}" in caplog.text + assert f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{second_progress}" in caplog.text + assert f"{self._OPENJD_STATUS_STDOUT_PREFIX}{second_status_message}" in caplog.text assert "run_data:\n\tkey1 = value1\n\tkey2 = value2\n\tkey3 = value3" in result def test_start_end_cleanup(self, tmpdir, capsys) -> None: diff --git a/test/openjd/adaptor_runtime/integ/test_progress_in_background_mode.py b/test/openjd/adaptor_runtime/integ/test_progress_in_background_mode.py new file mode 100644 index 00000000..bacef5b1 --- /dev/null +++ b/test/openjd/adaptor_runtime/integ/test_progress_in_background_mode.py @@ -0,0 +1,151 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +""" +Test to reproduce a bug where progress reporting via update_status() does not +propagate from the background adaptor to the frontend process's stdout. + +The flow is: + 1. Application outputs progress info + 2. Background adaptor (backend process) calls update_status(progress=X) + 3. update_status writes "openjd_progress: X" to sys.stdout + 4. Frontend process should emit "openjd_progress: X" on its own stdout + so the worker agent can detect it + +The bug: In daemon/background mode, the backend process replaces its stream +handler with a LogBufferHandler. But update_status() writes directly to +sys.stdout (not through the logging framework), so the progress message +never reaches the log buffer, and thus never reaches the frontend. +""" + +from __future__ import annotations + +import json +import os +import pathlib +import sys +from logging import INFO +from unittest.mock import patch +from pathlib import Path + +import pytest + +import openjd.adaptor_runtime._entrypoint as runtime_entrypoint +from openjd.adaptor_runtime import EntryPoint + +mod_path = Path(__file__).parent.resolve() +sys.path.append(str(mod_path)) + +if (_pypath := os.environ.get("PYTHONPATH")) is not None: + os.environ["PYTHONPATH"] = os.pathsep.join([_pypath, str(mod_path)]) +else: + os.environ["PYTHONPATH"] = str(mod_path) + +from ProgressReportingAdaptor import ProgressReportingAdaptor # noqa: E402 + + +class TestProgressReportingInDaemonMode: + """ + Tests that progress messages from update_status() in the background adaptor + are propagated to the frontend process's stdout so the worker agent can + detect them. + """ + + def test_progress_propagates_to_frontend_stdout( + self, + capfd: pytest.CaptureFixture, + caplog: pytest.LogCaptureFixture, + tmp_path: pathlib.Path, + ): + """ + Reproduces the bug: when running in daemon mode, calling + update_status(progress=50) in the backend adaptor should result in + "openjd_progress: 50" appearing on the frontend's stdout. + + The worker agent monitors the frontend process's stdout for these + messages. If they don't appear, the worker never knows about progress. + """ + caplog.set_level(INFO) + connection_file = tmp_path / "connection.json" + entrypoint = EntryPoint(ProgressReportingAdaptor) + + # Start the daemon + start_argv = [ + "program_filename.py", + "daemon", + "start", + "--connection-file", + str(connection_file), + ] + with ( + patch.object(runtime_entrypoint.sys, "argv", start_argv), + patch.object(runtime_entrypoint.logging.Logger, "setLevel"), + ): + entrypoint.start() + + # Run with progress=50 + run_argv = [ + "program_filename.py", + "daemon", + "run", + "--connection-file", + str(connection_file), + "--run-data", + json.dumps({"progress": 50.0}), + ] + with ( + patch.object(runtime_entrypoint.sys, "argv", run_argv), + patch.object(runtime_entrypoint.logging.Logger, "setLevel"), + ): + entrypoint.start() + + # Stop the daemon + stop_argv = [ + "program_filename.py", + "daemon", + "stop", + "--connection-file", + str(connection_file), + ] + with ( + patch.object(runtime_entrypoint.sys, "argv", stop_argv), + patch.object(runtime_entrypoint.logging.Logger, "setLevel"), + ): + entrypoint.start() + + # Verify that "openjd_progress: 50" appeared on stdout. + # The worker agent is watching stdout for this pattern. + captured = capfd.readouterr() + assert "openjd_progress: 50" in captured.out, ( + f"Expected 'openjd_progress: 50' on stdout but got:\n{captured.out}\n\n" + f"Log output:\n{caplog.text}" + ) + + def test_progress_works_in_run_mode( + self, + capfd: pytest.CaptureFixture, + caplog: pytest.LogCaptureFixture, + ): + """ + Sanity check: progress reporting works fine in non-daemon 'run' mode + because update_status writes directly to sys.stdout and the worker + agent is directly reading from that process's stdout. + """ + caplog.set_level(INFO) + entrypoint = EntryPoint(ProgressReportingAdaptor) + + run_argv = [ + "program_filename.py", + "run", + "--run-data", + json.dumps({"progress": 75.0}), + ] + with ( + patch.object(runtime_entrypoint.sys, "argv", run_argv), + patch.object(runtime_entrypoint.logging.Logger, "setLevel"), + ): + entrypoint.start() + + captured = capfd.readouterr() + assert ( + "openjd_progress: 75" in captured.out + ), f"Expected 'openjd_progress: 75' on stdout but got:\n{captured.out}" diff --git a/test/openjd/adaptor_runtime/unit/adaptors/test_base_adaptor.py b/test/openjd/adaptor_runtime/unit/adaptors/test_base_adaptor.py index f8562cf8..ffd69b4c 100644 --- a/test/openjd/adaptor_runtime/unit/adaptors/test_base_adaptor.py +++ b/test/openjd/adaptor_runtime/unit/adaptors/test_base_adaptor.py @@ -316,16 +316,17 @@ class TestStatusUpdate: param(1e-5), ], ) - def test_progress_update(self, capsys, progress: float): + def test_progress_update(self, caplog, progress: float): """Tests just updating the progress""" # GIVEN expected = f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{progress}" # WHEN - BaseAdaptor.update_status(progress=progress) + with caplog.at_level(0): + BaseAdaptor.update_status(progress=progress) # THEN - assert expected in capsys.readouterr().out + assert expected in caplog.text @pytest.mark.parametrize( "status_message", @@ -335,16 +336,17 @@ def test_progress_update(self, capsys, progress: float): param(""), ], ) - def test_status_message_update(self, capsys, status_message: str): + def test_status_message_update(self, caplog, status_message: str): """Tests just updating the status message""" # GIVEN expected = f"{self._OPENJD_STATUS_STDOUT_PREFIX}{status_message}" # WHEN - BaseAdaptor.update_status(status_message=status_message) + with caplog.at_level(0): + BaseAdaptor.update_status(status_message=status_message) # THEN - assert expected in capsys.readouterr().out + assert expected in caplog.text @pytest.mark.parametrize( "progress,status_message", @@ -360,30 +362,29 @@ def test_status_message_update(self, capsys, status_message: str): ), ], ) - def test_status_update(self, capsys, progress: float, status_message: str): + def test_status_update(self, caplog, progress: float, status_message: str): """Tests updating both progress and status messages""" # GIVEN expected_progress = f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{progress}" expected_status_message = f"{self._OPENJD_STATUS_STDOUT_PREFIX}{status_message}" # WHEN - BaseAdaptor.update_status(progress=progress, status_message=status_message) + with caplog.at_level(0): + BaseAdaptor.update_status(progress=progress, status_message=status_message) # THEN - result = capsys.readouterr().out - assert expected_progress in result - assert expected_status_message in result + assert expected_progress in caplog.text + assert expected_status_message in caplog.text - def test_ignore_status_update(self, capsys): + def test_ignore_status_update(self, caplog): """Tests we don't send any message if there's nothing to report""" - # GIVEN - expected = "" # nothing was captured in stdout - # WHEN - BaseAdaptor.update_status(progress=None, status_message=None) - BaseAdaptor.update_status(progress=float("NaN")) - BaseAdaptor.update_status(progress=float("inf")) + with caplog.at_level(0): + BaseAdaptor.update_status(progress=None, status_message=None) + BaseAdaptor.update_status(progress=float("NaN")) + BaseAdaptor.update_status(progress=float("inf")) # THEN - result = capsys.readouterr().out - assert expected == result + # Only warning messages should be logged, no progress/status messages + assert self._OPENJD_PROGRESS_STDOUT_PREFIX not in caplog.text + assert self._OPENJD_STATUS_STDOUT_PREFIX not in caplog.text