refactor(server.py): remove duplicate error logging in call_tool function to simplify error handling

refactor(shell_executor.py): remove debug print statements to clean up code and improve readability
test(tests): add tests for redirection path validation and IO handle closing functionality to ensure robustness of shell executor

Author: tumf

src/mcp_shell_server/server.py

@@ -113,15 +113,6 @@ async def call_tool(name: str, arguments: Any) -> Sequence[TextContent]:
         return await tool_handler.run_tool(arguments)
 
     except Exception as e:
-        logger.error(traceback.format_exc())
-        logger.error(f"Error during call_tool: {str(e)}")
-        raise RuntimeError(str(e)) from e
-
-        logger.error(traceback.format_exc())
-        logger.error(f"Error during call_tool: {str(e)}")
-        raise RuntimeError(str(e)) from e
-
-
         logger.error(traceback.format_exc())
         logger.error(f"Error during call_tool: {str(e)}")
         raise RuntimeError(f"Error executing command: {str(e)}") from e

src/mcp_shell_server/shell_executor.py

@@ -419,10 +419,8 @@ async def execute(
                 return await self._execute_pipeline(commands, directory, timeout)
 
             # Then check for other shell operators
-            print("cleaned_command", cleaned_command)
             for token in cleaned_command:
                 if token in [";", "&&", "||"]:
-                    print("token", token)
                     return {
                         "error": f"Unexpected shell operator: {token}",
                         "status": 1,
@@ -590,7 +588,6 @@ async def _execute_pipeline(
 
             for i, cmd in enumerate(parsed_commands):
                 shell_cmd = self._create_shell_command(cmd)
-                print(f"Executing command {i}: {shell_cmd}")  # デバッグ出力
 
                 process = await asyncio.create_subprocess_shell(
                     shell_cmd,
@@ -606,26 +603,15 @@ async def _execute_pipeline(
                         process.communicate(input=prev_stdout), timeout=timeout
                     )
 
-                    # デバッグ出力
-                    print(f"Command {i} stdout: {stdout!r}")
-                    print(f"Command {i} stderr: {stderr!r}")
-
-                    # 次のコマンドへの入力として使用
                     prev_stdout = stdout if stdout else b""
 
-                    # 最後のコマンドの場合は最終出力として保存
                     if i == len(parsed_commands) - 1:
                         final_stdout = stdout if stdout else b""
 
-                    # エラー出力を累積
                     final_stderr += stderr if stderr else b""
                     processes.append(process)
 
-                    # エラーチェック
                     if process.returncode != 0:
-                        print(
-                            f"Command {i} failed with code {process.returncode}"
-                        )  # デバッグ出力
                         raise ValueError(
                             f"Command failed with exit code {process.returncode}"
                         )
@@ -634,15 +620,11 @@ async def _execute_pipeline(
                     process.kill()
                     raise
 
-            # 出力の処理
             if last_stdout:
-                # ファイルにリイレクトされている場合
                 last_stdout.write(final_stdout.decode("utf-8", errors="replace"))
                 final_output = ""
             else:
-                # 標準出力に出力する場合
                 final_output = final_stdout.decode("utf-8", errors="replace")
-                print(f"Final output: {final_output!r}")  # デバッグ出力
 
             return {
                 "error": None,
@@ -654,7 +636,6 @@ async def _execute_pipeline(
             }
 
         except Exception as e:
-            print(f"Pipeline error: {str(e)}")  # デバッグ出力
             return {
                 "error": str(e),
                 "stdout": "",

tests/test_server.py

@@ -201,7 +201,7 @@ async def test_invalid_command_parameter():
 @pytest.mark.asyncio
 async def test_disallowed_command(monkeypatch):
     """Test error handling for disallowed command"""
-    monkeypatch.setenv("ALLOW_COMMANDS", "ls") # Add allowed command
+    monkeypatch.setenv("ALLOW_COMMANDS", "ls")  # Add allowed command
     with pytest.raises(RuntimeError) as exc:
         await call_tool(
             "shell_execute",

tests/test_shell_executor.py

@@ -1,5 +1,6 @@
 import os
 import tempfile
+from typing import IO
 
 import pytest
 
@@ -501,3 +502,71 @@ def test_validate_pipeline(executor, monkeypatch):
         executor._validate_pipeline(["echo", "hello", "|", "grep", "h", "&&", "ls"])
     assert "Unexpected shell operator in pipeline: &&" in str(exc.value)
     assert executor._preprocess_command([]) == []
+
+
+def test_redirection_path_validation(executor):
+    """Test validation of redirection paths"""
+    # Test missing output redirection path
+    with pytest.raises(ValueError, match="Missing path for output redirection"):
+        executor._parse_command(["echo", "hello", ">"])
+
+    # Test missing input redirection path
+    with pytest.raises(ValueError, match="Missing path for input redirection"):
+        executor._parse_command(["cat", "<"])
+
+    # Test operator as redirection target
+    with pytest.raises(ValueError, match="Invalid redirection target: operator found"):
+        executor._parse_command(["echo", "hello", ">", ">"])
+
+    # Test multiple operators
+    with pytest.raises(ValueError, match="Invalid redirection target: operator found"):
+        executor._parse_command(["echo", "hello", ">", ">>", "file.txt"])
+
+
+@pytest.mark.asyncio
+async def test_io_handle_close(executor, temp_test_dir, monkeypatch, mocker):
+    """Test IO handle closing functionality"""
+    monkeypatch.setenv("ALLOW_COMMANDS", "echo")
+    test_file = os.path.join(temp_test_dir, "test.txt")
+
+    # Create file handler that will raise IOError on close
+    mock_file = mocker.MagicMock(spec=IO)
+    mock_file.close.side_effect = IOError("Failed to close file")
+
+    # Patch the open function to return our mock
+    mocker.patch("builtins.open", return_value=mock_file)
+    await executor.execute(["echo", "hello", ">", test_file], directory=temp_test_dir)
+
+    # Verify our mock's close method was called
+    assert mock_file.close.called
+
+
+def test_preprocess_command_pipeline(executor):
+    """Test pipeline command preprocessing functionality"""
+    # Test empty command
+    assert executor._preprocess_command([]) == []
+
+    # Test single command without pipe
+    assert executor._preprocess_command(["echo", "hello"]) == ["echo", "hello"]
+
+    # Test simple pipe
+    assert executor._preprocess_command(["echo", "hello", "|", "grep", "h"]) == [
+        "echo",
+        "hello",
+        "|",
+        "grep",
+        "h",
+    ]
+
+    # Test multiple pipes
+    assert executor._preprocess_command(
+        ["cat", "file", "|", "grep", "pattern", "|", "wc", "-l"]
+    ) == ["cat", "file", "|", "grep", "pattern", "|", "wc", "-l"]
+
+    # Test command with attached pipe operator
+    assert executor._preprocess_command(["echo|", "grep", "pattern"]) == [
+        "echo",
+        "|",
+        "grep",
+        "pattern",
+    ]