diff --git a/astrbot/core/computer/booters/local.py b/astrbot/core/computer/booters/local.py index 44122361d6..7328f14712 100644 --- a/astrbot/core/computer/booters/local.py +++ b/astrbot/core/computer/booters/local.py @@ -79,8 +79,15 @@ def _try_decode(encoding: str) -> str | None: return output.decode("utf-8", errors="replace") -def _decode_shell_output(output: bytes | None) -> str: - return _decode_bytes_with_fallback(output, preferred_encoding="utf-8") +def _decode_process_output( + output: bytes | None, + *, + normalize_newlines: bool = False, +) -> str: + decoded = _decode_bytes_with_fallback(output, preferred_encoding="utf-8") + if normalize_newlines: + decoded = decoded.replace("\r\n", "\n") + return decoded @dataclass @@ -127,8 +134,8 @@ def _run() -> dict[str, Any]: capture_output=True, ) return { - "stdout": _decode_shell_output(result.stdout), - "stderr": _decode_shell_output(result.stderr), + "stdout": _decode_process_output(result.stdout), + "stderr": _decode_process_output(result.stderr), "exit_code": result.returncode, } @@ -150,10 +157,20 @@ def _run() -> dict[str, Any]: [os.environ.get("PYTHON", sys.executable), "-c", code], timeout=timeout, capture_output=True, - text=True, + text=False, + ) + stdout = ( + "" + if silent + else _decode_process_output( + result.stdout, + normalize_newlines=True, + ) + ) + stderr = _decode_process_output( + result.stderr, + normalize_newlines=True, ) - stdout = "" if silent else result.stdout - stderr = result.stderr if result.returncode != 0 else "" return { "data": { "output": {"text": stdout, "images": []}, diff --git a/tests/unit/test_computer.py b/tests/unit/test_computer.py index 71b31a301a..6d8faac9eb 100644 --- a/tests/unit/test_computer.py +++ b/tests/unit/test_computer.py @@ -5,6 +5,7 @@ """ import sys +import subprocess from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -225,6 +226,64 @@ async def test_exec_return_value(self): result = await python.exec("result = 1 + 1\nprint(result)") assert "2" in result["data"]["output"]["text"] + @pytest.mark.asyncio + async def test_exec_decodes_non_utf8_stdout_with_fallback(self): + """Test Python execution decodes captured bytes with fallback encodings.""" + python = LocalPythonComponent() + + def fake_run(*args, **kwargs): + assert kwargs.get("text") is False + return subprocess.CompletedProcess( + args=args[0], + returncode=0, + stdout="中文输出\n".encode("gbk"), + stderr=b"", + ) + + with patch("astrbot.core.computer.booters.local.subprocess.run", fake_run): + result = await python.exec("print('中文输出')") + + assert result["data"]["output"]["text"] == "中文输出\n" + + @pytest.mark.asyncio + async def test_exec_preserves_lone_carriage_returns(self): + """Test Python execution preserves lone carriage returns used by progress output.""" + python = LocalPythonComponent() + + def fake_run(*args, **kwargs): + assert kwargs.get("text") is False + return subprocess.CompletedProcess( + args=args[0], + returncode=0, + stdout=b"progress 10%\rprogress 20%\r\ncomplete\r", + stderr=b"", + ) + + with patch("astrbot.core.computer.booters.local.subprocess.run", fake_run): + result = await python.exec("print('progress')") + + assert result["data"]["output"]["text"] == "progress 10%\rprogress 20%\ncomplete\r" + + @pytest.mark.asyncio + async def test_exec_keeps_success_stderr(self): + """Test Python execution keeps diagnostic stderr even on success.""" + python = LocalPythonComponent() + + def fake_run(*args, **kwargs): + assert kwargs.get("text") is False + return subprocess.CompletedProcess( + args=args[0], + returncode=0, + stdout=b"ok\n", + stderr=b"warning\r\n", + ) + + with patch("astrbot.core.computer.booters.local.subprocess.run", fake_run): + result = await python.exec("print('ok')") + + assert result["data"]["output"]["text"] == "ok\n" + assert result["data"]["error"] == "warning\n" + class TestLocalFileSystemComponent: """Tests for LocalFileSystemComponent."""