refactor: Move command processing logic to CommandPreProcessor

Author: tumf

src/mcp_shell_server/shell_executor.py

@@ -4,8 +4,9 @@
 import pwd
 import shlex
 import time
-from typing import IO, Any, Dict, List, Optional, Tuple, Union
+from typing import IO, Any, Dict, List, Optional
 
+from mcp_shell_server.command_preprocessor import CommandPreProcessor
 from mcp_shell_server.command_validator import CommandValidator
 from mcp_shell_server.directory_manager import DirectoryManager
 from mcp_shell_server.io_redirection_handler import IORedirectionHandler
@@ -23,38 +24,7 @@ def __init__(self):
         self.validator = CommandValidator()
         self.directory_manager = DirectoryManager()
         self.io_handler = IORedirectionHandler()
-
-    def _clean_command(self, command: List[str]) -> List[str]:
-        """
-        Clean command by trimming whitespace from each part.
-        Removes empty strings but preserves arguments that are meant to be spaces.
-
-        Args:
-            command (List[str]): Original command and its arguments
-
-        Returns:
-            List[str]: Cleaned command
-        """
-        return [arg for arg in command if arg]  # Remove empty strings
-
-    def _create_shell_command(self, command: List[str]) -> str:
-        """
-        Create a shell command string from a list of arguments.
-        Handles wildcards and arguments properly.
-        """
-        if not command:
-            return ""
-
-        escaped_args = []
-        for arg in command:
-            if arg.isspace():
-                # Wrap space-only arguments in single quotes
-                escaped_args.append(f"'{arg}'")
-            else:
-                # Properly escape all arguments including those with wildcards
-                escaped_args.append(shlex.quote(arg.strip()))
-
-        return " ".join(escaped_args)
+        self.preprocessor = CommandPreProcessor()
 
     def _validate_command(self, command: List[str]) -> None:
         """
@@ -95,97 +65,6 @@ def _validate_pipeline(self, commands: List[str]) -> Dict[str, str]:
         """
         return self.validator.validate_pipeline(commands)
 
-    def _split_pipe_commands(self, command: List[str]) -> List[List[str]]:
-        """
-        Split commands by pipe operator into separate commands.
-
-        Args:
-            command (List[str]): Command and its arguments with pipe operators
-
-        Returns:
-            List[List[str]]: List of commands split by pipe operator
-        """
-        commands: List[List[str]] = []
-        current_command: List[str] = []
-
-        for arg in command:
-            if arg.strip() == "|":
-                if current_command:
-                    commands.append(current_command)
-                    current_command = []
-            else:
-                current_command.append(arg)
-
-        if current_command:
-            commands.append(current_command)
-
-        return commands
-
-    def _parse_command(
-        self, command: List[str]
-    ) -> Tuple[List[str], Dict[str, Union[None, str, bool]]]:
-        """
-        Parse command and extract redirections.
-        """
-        cmd = []
-        redirects: Dict[str, Union[None, str, bool]] = {
-            "stdin": None,
-            "stdout": None,
-            "stdout_append": False,
-        }
-
-        i = 0
-        while i < len(command):
-            token = command[i]
-
-            # Shell operators check
-            if token in ["|", ";", "&&", "||"]:
-                raise ValueError(f"Unexpected shell operator: {token}")
-
-            # Output redirection
-            if token in [">", ">>"]:
-                if i + 1 >= len(command):
-                    raise ValueError("Missing path for output redirection")
-                if i + 1 < len(command) and command[i + 1] in [">", ">>", "<"]:
-                    raise ValueError("Invalid redirection target: operator found")
-                path = command[i + 1]
-                redirects["stdout"] = path
-                redirects["stdout_append"] = token == ">>"
-                i += 2
-                continue
-
-            # Input redirection
-            if token == "<":
-                if i + 1 >= len(command):
-                    raise ValueError("Missing path for input redirection")
-                path = command[i + 1]
-                redirects["stdin"] = path
-                i += 2
-                continue
-
-            cmd.append(token)
-            i += 1
-
-        return cmd, redirects
-
-    def _preprocess_command(self, command: List[str]) -> List[str]:
-        """
-        Preprocess the command to handle cases where '|' is attached to a command.
-        """
-        preprocessed_command = []
-        for token in command:
-            if token in ["||", "&&", ";"]:  # 特別なシェル演算子を保護
-                preprocessed_command.append(token)
-            elif "|" in token and token != "|":
-                parts = token.split("|")
-                preprocessed_command.extend(
-                    [part.strip() for part in parts if part.strip()]
-                )
-                preprocessed_command.append("|")
-            else:
-                preprocessed_command.append(token)
-        return preprocessed_command
-
     def _get_default_shell(self) -> str:
         """Get the login shell of the current user"""
         try:
