diff --git a/gui_agents/s3/utils/local_env.py b/gui_agents/s3/utils/local_env.py index d69b1b5f..7ba94847 100644 --- a/gui_agents/s3/utils/local_env.py +++ b/gui_agents/s3/utils/local_env.py @@ -1,6 +1,14 @@ +import locale import subprocess import sys -from typing import Dict +from typing import Dict, Optional, Union + + +def _decode_output(output: Optional[Union[bytes, str]]) -> str: + # TimeoutExpired captures bytes even with text=True, possibly ending mid-character. + if isinstance(output, bytes): + return output.decode(locale.getpreferredencoding(False), errors="replace") + return output or "" class LocalController: @@ -34,7 +42,7 @@ def run_bash_script(self, code: str, timeout: int = 30) -> Dict: return { "status": "error", "returncode": -1, - "output": e.stdout or "", + "output": _decode_output(e.stdout) + _decode_output(e.stderr), "error": f"TimeoutExpired: {str(e)}", } except Exception as e: diff --git a/tests/test_local_env.py b/tests/test_local_env.py new file mode 100644 index 00000000..d354fa9c --- /dev/null +++ b/tests/test_local_env.py @@ -0,0 +1,81 @@ +import json +import locale +import os +import subprocess +import unittest +from unittest.mock import patch + +from gui_agents.s3.utils.local_env import LocalController + + +class TestBashTimeoutOutput(unittest.TestCase): + def test_timeout_output_remains_json_serializable(self): + error = subprocess.TimeoutExpired( + "bash", 0.1, output=b"partial stdout", stderr=b"partial stderr" + ) + with patch("subprocess.run", side_effect=error): + result = LocalController().run_bash_script("unused", timeout=1) + + self.assertEqual(result["status"], "error") + self.assertEqual(result["returncode"], -1) + self.assertEqual(result["output"], "partial stdoutpartial stderr") + self.assertIn("TimeoutExpired", result["error"]) + json.dumps(result) + + def test_timeout_decodes_using_the_subprocess_text_encoding(self): + error = subprocess.TimeoutExpired("bash", 1, output=b"caf\xe9") + with patch("subprocess.run", side_effect=error), patch.object( + locale, "getpreferredencoding", return_value="cp1252" + ): + result = LocalController().run_bash_script("unused") + self.assertEqual(result["output"], "café") + json.dumps(result) + + def test_partial_invalid_output_does_not_mask_timeout(self): + error = subprocess.TimeoutExpired("bash", 1, output=b"partial\xff") + with patch("subprocess.run", side_effect=error), patch.object( + locale, "getpreferredencoding", return_value="utf-8" + ): + result = LocalController().run_bash_script("unused") + self.assertEqual(result["output"], "partial\ufffd") + self.assertIn("TimeoutExpired", result["error"]) + + def test_timeout_also_preserves_text_or_missing_output(self): + for stdout, stderr, expected in [ + (None, None, ""), + ("stdout", "stderr", "stdoutstderr"), + (None, b"stderr", "stderr"), + ]: + with self.subTest(stdout=stdout, stderr=stderr): + error = subprocess.TimeoutExpired( + "bash", 1, output=stdout, stderr=stderr + ) + with patch("subprocess.run", side_effect=error): + result = LocalController().run_bash_script("unused") + self.assertEqual(result["output"], expected) + json.dumps(result) + + @unittest.skipUnless(os.path.exists("/bin/bash"), "requires bash") + def test_real_timeout_keeps_partial_output_as_text(self): + result = LocalController().run_bash_script( + "printf 'stdout'; printf 'stderr' >&2; while :; do :; done", timeout=0.5 + ) + self.assertEqual(result["status"], "error") + # A login shell may also print output from the user's profile. + self.assertIsInstance(result["output"], str) + self.assertIn("stdout", result["output"]) + self.assertIn("stderr", result["output"]) + json.dumps(result) + + def test_success_and_nonzero_exit_remain_text(self): + for code in (0, 2): + with self.subTest(code=code): + completed = subprocess.CompletedProcess( + [], code, stdout="stdout", stderr="stderr" + ) + with patch("subprocess.run", return_value=completed): + result = LocalController().run_bash_script("unused") + self.assertEqual(result["output"], "stdoutstderr") + self.assertEqual(result["returncode"], code) + self.assertEqual(result["status"], "ok" if code == 0 else "error") + json.dumps(result)