test: add unit tests for Shell tool on Windows

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"

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.")
+    )