@@ -218,8 +97,8 @@ async def execute(
                 }
 
             # Process command
-            preprocessed_command = self._preprocess_command(command)
-            cleaned_command = self._clean_command(preprocessed_command)
+            preprocessed_command = self.preprocessor.preprocess_command(command)
+            cleaned_command = self.preprocessor.clean_command(preprocessed_command)
             if not cleaned_command:
                 return {
                     "error": "Empty command",
@@ -245,19 +124,9 @@ async def execute(
                         }
 
                     # Split commands
-                    commands: List[List[str]] = []
-                    current_cmd: List[str] = []
-                    for token in cleaned_command:
-                        if token == "|":
-                            if current_cmd:
-                                commands.append(current_cmd)
-                                current_cmd = []
-                            else:
-                                raise ValueError("Empty command before pipe operator")
-                        else:
-                            current_cmd.append(token)
-                    if current_cmd:
-                        commands.append(current_cmd)
+                    commands = self.preprocessor.split_pipe_commands(cleaned_command)
+                    if not commands:
+                        raise ValueError("Empty command before pipe operator")
 
                     return await self._execute_pipeline(
                         commands, directory, timeout, envs
@@ -286,7 +155,7 @@ async def execute(
 
             # Single command execution
             try:
-                cmd, redirects = self._parse_command(cleaned_command)
+                cmd, redirects = self.preprocessor.parse_command(cleaned_command)
             except ValueError as e:
                 return {
                     "error": str(e),
@@ -349,7 +218,7 @@ async def execute(
 
             # Execute the command with interactive shell
             shell = self._get_default_shell()
-            shell_cmd = self._create_shell_command(cmd)
+            shell_cmd = self.preprocessor.create_shell_command(cmd)
             shell_cmd = f"{shell} -i -c {shlex.quote(shell_cmd)}"
 
             process = await asyncio.create_subprocess_shell(
@@ -488,7 +357,7 @@ async def _execute_pipeline(
             # Execute each command in the pipeline
             for i, cmd in enumerate(parsed_commands):
                 # Create the shell command
-                shell_cmd = self._create_shell_command(cmd)
+                shell_cmd = self.preprocessor.create_shell_command(cmd)
 
                 # Apply interactive mode to first command only
                 if i == 0:

tests/test_shell_executor.py

@@ -440,55 +440,55 @@ def test_validate_redirection_syntax(executor):
 def test_create_shell_command(executor):
     """Test shell command creation with various input combinations"""
     # Test basic command
-    assert executor._create_shell_command(["echo", "hello"]) == "echo hello"
+    assert executor.preprocessor.create_shell_command(["echo", "hello"]) == "echo hello"
 
     # Test command with space-only argument
-    assert executor._create_shell_command(["echo", " "]) == "echo ' '"
+    assert executor.preprocessor.create_shell_command(["echo", " "]) == "echo ' '"
 
     # Test command with wildcards
-    assert executor._create_shell_command(["ls", "*.txt"]) == "ls '*.txt'"
+    assert executor.preprocessor.create_shell_command(["ls", "*.txt"]) == "ls '*.txt'"
 
     # Test command with special characters
     assert (
-        executor._create_shell_command(["echo", "hello;", "world"])
+        executor.preprocessor.create_shell_command(["echo", "hello;", "world"])
         == "echo 'hello;' world"
     )
 
     # Test empty command
-    assert executor._create_shell_command([]) == ""
+    assert executor.preprocessor.create_shell_command([]) == ""
 
 
 def test_preprocess_command(executor):
     """Test command preprocessing for pipeline handling"""
     # Test basic command
-    assert executor._preprocess_command(["ls"]) == ["ls"]
+    assert executor.preprocessor.preprocess_command(["ls"]) == ["ls"]
 
     # Test command with separate pipe
-    assert executor._preprocess_command(["ls", "|", "grep", "test"]) == [
+    assert executor.preprocessor.preprocess_command(["ls", "|", "grep", "test"]) == [
         "ls",
         "|",
         "grep",
         "test",
     ]
 
     # Test command with attached pipe
-    assert executor._preprocess_command(["ls|", "grep", "test"]) == [
+    assert executor.preprocessor.preprocess_command(["ls|", "grep", "test"]) == [
         "ls",
         "|",
         "grep",
         "test",
     ]
 
     # Test command with special operators
-    assert executor._preprocess_command(["echo", "hello", "&&", "ls"]) == [
+    assert executor.preprocessor.preprocess_command(["echo", "hello", "&&", "ls"]) == [
         "echo",
         "hello",
         "&&",
         "ls",
     ]
 
     # Test empty command
-    assert executor._preprocess_command([]) == []
+    assert executor.preprocessor.preprocess_command([]) == []
 
 
 def test_validate_pipeline(executor, monkeypatch):
@@ -507,33 +507,35 @@ def test_validate_pipeline(executor, monkeypatch):
 
     # Test disallowed commands in pipeline
     with pytest.raises(ValueError) as exc:
-        executor._validate_pipeline(["rm", "|", "grep", "test"])
+        executor.validator.validate_pipeline(["rm", "|", "grep", "test"])
     assert "Command not allowed: rm" in str(exc.value)
 
     # Test shell operators in pipeline
     with pytest.raises(ValueError) as exc:
-        executor._validate_pipeline(["echo", "hello", "|", "grep", "h", "&&", "ls"])
+        executor.validator.validate_pipeline(
+            ["echo", "hello", "|", "grep", "h", "&&", "ls"]
+        )
     assert "Unexpected shell operator in pipeline: &&" in str(exc.value)
-    assert executor._preprocess_command([]) == []
+    assert executor.preprocessor.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", ">"])
+        executor.preprocessor.parse_command(["echo", "hello", ">"])
 
     # Test missing input redirection path
     with pytest.raises(ValueError, match="Missing path for input redirection"):
-        executor._parse_command(["cat", "<"])
+        executor.preprocessor.parse_command(["cat", "<"])
 
     # Test operator as redirection target
     with pytest.raises(ValueError, match="Invalid redirection target: operator found"):
-        executor._parse_command(["echo", "hello", ">", ">"])
+        executor.preprocessor.parse_command(["echo", "hello", ">", ">"])
 
     # Test multiple operators
     with pytest.raises(ValueError, match="Invalid redirection target: operator found"):
-        executor._parse_command(["echo", "hello", ">", ">>", "file.txt"])
+        executor.preprocessor.parse_command(["echo", "hello", ">", ">>", "file.txt"])
 
 
 @pytest.mark.asyncio
@@ -565,13 +567,18 @@ async def test_io_handle_close(executor, temp_test_dir, monkeypatch, mocker):
 def test_preprocess_command_pipeline(executor):
     """Test pipeline command preprocessing functionality"""
     # Test empty command
-    assert executor._preprocess_command([]) == []
+    assert executor.preprocessor.preprocess_command([]) == []
 
     # Test single command without pipe
-    assert executor._preprocess_command(["echo", "hello"]) == ["echo", "hello"]
+    assert executor.preprocessor.preprocess_command(["echo", "hello"]) == [
+        "echo",
+        "hello",
+    ]
 
     # Test simple pipe
-    assert executor._preprocess_command(["echo", "hello", "|", "grep", "h"]) == [
+    assert executor.preprocessor.preprocess_command(
+        ["echo", "hello", "|", "grep", "h"]
+    ) == [
         "echo",
         "hello",
         "|",
@@ -580,12 +587,12 @@ def test_preprocess_command_pipeline(executor):
     ]
 
     # Test multiple pipes
-    assert executor._preprocess_command(
+    assert executor.preprocessor.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"]) == [
+    assert executor.preprocessor.preprocess_command(["echo|", "grep", "pattern"]) == [
         "echo",
         "|",
         "grep",

tests/test_shell_executor_pipeline.py

@@ -30,18 +30,20 @@ def temp_test_dir():
 async def test_pipeline_split(executor):
     """Test pipeline command splitting functionality"""
     # Test basic pipe command
-    commands = executor._split_pipe_commands(["echo", "hello", "|", "grep", "h"])
+    commands = executor.preprocessor.split_pipe_commands(
+        ["echo", "hello", "|", "grep", "h"]
+    )
     assert len(commands) == 2
     assert commands[0] == ["echo", "hello"]
     assert commands[1] == ["grep", "h"]
 
     # Test empty pipe sections
-    commands = executor._split_pipe_commands(["|", "grep", "pattern"])
+    commands = executor.preprocessor.split_pipe_commands(["|", "grep", "pattern"])
     assert len(commands) == 1
     assert commands[0] == ["grep", "pattern"]
 
     # Test multiple pipes
-    commands = executor._split_pipe_commands(
+    commands = executor.preprocessor.split_pipe_commands(
         ["cat", "file.txt", "|", "grep", "pattern", "|", "wc", "-l"]
     )
     assert len(commands) == 3
@@ -50,7 +52,7 @@ async def test_pipeline_split(executor):
     assert commands[2] == ["wc", "-l"]
 
     # Test trailing pipe
-    commands = executor._split_pipe_commands(["echo", "hello", "|"])
+    commands = executor.preprocessor.split_pipe_commands(["echo", "hello", "|"])
     assert len(commands) == 1
     assert commands[0] == ["echo", "hello"]