diff --git a/src/kimi_cli/tools/shell/__init__.py b/src/kimi_cli/tools/shell/__init__.py index 154cd104..c31b60e9 100644 --- a/src/kimi_cli/tools/shell/__init__.py +++ b/src/kimi_cli/tools/shell/__init__.py @@ -82,6 +82,13 @@ async def _stream_subprocess( stderr_cb: Callable[[bytes], None], timeout: int, ) -> int: + if platform.system() == "Windows": + # Use cmd.exe explicitly so Windows syntax (e.g., %VAR%) works. + args = ["cmd.exe", "/d", "/s", "/c", command] + else: + # Use bash for POSIX features relied on by tests (pipes, env, etc.). + args = ["bash", "-c", command] + async def _read_stream(stream: asyncio.StreamReader, cb: Callable[[bytes], None]): while True: line = await stream.readline() @@ -91,8 +98,8 @@ async def _read_stream(stream: asyncio.StreamReader, cb: Callable[[bytes], None] break # FIXME: if the event loop is cancelled, an exception may be raised when the process finishes - process = await asyncio.create_subprocess_shell( - command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE + process = await asyncio.create_subprocess_exec( + *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) assert process.stdout is not None, "stdout is None" diff --git a/tests/test_cmd.py b/tests/test_cmd.py new file mode 100644 index 00000000..57095230 --- /dev/null +++ b/tests/test_cmd.py @@ -0,0 +1,72 @@ +"""Basic Windows cmd tests for the shell tool.""" + +from __future__ import annotations + +import platform + +import pytest +from inline_snapshot import snapshot +from kosong.tooling import ToolError, ToolOk + +from kaos.path import KaosPath +from kimi_cli.tools.shell import Params, Shell + +pytestmark = pytest.mark.skipif( + platform.system() != "Windows", reason="Shell cmd tests run only on Windows." +) + + +@pytest.mark.asyncio +async def test_simple_command(shell_tool: Shell): + """Ensure a basic cmd command runs.""" + result = await shell_tool(Params(command="echo Hello Windows")) + + assert isinstance(result, ToolOk) + assert isinstance(result.output, str) + assert result.output.strip() == snapshot("Hello Windows") + assert "Command executed successfully" in result.message + + +@pytest.mark.asyncio +async def test_command_with_error(shell_tool: Shell): + """Failing commands should return a ToolError with exit code info.""" + result = await shell_tool(Params(command='python -c "import sys; sys.exit(1)"')) + + assert isinstance(result, ToolError) + assert result.output == snapshot("") + assert "Command failed with exit code: 1" in result.message + assert "Failed with exit code: 1" in result.brief + + +@pytest.mark.asyncio +async def test_command_chaining(shell_tool: Shell): + """Chaining commands with && should work.""" + result = await shell_tool(Params(command="echo First&& echo Second")) + + assert isinstance(result, ToolOk) + assert isinstance(result.output, str) + assert result.output.replace("\r\n", "\n") == snapshot("First\nSecond\n") + + +@pytest.mark.asyncio +async def test_environment_variables(shell_tool: Shell): + """Environment variables should be usable within one cmd session.""" + result = await shell_tool(Params(command="set TEST_VAR=test_value&& echo %TEST_VAR%")) + + assert isinstance(result, ToolOk) + assert isinstance(result.output, str) + assert result.output.strip() == snapshot("test_value") + + +@pytest.mark.asyncio +async def test_file_operations(shell_tool: Shell, temp_work_dir: KaosPath): + """Basic file write/read using cmd redirection.""" + file_path = temp_work_dir / "test_file.txt" + + create_result = await shell_tool(Params(command=f'echo Test content> "{file_path}"')) + assert create_result == snapshot(ToolOk(output="", message="Command executed successfully.")) + + read_result = await shell_tool(Params(command=f'type "{file_path}"')) + assert read_result == snapshot( + ToolOk(output="Test content\r\n", message="Command executed successfully.") + )