Merge branch 'develop'

Author: tumf

.github/workflows/publish.yml

@@ -1,15 +1,17 @@
 name: Publish
+
 on:
   push:
     tags:
       - "v*"
-    strategy:
-      matrix:
-        python-version: ["3.11"]
 
 jobs:
   test:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+
     steps:
       - uses: actions/checkout@v4
 
@@ -23,14 +25,14 @@ jobs:
           python -m pip install --upgrade pip
           pip install uv
 
-      - name: Install dev/test dependencies
+      - name: Install dependencies
         run: |
-          pip install -e ".[dev]"
-          pip install -e ".[test]"
+          uv venv
+          uv pip install -e ".[dev,test]"
 
-      - name: Run tests
+      - name: Run tests and checks
         run: |
-          make check
+          uv run make all
 
   publish:
     needs: test
@@ -44,14 +46,13 @@ jobs:
 
       - name: Update version from tag
         run: |
-          # Strip 'v' prefix from tag and update version.py
           VERSION=${GITHUB_REF#refs/tags/v}
           echo "__version__ = \"${VERSION}\"" > src/mcp_shell_server/version.py
 
-      - name: Set up Python ${{ matrix.python-version }}
+      - name: Set up Python 3.11
         uses: actions/setup-python@v5
         with:
-          python-version: ${{ matrix.python-version }}
+          python-version: "3.11"
 
       - name: Install uv
         run: |
@@ -60,10 +61,12 @@ jobs:
 
       - name: Build package
         run: |
-          uv build
+          uv venv
+          uv pip install build
+          uv run python -m build
 
       - name: Publish to PyPI
-        env:
-          PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
-        run: |
-          uv publish --token $PYPI_TOKEN
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          password: ${{ secrets.PYPI_TOKEN }}
+          password: ${{ secrets.PYPI_TOKEN }}

.github/workflows/test.yml

@@ -26,18 +26,18 @@ jobs:
           python -m pip install --upgrade pip
           pip install uv
 
-      - name: Install dev/test dependencies
+      - name: Install dependencies
         run: |
-          pip install -e ".[dev]"
-          pip install -e ".[test]"
+          uv venv
+          uv pip install -e ".[dev,test]"
 
-      - name: Run lint and typecheck
+      - name: Run format checks and typecheck
         run: |
-          make lint typecheck
+          uv run make check
 
       - name: Run tests with coverage
         run: |
-          pytest --cov --cov-report=xml
+          uv run make coverage
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v5

pyproject.toml

@@ -41,10 +41,15 @@ asyncio_mode = "strict"
 testpaths = "tests"
 # Set default event loop scope for async tests 
 asyncio_default_fixture_loop_scope = "function"
+markers = [
+    "macos: marks tests that should only run on macOS",
+    "slow: marks tests as slow running",
+]
 filterwarnings = [
     "ignore::RuntimeWarning:selectors:",
     "ignore::pytest.PytestUnhandledCoroutineWarning:",
     "ignore::pytest.PytestUnraisableExceptionWarning:",
+    "ignore::DeprecationWarning:pytest_asyncio.plugin:",
 ]
 
 [tool.ruff]

src/mcp_shell_server/command_preprocessor.py

@@ -0,0 +1,131 @@
+import shlex
+from typing import Dict, List, Tuple, Union
+
+
+class CommandPreProcessor:
+    """
+    Pre-processes and validates shell commands before execution
+    """
+
+    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 ["||", "&&", ";"]:  # Special shell operators
+                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 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)
+
+    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

src/mcp_shell_server/command_validator.py

@@ -0,0 +1,104 @@
+"""
+Provides validation for shell commands and ensures they are allowed to be executed.
+"""
+
+import os
+from typing import Dict, List
+
+
+class CommandValidator:
+    """
+    Validates shell commands against a whitelist and checks for unsafe operators.
+    """
+
+    def __init__(self):
+        """
+        Initialize the validator.
+        """
+        pass
+
+    def _get_allowed_commands(self) -> set[str]:
+        """Get the set of allowed commands from environment variables"""
+        allow_commands = os.environ.get("ALLOW_COMMANDS", "")
+        allowed_commands = os.environ.get("ALLOWED_COMMANDS", "")
+        commands = allow_commands + "," + allowed_commands
+        return {cmd.strip() for cmd in commands.split(",") if cmd.strip()}
+
+    def get_allowed_commands(self) -> list[str]:
+        """Get the list of allowed commands from environment variables"""
+        return list(self._get_allowed_commands())
+
+    def is_command_allowed(self, command: str) -> bool:
+        """Check if a command is in the allowed list"""
+        cmd = command.strip()
+        return cmd in self._get_allowed_commands()
+
+    def validate_no_shell_operators(self, cmd: str) -> None:
+        """
+        Validate that the command does not contain shell operators.
+
+        Args:
+            cmd (str): Command to validate
+
+        Raises:
+            ValueError: If the command contains shell operators
+        """
+        if cmd in [";", "&&", "||", "|"]:
+            raise ValueError(f"Unexpected shell operator: {cmd}")
+
+    def validate_pipeline(self, commands: List[str]) -> Dict[str, str]:
+        """
+        Validate pipeline command and ensure all parts are allowed.
+
+        Args:
+            commands (List[str]): List of commands to validate
+
+        Returns:
+            Dict[str, str]: Error message if validation fails, empty dict if success
+
+        Raises:
+            ValueError: If validation fails
+        """
+        current_cmd: List[str] = []
+
+        for token in commands:
+            if token == "|":
+                if not current_cmd:
+                    raise ValueError("Empty command before pipe operator")
+                if not self.is_command_allowed(current_cmd[0]):
+                    raise ValueError(f"Command not allowed: {current_cmd[0]}")
+                current_cmd = []
+            elif token in [";", "&&", "||"]:
+                raise ValueError(f"Unexpected shell operator in pipeline: {token}")
+            else:
+                current_cmd.append(token)
+
+        if current_cmd:
+            if not self.is_command_allowed(current_cmd[0]):
+                raise ValueError(f"Command not allowed: {current_cmd[0]}")
+
+        return {}
+
+    def validate_command(self, command: List[str]) -> None:
+        """
+        Validate if the command is allowed to be executed.
+
+        Args:
+            command (List[str]): Command and its arguments
+
+        Raises:
+            ValueError: If the command is empty, not allowed, or contains invalid shell operators
+        """
+        if not command:
+            raise ValueError("Empty command")
+
+        allowed_commands = self._get_allowed_commands()
+        if not allowed_commands:
+            raise ValueError(
+                "No commands are allowed. Please set ALLOW_COMMANDS environment variable."
+            )
+
+        # Clean and check the first command
+        cleaned_cmd = command[0].strip()
+        if cleaned_cmd not in allowed_commands:
+            raise ValueError(f"Command not allowed: {cleaned_cmd}")

src/mcp_shell_server/directory_manager.py

@@ -0,0 +1,52 @@
+import os
+from typing import Optional
+
+
+class DirectoryManager:
+    """
+    Manages directory validation and path operations for shell command execution.
+    """
+
+    def validate_directory(self, directory: Optional[str]) -> None:
+        """
+        Validate if the directory exists and is accessible.
+
+        Args:
+            directory (Optional[str]): Directory path to validate
+
+        Raises:
+            ValueError: If the directory doesn't exist, not absolute or is not accessible
+        """
+        # make directory required
+        if directory is None:
+            raise ValueError("Directory is required")
+
+        # verify directory is absolute path
+        if not os.path.isabs(directory):
+            raise ValueError(f"Directory must be an absolute path: {directory}")
+
+        if not os.path.exists(directory):
+            raise ValueError(f"Directory does not exist: {directory}")
+
+        if not os.path.isdir(directory):
+            raise ValueError(f"Not a directory: {directory}")
+
+        if not os.access(directory, os.R_OK | os.X_OK):
+            raise ValueError(f"Directory is not accessible: {directory}")
+
+    def get_absolute_path(self, path: str, base_directory: Optional[str] = None) -> str:
+        """
+        Get absolute path by joining base directory with path if path is relative.
+
+        Args:
+            path (str): The path to make absolute
+            base_directory (Optional[str]): Base directory to join with relative paths
+
+        Returns:
+            str: Absolute path
+        """
+        if os.path.isabs(path):
+            return path
+        if not base_directory:
+            return os.path.abspath(path)
+        return os.path.join(base_directory, path)

src/mcp_shell_server/io_redirection_handler.py

@@ -0,0 +1,153 @@
+"""IO redirection handling module for MCP Shell Server."""
+
+import asyncio
+import os
+from typing import IO, Any, Dict, List, Optional, Tuple, Union
+
+
+class IORedirectionHandler:
+    """Handles input/output redirection for shell commands."""
+
+    def validate_redirection_syntax(self, command: List[str]) -> None:
+        """
+        Validate the syntax of redirection operators in the command.
+
+        Args:
+            command (List[str]): Command and its arguments including redirections
+
+        Raises:
+            ValueError: If the redirection syntax is invalid
+        """
+        prev_token = None
+        for token in command:
+            if token in [">", ">>", "<"]:
+                if prev_token and prev_token in [">", ">>", "<"]:
+                    raise ValueError(
+                        "Invalid redirection syntax: consecutive operators"
+                    )
+            prev_token = token
+
+    def process_redirections(
+        self, command: List[str]
+    ) -> Tuple[List[str], Dict[str, Union[None, str, bool]]]:
+        """
+        Process input/output redirections in the command.
+
+        Args:
+            command (List[str]): Command and its arguments including redirections
+
+        Returns:
+            Tuple[List[str], Dict[str, Any]]: Processed command without redirections and
+                                           redirection configuration
+
+        Raises:
+            ValueError: If the redirection syntax is invalid
+        """
+        self.validate_redirection_syntax(command)
+
+        cmd = []
+        redirects: Dict[str, Union[None, str, bool]] = {
+            "stdin": None,
+            "stdout": None,
+            "stdout_append": False,
+        }
+
+        i = 0
+        while i < len(command):
+            token = command[i]
+
+            # 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]
+                if path in [">", ">>", "<"]:
+                    raise ValueError("Invalid redirection target: operator found")
+                redirects["stdin"] = path
+                i += 2
+                continue
+
+            cmd.append(token)
+            i += 1
+
+        return cmd, redirects
+
+    async def setup_redirects(
+        self,
+        redirects: Dict[str, Union[None, str, bool]],
+        directory: Optional[str] = None,
+    ) -> Dict[str, Union[IO[Any], int, str, None]]:
+        """
+        Set up file handles for redirections.
+
+        Args:
+            redirects (Dict[str, Union[None, str, bool]]): Redirection configuration
+            directory (Optional[str]): Working directory for file paths
+
+        Returns:
+            Dict[str, Union[IO[Any], int, str, None]]: File handles for subprocess
+        """
+        handles: Dict[str, Union[IO[Any], int, str, None]] = {}
+
+        # Handle input redirection
+        if redirects["stdin"]:
+            path = (
+                os.path.join(directory or "", str(redirects["stdin"]))
+                if directory and redirects["stdin"]
+                else str(redirects["stdin"])
+            )
+            try:
+                file = open(path, "r")
+                handles["stdin"] = asyncio.subprocess.PIPE
+                handles["stdin_data"] = file.read()
+                file.close()
+            except (FileNotFoundError, IOError) as e:
+                raise ValueError("Failed to open input file") from e
+
+        # Handle output redirection
+        if redirects["stdout"]:
+            path = (
+                os.path.join(directory or "", str(redirects["stdout"]))
+                if directory and redirects["stdout"]
+                else str(redirects["stdout"])
+            )
+            mode = "a" if redirects.get("stdout_append") else "w"
+            try:
+                handles["stdout"] = open(path, mode)
+            except (IOError, PermissionError) as e:
+                raise ValueError(f"Failed to open output file: {e}") from e
+        else:
+            handles["stdout"] = asyncio.subprocess.PIPE
+
+        handles["stderr"] = asyncio.subprocess.PIPE
+
+        return handles
+
+    async def cleanup_handles(
+        self, handles: Dict[str, Union[IO[Any], int, None]]
+    ) -> None:
+        """
+        Clean up file handles after command execution.
+
+        Args:
+            handles (Dict[str, Union[IO[Any], int, None]]): File handles to clean up
+        """
+        for key in ["stdout", "stderr"]:
+            handle = handles.get(key)
+            if handle and hasattr(handle, "close") and not isinstance(handle, int):
+                try:
+                    handle.close()
+                except (IOError, ValueError):
+                    pass

src/mcp_shell_server/process_manager.py

@@ -0,0 +1,314 @@
+"""Process management for shell command execution."""
+
+import asyncio
+import logging
+import os
+import signal
+from typing import IO, Any, Dict, List, Optional, Set, Tuple, Union
+from weakref import WeakSet
+
+
+class ProcessManager:
+    """Manages process creation, execution, and cleanup for shell commands."""
+
+    def __init__(self):
+        """Initialize ProcessManager with signal handling setup."""
+        self._processes: Set[asyncio.subprocess.Process] = WeakSet()
+        self._original_sigint_handler = None
+        self._original_sigterm_handler = None
+        self._setup_signal_handlers()
+
+    def _setup_signal_handlers(self) -> None:
+        """Set up signal handlers for graceful process management."""
+        if os.name != "posix":
+            return
+
+        def handle_termination(signum: int, _: Any) -> None:
+            """Handle termination signals by cleaning up processes."""
+            if self._processes:
+                for process in self._processes:
+                    try:
+                        if process.returncode is None:
+                            process.terminate()
+                    except Exception as e:
+                        logging.warning(
+                            f"Error terminating process on signal {signum}: {e}"
+                        )
+
+            # Restore original handler and re-raise signal
+            if signum == signal.SIGINT and self._original_sigint_handler:
+                signal.signal(signal.SIGINT, self._original_sigint_handler)
+            elif signum == signal.SIGTERM and self._original_sigterm_handler:
+                signal.signal(signal.SIGTERM, self._original_sigterm_handler)
+
+            # Re-raise signal
+            os.kill(os.getpid(), signum)
+
+        # Store original handlers
+        self._original_sigint_handler = signal.signal(signal.SIGINT, handle_termination)
+        self._original_sigterm_handler = signal.signal(
+            signal.SIGTERM, handle_termination
+        )
+
+    async def start_process_async(
+        self, cmd: List[str], timeout: Optional[int] = None
+    ) -> asyncio.subprocess.Process:
+        """Start a new process asynchronously.
+
+        Args:
+            cmd: Command to execute as list of strings
+            timeout: Optional timeout in seconds
+
+        Returns:
+            Process object
+        """
+        process = await self.create_process(
+            " ".join(cmd), directory=None, timeout=timeout
+        )
+        process.is_running = lambda self=process: self.returncode is None  # type: ignore
+        return process
+
+    async def start_process(
+        self, cmd: List[str], timeout: Optional[int] = None
+    ) -> asyncio.subprocess.Process:
+        """Start a new process asynchronously.
+
+        Args:
+            cmd: Command to execute as list of strings
+            timeout: Optional timeout in seconds
+
+        Returns:
+            Process object
+        """
+        process = await self.create_process(
+            " ".join(cmd), directory=None, timeout=timeout
+        )
+        process.is_running = lambda self=process: self.returncode is None  # type: ignore
+        return process
+
+    async def cleanup_processes(
+        self, processes: List[asyncio.subprocess.Process]
+    ) -> None:
+        """Clean up processes by killing them if they're still running.
+
+        Args:
+            processes: List of processes to clean up
+        """
+        cleanup_tasks = []
+        for process in processes:
+            if process.returncode is None:
+                try:
+                    # Force kill immediately as required by tests
+                    process.kill()
+                    cleanup_tasks.append(asyncio.create_task(process.wait()))
+                except Exception as e:
+                    logging.warning(f"Error killing process: {e}")
+
+        if cleanup_tasks:
+            try:
+                # Wait for all processes to be killed
+                await asyncio.wait(cleanup_tasks, timeout=5)
+            except asyncio.TimeoutError:
+                logging.error("Process cleanup timed out")
+            except Exception as e:
+                logging.error(f"Error during process cleanup: {e}")
+
+    async def cleanup_all(self) -> None:
+        """Clean up all tracked processes."""
+        if self._processes:
+            processes = list(self._processes)
+            await self.cleanup_processes(processes)
+            self._processes.clear()
+
+    async def create_process(
+        self,
+        shell_cmd: str,
+        directory: Optional[str],
+        stdin: Optional[str] = None,
+        stdout_handle: Any = asyncio.subprocess.PIPE,
+        envs: Optional[Dict[str, str]] = None,
+        timeout: Optional[int] = None,
+    ) -> asyncio.subprocess.Process:
+        """Create a new subprocess with the given parameters.
+
+        Args:
+            shell_cmd (str): Shell command to execute
+            directory (Optional[str]): Working directory
+            stdin (Optional[str]): Input to be passed to the process
+            stdout_handle: File handle or PIPE for stdout
+            envs (Optional[Dict[str, str]]): Additional environment variables
+            timeout (Optional[int]): Timeout in seconds
+
+        Returns:
+            asyncio.subprocess.Process: Created process
+
+        Raises:
+            ValueError: If process creation fails
+        """
+        try:
+            process = await asyncio.create_subprocess_shell(
+                shell_cmd,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=stdout_handle,
+                stderr=asyncio.subprocess.PIPE,
+                env={**os.environ, **(envs or {})},
+                cwd=directory,
+            )
+
+            # Add process to tracked set
+            self._processes.add(process)
+            return process
+
+        except OSError as e:
+            raise ValueError(f"Failed to create process: {str(e)}") from e
+        except Exception as e:
+            raise ValueError(
+                f"Unexpected error during process creation: {str(e)}"
+            ) from e
+
+    async def execute_with_timeout(
+        self,
+        process: asyncio.subprocess.Process,
+        stdin: Optional[str] = None,
+        timeout: Optional[int] = None,
+    ) -> Tuple[bytes, bytes]:
+        """Execute the process with timeout handling.
+
+        Args:
+            process: Process to execute
+            stdin (Optional[str]): Input to pass to the process
+            timeout (Optional[int]): Timeout in seconds
+
+        Returns:
+            Tuple[bytes, bytes]: Tuple of (stdout, stderr)
+
+        Raises:
+            asyncio.TimeoutError: If execution times out
+        """
+        stdin_bytes = stdin.encode() if stdin else None
+
+        async def _kill_process():
+            if process.returncode is not None:
+                return
+
+            try:
+                # Try graceful termination first
+                process.terminate()
+                for _ in range(5):  # Wait up to 0.5 seconds
+                    if process.returncode is not None:
+                        return
+                    await asyncio.sleep(0.1)
+
+                # Force kill if still running
+                if process.returncode is None:
+                    process.kill()
+                    await asyncio.wait_for(process.wait(), timeout=1.0)
+            except Exception as e:
+                logging.warning(f"Error killing process: {e}")
+
+        try:
+            if timeout:
+                try:
+                    return await asyncio.wait_for(
+                        process.communicate(input=stdin_bytes), timeout=timeout
+                    )
+                except asyncio.TimeoutError:
+                    await _kill_process()
+                    raise
+            return await process.communicate(input=stdin_bytes)
+        except Exception as e:
+            await _kill_process()
+            raise e
+
+    async def execute_pipeline(
+        self,
+        commands: List[str],
+        first_stdin: Optional[bytes] = None,
+        last_stdout: Union[IO[Any], int, None] = None,
+        directory: Optional[str] = None,
+        timeout: Optional[int] = None,
+        envs: Optional[Dict[str, str]] = None,
+    ) -> Tuple[bytes, bytes, int]:
+        """Execute a pipeline of commands.
+
+        Args:
+            commands: List of shell commands to execute in pipeline
+            first_stdin: Input to pass to the first command
+            last_stdout: Output handle for the last command
+            directory: Working directory
+            timeout: Timeout in seconds
+            envs: Additional environment variables
+
+        Returns:
+            Tuple[bytes, bytes, int]: Tuple of (stdout, stderr, return_code)
+
+        Raises:
+            ValueError: If no commands provided or command execution fails
+        """
+        if not commands:
+            raise ValueError("No commands provided")
+
+        processes: List[asyncio.subprocess.Process] = []
+        try:
+            prev_stdout: Optional[bytes] = first_stdin
+            final_stderr: bytes = b""
+            final_stdout: bytes = b""
+
+            for i, cmd in enumerate(commands):
+                process = await self.create_process(
+                    cmd,
+                    directory,
+                    stdout_handle=(
+                        asyncio.subprocess.PIPE
+                        if i < len(commands) - 1 or not last_stdout
+                        else last_stdout
+                    ),
+                    envs=envs,
+                )
+                if not hasattr(process, "is_running"):
+                    process.is_running = lambda self=process: self.returncode is None  # type: ignore
+                processes.append(process)
+
+                try:
+                    stdout, stderr = await self.execute_with_timeout(
+                        process,
+                        stdin=prev_stdout.decode() if prev_stdout else None,
+                        timeout=timeout,
+                    )
+
+                    final_stderr += stderr if stderr else b""
+                    if process.returncode != 0:
+                        error_msg = stderr.decode("utf-8", errors="replace").strip()
+                        if not error_msg:
+                            error_msg = (
+                                f"Command failed with exit code {process.returncode}"
+                            )
+                        raise ValueError(error_msg)
+
+                    if i == len(commands) - 1:
+                        if last_stdout and isinstance(last_stdout, IO):
+                            last_stdout.write(stdout.decode("utf-8", errors="replace"))
+                        else:
+                            final_stdout = stdout if stdout else b""
+                    else:
+                        prev_stdout = stdout if stdout else b""
+
+                except asyncio.TimeoutError:
+                    process.kill()
+                    raise
+                except Exception:
+                    process.kill()
+                    raise
+
+            return (
+                final_stdout,
+                final_stderr,
+                (
+                    processes[-1].returncode
+                    if processes and processes[-1].returncode is not None
+                    else 1
+                ),
+            )
+
+        finally:
+            await self.cleanup_processes(processes)

src/mcp_shell_server/server.py

@@ -28,7 +28,7 @@ def __init__(self):
 
     def get_allowed_commands(self) -> list[str]:
         """Get the allowed commands"""
-        return self.executor.get_allowed_commands()
+        return self.executor.validator.get_allowed_commands()
 
     def get_tool_description(self) -> Tool:
         """Get the tool description for the execute command"""

src/mcp_shell_server/shell_executor.py

@@ -1,12 +1,98 @@
-import os
-
-
-# Configure pytest-asyncio
-def pytest_configure(config):
-    """Configure pytest-asyncio defaults"""
-    # Enable command execution for tests
-    os.environ["ALLOW_COMMANDS"] = "1"
-    # Add allowed commands for tests
-    os.environ["ALLOWED_COMMANDS"] = (
-        "echo,sleep,cat,ls,pwd,touch,mkdir,rm,mv,cp,grep,awk,sed"
-    )
+"""
+Test configuration and fixtures.
+"""
+
+import asyncio
+from typing import IO
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+import pytest_asyncio
+
+from mcp_shell_server.shell_executor import ShellExecutor
+
+
+@pytest.fixture
+def mock_file(mocker):
+    """Provide a mock file object."""
+    mock = mocker.MagicMock(spec=IO)
+    mock.close = mocker.MagicMock()
+    return mock
+
+
+@pytest_asyncio.fixture
+async def mock_process_manager():
+    """Provide a mock process manager."""
+    manager = MagicMock()
+
+    # Mock process object
+    process = AsyncMock()
+    process.returncode = 0
+
+    # Mock manager methods
+    manager.create_process = AsyncMock()
+
+    async def create_process_side_effect(*args, **kwargs):
+        process = AsyncMock()
+        process.returncode = 0
+        process.communicate = AsyncMock(return_value=(b"", b""))
+        process.kill = AsyncMock()
+        process.wait = AsyncMock()
+        return process
+
+    manager.create_process.side_effect = create_process_side_effect
+
+    manager.execute_with_timeout = AsyncMock()
+    manager.execute_pipeline = AsyncMock()
+    manager.cleanup_processes = AsyncMock()
+
+    # Set empty default return values - tests should override these as needed
+    manager.execute_with_timeout.return_value = (b"", b"")
+    manager.execute_pipeline.return_value = (b"", b"", 0)
+
+    return manager
+
+
+@pytest_asyncio.fixture
+async def shell_executor_with_mock(mock_process_manager):
+    """Provide a shell executor with mock process manager."""
+    executor = ShellExecutor(process_manager=mock_process_manager)
+    return executor
+
+
+@pytest.fixture
+def temp_test_dir(tmpdir):
+    """Provide a temporary test directory."""
+    return str(tmpdir)
+
+
+@pytest_asyncio.fixture(scope="function")
+async def event_loop():
+    """Create and provide a new event loop for each module."""
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    yield loop
+
+    # Clean up the event loop
+    try:
+        # Close all tasks
+        tasks = asyncio.all_tasks(loop)
+        if tasks:
+            # Cancel all tasks and wait for their completion
+            for task in tasks:
+                task.cancel()
+            loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))
+
+        # Clean up all transports
+        if hasattr(loop, "_transports"):
+            for transport in list(loop._transports.values()):
+                if hasattr(transport, "close"):
+                    transport.close()
+
+        # Cleanup
+        loop.stop()
+        asyncio.set_event_loop(None)
+        await loop.shutdown_asyncgens()
+        loop.close()
+    except Exception as e:
+        print(f"Error during loop cleanup: {e}")

tests/conftest.py

@@ -0,0 +1,29 @@
+"""
+Test configuration and fixtures.
+"""
+
+import asyncio
+import os
+
+import pytest_asyncio
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def cleanup_loop():
+    """Run after each test to ensure proper event loop cleanup."""
+    yield
+    if hasattr(asyncio, "current_task") and asyncio.current_task() is not None:
+        try:
+            await asyncio.sleep(0)  # Allow pending callbacks to complete
+        except Exception:
+            pass
+
+
+def pytest_configure(config):
+    """Configure pytest-asyncio defaults"""
+    # Enable command execution for tests
+    os.environ["ALLOW_COMMANDS"] = "1"
+    # Add allowed commands for tests
+    os.environ["ALLOWED_COMMANDS"] = (
+        "echo,sleep,cat,ls,pwd,touch,mkdir,rm,mv,cp,grep,awk,sed"
+    )

tests/conftest_new.py

@@ -0,0 +1,74 @@
+"""Test cases for the CommandValidator class."""
+
+import pytest
+
+from mcp_shell_server.command_validator import CommandValidator
+
+
+def clear_env(monkeypatch):
+    monkeypatch.delenv("ALLOW_COMMANDS", raising=False)
+    monkeypatch.delenv("ALLOWED_COMMANDS", raising=False)
+
+
+@pytest.fixture
+def validator():
+    return CommandValidator()
+
+
+def test_get_allowed_commands(validator, monkeypatch):
+    clear_env(monkeypatch)
+    monkeypatch.setenv("ALLOW_COMMANDS", "cmd1,cmd2")
+    monkeypatch.setenv("ALLOWED_COMMANDS", "cmd3,cmd4")
+    assert set(validator.get_allowed_commands()) == {"cmd1", "cmd2", "cmd3", "cmd4"}
+
+
+def test_is_command_allowed(validator, monkeypatch):
+    clear_env(monkeypatch)
+    monkeypatch.setenv("ALLOW_COMMANDS", "allowed_cmd")
+    assert validator.is_command_allowed("allowed_cmd")
+    assert not validator.is_command_allowed("disallowed_cmd")
+
+
+def test_validate_no_shell_operators(validator):
+    validator.validate_no_shell_operators("echo")  # Should not raise
+    with pytest.raises(ValueError, match="Unexpected shell operator"):
+        validator.validate_no_shell_operators(";")
+    with pytest.raises(ValueError, match="Unexpected shell operator"):
+        validator.validate_no_shell_operators("&&")
+
+
+def test_validate_pipeline(validator, monkeypatch):
+    clear_env(monkeypatch)
+    monkeypatch.setenv("ALLOW_COMMANDS", "ls,grep")
+
+    # Valid pipeline
+    validator.validate_pipeline(["ls", "|", "grep", "test"])
+
+    # Empty command before pipe
+    with pytest.raises(ValueError, match="Empty command before pipe operator"):
+        validator.validate_pipeline(["|", "grep", "test"])
+
+    # Command not allowed
+    with pytest.raises(ValueError, match="Command not allowed"):
+        validator.validate_pipeline(["invalid_cmd", "|", "grep", "test"])
+
+
+def test_validate_command(validator, monkeypatch):
+    clear_env(monkeypatch)
+
+    # No allowed commands
+    with pytest.raises(ValueError, match="No commands are allowed"):
+        validator.validate_command(["cmd"])
+
+    monkeypatch.setenv("ALLOW_COMMANDS", "allowed_cmd")
+
+    # Empty command
+    with pytest.raises(ValueError, match="Empty command"):
+        validator.validate_command([])
+
+    # Command not allowed
+    with pytest.raises(ValueError, match="Command not allowed"):
+        validator.validate_command(["disallowed_cmd"])
+
+    # Command allowed
+    validator.validate_command(["allowed_cmd", "-arg"])  # Should not raise

tests/test_command_validator.py

@@ -0,0 +1,56 @@
+"""Tests for directory_manager module."""
+
+import os
+
+import pytest
+
+from mcp_shell_server.directory_manager import DirectoryManager
+
+
+def test_validate_directory(tmp_path):
+    """Test directory validation."""
+    manager = DirectoryManager()
+    test_dir = str(tmp_path)
+
+    # Valid directory
+    manager.validate_directory(test_dir)
+
+    # None directory
+    with pytest.raises(ValueError, match="Directory is required"):
+        manager.validate_directory(None)
+
+    # Relative path
+    with pytest.raises(ValueError, match="Directory must be an absolute path"):
+        manager.validate_directory("relative/path")
+
+    # Non-existent directory
+    nonexistent = os.path.join(test_dir, "nonexistent")
+    with pytest.raises(ValueError, match="Directory does not exist"):
+        manager.validate_directory(nonexistent)
+
+    # Not a directory (create a file)
+    test_file = os.path.join(test_dir, "test.txt")
+    with open(test_file, "w") as f:
+        f.write("test")
+    with pytest.raises(ValueError, match="Not a directory"):
+        manager.validate_directory(test_file)
+
+
+def test_get_absolute_path(tmp_path):
+    """Test absolute path resolution."""
+    manager = DirectoryManager()
+    test_dir = str(tmp_path)
+
+    # Already absolute path
+    abs_path = os.path.join(test_dir, "test")
+    assert manager.get_absolute_path(abs_path) == abs_path
+
+    # Relative path without base directory
+    rel_path = "test/path"
+    expected = os.path.abspath(rel_path)
+    assert manager.get_absolute_path(rel_path) == expected
+
+    # Relative path with base directory
+    rel_path = "test/path"
+    expected = os.path.join(test_dir, rel_path)
+    assert manager.get_absolute_path(rel_path, test_dir) == expected

tests/test_directory_manager.py

@@ -0,0 +1,193 @@
+"""Tests for the ProcessManager class."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from mcp_shell_server.process_manager import ProcessManager
+
+
+def create_mock_process():
+    """Create a mock process with all required attributes."""
+    process = MagicMock()
+    process.returncode = 0
+    process.communicate = AsyncMock(return_value=(b"output", b"error"))
+    process.wait = AsyncMock(return_value=0)
+    process.terminate = MagicMock()
+    process.kill = MagicMock()
+    return process
+
+
+@pytest.fixture
+def process_manager():
+    """Fixture for ProcessManager instance."""
+    return ProcessManager()
+
+
+@pytest.mark.asyncio
+async def test_create_process(process_manager):
+    """Test creating a process with basic parameters."""
+    mock_proc = create_mock_process()
+    with patch(
+        "mcp_shell_server.process_manager.asyncio.create_subprocess_shell",
+        new_callable=AsyncMock,
+        return_value=mock_proc,
+    ) as mock_create:
+        process = await process_manager.create_process(
+            "echo 'test'",
+            directory="/tmp",
+            stdin="input",
+        )
+
+        assert process == mock_proc
+        assert process == mock_proc
+        mock_create.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_execute_with_timeout_success(process_manager):
+    """Test executing a process with successful completion."""
+    mock_proc = create_mock_process()
+    mock_proc.returncode = 0
+    mock_proc.communicate.return_value = (b"output", b"error")
+
+    stdout, stderr = await process_manager.execute_with_timeout(
+        mock_proc,
+        stdin="input",
+        timeout=10,
+    )
+
+    assert stdout == b"output"
+    assert stderr == b"error"
+    mock_proc.communicate.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_execute_with_timeout_timeout(process_manager):
+    """Test executing a process that times out."""
+    mock_proc = create_mock_process()
+    exc = asyncio.TimeoutError("Process timed out")
+    mock_proc.communicate.side_effect = exc
+    mock_proc.returncode = None  # プロセスがまだ実行中の状態をシミュレート
+
+    # プロセスの終了状態をシミュレート
+    async def set_returncode():
+        mock_proc.returncode = -15  # SIGTERM
+
+    mock_proc.wait.side_effect = set_returncode
+
+    with pytest.raises(TimeoutError):
+        await process_manager.execute_with_timeout(
+            mock_proc,
+            timeout=1,
+        )
+
+    mock_proc.terminate.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_execute_pipeline_success(process_manager):
+    """Test executing a pipeline of commands successfully."""
+    mock_proc1 = create_mock_process()
+    mock_proc1.communicate.return_value = (b"output1", b"")
+    mock_proc1.returncode = 0
+
+    mock_proc2 = create_mock_process()
+    mock_proc2.communicate.return_value = (b"final output", b"")
+    mock_proc2.returncode = 0
+
+    with patch(
+        "mcp_shell_server.process_manager.asyncio.create_subprocess_shell",
+        new_callable=AsyncMock,
+        side_effect=[mock_proc1, mock_proc2],
+    ) as mock_create:
+        stdout, stderr, return_code = await process_manager.execute_pipeline(
+            ["echo 'test'", "grep test"],
+            directory="/tmp",
+            timeout=10,
+        )
+
+        assert stdout == b"final output"
+        assert stderr == b""
+        assert return_code == 0
+        assert mock_create.call_count == 2  # Verify subprocess creation calls
+
+        # Verify the command arguments for each subprocess call
+        calls = mock_create.call_args_list
+        assert "echo" in calls[0].args[0]
+        assert "grep" in calls[1].args[0]
+
+
+@pytest.mark.asyncio
+async def test_execute_pipeline_with_error(process_manager):
+    """Test executing a pipeline where a command fails."""
+    mock_proc = create_mock_process()
+    mock_proc.communicate.return_value = (b"", b"error message")
+    mock_proc.returncode = 1
+
+    create_process_mock = AsyncMock(return_value=mock_proc)
+
+    with patch.object(process_manager, "create_process", create_process_mock):
+        with pytest.raises(ValueError, match="error message"):
+            await process_manager.execute_pipeline(
+                ["invalid_command"],
+                directory="/tmp",
+            )
+
+
+@pytest.mark.asyncio
+async def test_cleanup_processes(process_manager):
+    """Test cleaning up processes."""
+    # Create mock processes with different states
+    running_proc = create_mock_process()
+    running_proc.returncode = None
+
+    completed_proc = create_mock_process()
+    completed_proc.returncode = 0
+
+    # Execute cleanup
+    await process_manager.cleanup_processes([running_proc, completed_proc])
+
+    # Verify running process was killed and waited for
+    running_proc.kill.assert_called_once()
+    running_proc.wait.assert_awaited_once()
+
+    # Verify completed process was not killed or waited for
+    completed_proc.kill.assert_not_called()
+    completed_proc.wait.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_create_process_with_error(process_manager):
+    """Test creating a process that fails to start."""
+    with patch(
+        "asyncio.create_subprocess_shell",
+        new_callable=AsyncMock,
+        side_effect=OSError("Failed to create process"),
+    ):
+        with pytest.raises(ValueError, match="Failed to create process"):
+            await process_manager.create_process("invalid command", directory="/tmp")
+
+
+@pytest.mark.asyncio
+async def test_execute_pipeline_empty_commands(process_manager):
+    """Test executing a pipeline with no commands."""
+    with pytest.raises(ValueError, match="No commands provided"):
+        await process_manager.execute_pipeline([], directory="/tmp")
+
+
+@pytest.mark.asyncio
+async def test_execute_pipeline_timeout(process_manager):
+    """Test executing a pipeline that times out."""
+    mock_proc = create_mock_process()
+    mock_proc.communicate.side_effect = TimeoutError("Process timed out")
+
+    with patch.object(process_manager, "create_process", return_value=mock_proc):
+        with pytest.raises(TimeoutError, match="Process timed out"):
+            await process_manager.execute_pipeline(
+                ["sleep 10"],
+                directory="/tmp",
+                timeout=1,
+            )
+            mock_proc.kill.assert_called_once()

tests/test_io_redirection_handler.py

@@ -0,0 +1,149 @@
+# tests/test_process_manager_macos.py
+import asyncio
+import platform
+import subprocess
+
+import pytest
+
+pytestmark = [
+    pytest.mark.skipif(
+        platform.system() != "Darwin", reason="These tests only run on macOS"
+    ),
+    pytest.mark.macos,
+    pytest.mark.slow,
+]
+
+
+@pytest.fixture
+def process_manager():
+    from mcp_shell_server.process_manager import ProcessManager
+
+    pm = ProcessManager()
+    try:
+        yield pm
+    finally:
+        asyncio.run(pm.cleanup_all())
+
+
+def get_process_status(pid: int) -> str:
+    """Get process status using ps command."""
+    try:
+        ps = subprocess.run(
+            ["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True
+        )
+        return ps.stdout.strip()
+    except subprocess.CalledProcessError:
+        return ""
+
+
+@pytest.mark.asyncio
+async def test_zombie_process_cleanup(process_manager):
+    """Test that background processes don't become zombies."""
+    cmd = ["sh", "-c", "sleep 0.5 & wait"]
+    process = await process_manager.start_process(cmd)
+
+    # Wait for the background process to finish
+    await asyncio.sleep(1)
+
+    # Get process status
+    status = get_process_status(process.pid)
+
+    # Verify process is either gone or not zombie (Z state)
+    assert "Z" not in status, f"Process {process.pid} is zombie (status: {status})"
+
+
+@pytest.mark.asyncio
+async def test_process_timeout(process_manager):
+    """Test process timeout functionality."""
+    # Start a process that should timeout
+    cmd = ["sleep", "10"]
+    process = await process_manager.start_process(cmd)
+
+    try:
+        # Communicate with timeout
+        with pytest.raises(TimeoutError):
+            _, _ = await process_manager.execute_with_timeout(process, timeout=1)
+
+        # プロセスが終了するまで待つ
+        try:
+            await asyncio.wait_for(process.wait(), timeout=1.0)
+        except asyncio.TimeoutError:
+            process.kill()  # Force kill
+
+        # Wait for termination
+        await asyncio.wait_for(process.wait(), timeout=0.5)
+
+        # Verify process was terminated
+        assert process.returncode is not None
+        assert not process.is_running()
+    finally:
+        if process.returncode is None:
+            try:
+                process.kill()
+                await asyncio.wait_for(process.wait(), timeout=0.5)
+            except (ProcessLookupError, asyncio.TimeoutError):
+                pass
+
+
+@pytest.mark.asyncio
+async def test_multiple_process_cleanup(process_manager):
+    """Test cleanup of multiple processes."""
+    # Start multiple background processes
+    # Start multiple processes in parallel
+    processes = await asyncio.gather(
+        *[process_manager.start_process(["sleep", "2"]) for _ in range(3)]
+    )
+
+    # Give them a moment to start
+    await asyncio.sleep(0.1)
+
+    try:
+        # Verify they're all running
+        assert all(p.is_running() for p in processes)
+
+        # Cleanup
+        await process_manager.cleanup_all()
+
+        # Give cleanup a moment to complete
+        await asyncio.sleep(0.1)
+
+        # Verify all processes are gone
+        for p in processes:
+            status = get_process_status(p.pid)
+            assert status == "", f"Process {p.pid} still exists with status: {status}"
+    finally:
+        # Ensure cleanup in case of test failure
+        for p in processes:
+            if p.returncode is None:
+                try:
+                    p.kill()
+                except ProcessLookupError:
+                    pass
+
+
+@pytest.mark.asyncio
+async def test_process_group_termination(process_manager):
+    """Test that entire process group is terminated."""
+    # Create a process that spawns children
+    cmd = ["sh", "-c", "sleep 10 & sleep 10 & sleep 10 & wait"]
+    process = await process_manager.start_process(cmd)
+
+    try:
+        # Give processes time to start
+        await asyncio.sleep(0.5)
+
+        # Kill the main process
+        process.kill()
+
+        # Wait a moment for cleanup
+        await asyncio.sleep(0.5)
+
+        # Check if any processes from the group remain
+        ps = subprocess.run(["pgrep", "-g", str(process.pid)], capture_output=True)
+        assert ps.returncode != 0, "Process group still exists"
+    finally:
+        if process.returncode is None:
+            try:
+                process.kill()
+            except ProcessLookupError:
+                pass

tests/test_process_manager.py

@@ -44,7 +44,14 @@ def setup_mock_subprocess(monkeypatch):
     """Set up mock subprocess to avoid interactive shell warnings"""
 
     async def mock_create_subprocess_shell(
-        cmd, stdin=None, stdout=None, stderr=None, env=None, cwd=None
+        cmd,
+        stdin=None,
+        stdout=None,
+        stderr=None,
+        env=None,
+        cwd=None,
+        preexec_fn=None,
+        start_new_session=None,
     ):
         # Return appropriate output based on command
         if "echo" in cmd:
@@ -83,8 +90,9 @@ async def test_list_tools():
 
 
 @pytest.mark.asyncio
-async def test_tool_execution_timeout():
+async def test_tool_execution_timeout(monkeypatch):
     """Test tool execution with timeout"""
+    monkeypatch.setenv("ALLOW_COMMANDS", "sleep")
     with pytest.raises(RuntimeError, match="Command execution timed out"):
         await call_tool(
             "shell_execute",

tests/test_process_manager_macos.py

@@ -42,15 +42,15 @@ async def test_no_allowed_commands_validation(monkeypatch):
         ValueError,
         match="No commands are allowed. Please set ALLOW_COMMANDS environment variable.",
     ):
-        executor._validate_command(["any_command"])
+        executor.validator.validate_command(["any_command"])
 
 
 @pytest.mark.asyncio
 async def test_shell_operator_validation():
     """Test validation of shell operators"""
     executor = ShellExecutor()
 
-    operators = [";" "&&", "||", "|"]
+    operators = [";", "&&", "||", "|"]
     for op in operators:
         # Test shell operator validation
         with pytest.raises(ValueError, match=f"Unexpected shell operator: {op}"):

tests/test_server.py

@@ -20,21 +20,21 @@ async def test_redirection_validation():
 
     # Missing path for output redirection
     with pytest.raises(ValueError, match="Missing path for output redirection"):
-        executor._process_redirections(["echo", "test", ">"])
+        executor.io_handler.process_redirections(["echo", "test", ">"])
 
     # Invalid redirection target (operator)
     with pytest.raises(
         ValueError, match="Invalid redirection syntax: consecutive operators"
     ):
-        executor._process_redirections(["echo", "test", ">", ">"])
+        executor.io_handler.process_redirections(["echo", "test", ">", ">"])
 
     # Missing path for input redirection
     with pytest.raises(ValueError, match="Missing path for input redirection"):
-        executor._process_redirections(["cat", "<"])
+        executor.io_handler.process_redirections(["cat", "<"])
 
     # Missing path for output redirection after input redirection
     with pytest.raises(ValueError, match="Missing path for output redirection"):
-        executor._process_redirections(["cat", "<", "input.txt", ">"])
+        executor.io_handler.process_redirections(["cat", "<", "input.txt", ">"])
 
 
 @pytest.mark.asyncio
@@ -45,25 +45,31 @@ async def test_directory_validation(monkeypatch):
 
     # Directory validation is performed in the _validate_directory method
     with pytest.raises(ValueError, match="Directory is required"):
-        executor._validate_directory(None)
+        executor.directory_manager.validate_directory(None)
 
     # Directory is not absolute path
     with pytest.raises(ValueError, match="Directory must be an absolute path"):
-        executor._validate_directory("relative/path")
+        executor.directory_manager.validate_directory("relative/path")
 
     # Directory does not exist
     with pytest.raises(ValueError, match="Directory does not exist"):
-        executor._validate_directory("/path/does/not/exist")
+        executor.directory_manager.validate_directory("/path/does/not/exist")
 
 
 @pytest.mark.asyncio
-async def test_process_timeout(monkeypatch, temp_test_dir):
+async def test_process_timeout(
+    shell_executor_with_mock, temp_test_dir, mock_process_manager, monkeypatch
+):
     """Test process timeout handling"""
     monkeypatch.setenv("ALLOW_COMMANDS", "sleep")
-    executor = ShellExecutor()
-
-    # Process timeout with sleep command
-    command = ["sleep", "5"]
-    result = await executor.execute(command=command, directory=temp_test_dir, timeout=1)
+    # Mock timeout behavior
+    mock_process_manager.execute_with_timeout.side_effect = TimeoutError(
+        "Command timed out after 1 seconds"
+    )
+
+    # Process timeout test
+    result = await shell_executor_with_mock.execute(
+        command=["sleep", "5"], directory=temp_test_dir, timeout=1
+    )
     assert result["error"] == "Command timed out after 1 seconds"
     assert result["status"] == -1

tests/test_shell_executor.py

@@ -1,5 +1,6 @@
 import os
 import tempfile
+from unittest.mock import AsyncMock
 
 import pytest
 
@@ -23,11 +24,14 @@ def executor():
 async def test_basic_pipe_command(executor, temp_test_dir, monkeypatch):
     """Test basic pipe functionality with allowed commands"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo,grep")
+    mock_process_manager = AsyncMock()
+    mock_process_manager.execute_pipeline.return_value = (b"world\n", b"", 0)
+    executor.process_manager = mock_process_manager
     result = await executor.execute(
         ["echo", "hello world", "|", "grep", "world"], temp_test_dir
     )
     assert result["status"] == 0
-    assert result["stdout"].strip() == "hello world"
+    assert result["stdout"].strip() == "world"
 
 
 @pytest.mark.asyncio

tests/test_shell_executor.py.bak

@@ -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,49 +52,52 @@ 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"]
 
 
 @pytest.mark.asyncio
-async def test_pipeline_execution_success(executor, temp_test_dir, monkeypatch):
+async def test_pipeline_execution_success(
+    shell_executor_with_mock, temp_test_dir, mock_process_manager, monkeypatch
+):
     """Test successful pipeline execution with proper return value"""
-    clear_env(monkeypatch)
-    monkeypatch.setenv("ALLOWED_COMMANDS", "echo,grep")
+    monkeypatch.setenv("ALLOW_COMMANDS", "echo,grep")
+    # Set up mock for pipeline execution
+    expected_output = b"mocked pipeline output\n"
+    mock_process_manager.execute_pipeline.return_value = (expected_output, b"", 0)
 
-    result = await executor.execute(
+    result = await shell_executor_with_mock.execute(
         ["echo", "hello world", "|", "grep", "world"],
         directory=temp_test_dir,
         timeout=5,
     )
 
     assert result["error"] is None
     assert result["status"] == 0
-    assert "world" in result["stdout"]
+    assert result["stdout"].rstrip() == "mocked pipeline output"
     assert "execution_time" in result
 
 
 @pytest.mark.asyncio
-async def test_pipeline_cleanup_and_timeouts(executor, temp_test_dir, monkeypatch):
+async def test_pipeline_cleanup_and_timeouts(
+    shell_executor_with_mock, temp_test_dir, mock_process_manager, monkeypatch
+):
     """Test cleanup of processes in pipelines and timeout handling"""
-    clear_env(monkeypatch)
-    monkeypatch.setenv("ALLOW_COMMANDS", "cat,tr,head,sleep")
-
-    # Test pipeline with early termination
-    test_file = os.path.join(temp_test_dir, "test.txt")
-    with open(test_file, "w") as f:
-        f.write("test\n" * 1000)
+    monkeypatch.setenv("ALLOW_COMMANDS", "echo,grep")
+    # Mock timeout behavior for pipeline
+    mock_process_manager.execute_pipeline.side_effect = TimeoutError(
+        "Command timed out after 1 seconds"
+    )
 
-    result = await executor.execute(
-        ["cat", test_file, "|", "tr", "[:lower:]", "[:upper:]", "|", "head", "-n", "1"],
+    result = await shell_executor_with_mock.execute(
+        ["echo", "test", "|", "grep", "test"],  # Use a pipeline command
         temp_test_dir,
-        timeout=2,
+        timeout=1,
     )
-    assert result["status"] == 0
-    assert result["stdout"].strip() == "TEST"
 
-    # Test timeout handling in pipeline
-    result = await executor.execute(["sleep", "5"], temp_test_dir, timeout=1)
     assert result["status"] == -1
-    assert "timed out" in result["error"].lower()  # タイムアウトエラーの確認
+    assert "timed out" in result["error"].lower()
+
+    # Verify cleanup was called
+    mock_process_manager.cleanup_processes.assert_called_once()

tests/test_shell_executor_error_cases.py

@@ -1,132 +1,155 @@
-import os
-import tempfile
-from io import TextIOWrapper
+"""Tests for IO redirection in shell executor with mocked file operations."""
+
+from unittest.mock import MagicMock, patch
 
 import pytest
 
-from mcp_shell_server.shell_executor import ShellExecutor
+from mcp_shell_server.io_redirection_handler import IORedirectionHandler
 
 
 @pytest.fixture
-def temp_test_dir():
-    """Create a temporary directory for testing"""
-    with tempfile.TemporaryDirectory() as tmpdirname:
-        yield os.path.realpath(tmpdirname)
+def mock_file():
+    """Create a mock file object."""
+    file_mock = MagicMock()
+    file_mock.closed = False
+    file_mock.close = MagicMock()
+    file_mock.write = MagicMock()
+    file_mock.read = MagicMock(return_value="test content")
+    return file_mock
 
 
-@pytest.mark.asyncio
-async def test_redirection_setup(temp_test_dir):
-    """Test setup of redirections with files"""
-    executor = ShellExecutor()
-
-    # Create a test input file
-    with open(os.path.join(temp_test_dir, "input.txt"), "w") as f:
-        f.write("test content")
-
-    # Test input redirection setup
-    redirects = {
-        "stdin": "input.txt",
-        "stdout": None,
-        "stdout_append": False,
-    }
-    handles = await executor._setup_redirects(redirects, temp_test_dir)
-    assert "stdin" in handles
-    assert "stdin_data" in handles
-    assert handles["stdin_data"] == "test content"
-    assert isinstance(handles["stdout"], int)
-    assert isinstance(handles["stderr"], int)
-
-    # Test output redirection setup
-    output_file = os.path.join(temp_test_dir, "output.txt")
-    redirects = {
-        "stdin": None,
-        "stdout": output_file,
-        "stdout_append": False,
-    }
-    handles = await executor._setup_redirects(redirects, temp_test_dir)
-    assert isinstance(handles["stdout"], TextIOWrapper)
-    assert not handles["stdout"].closed
-    await executor._cleanup_handles(handles)
-    try:
-        assert handles["stdout"].closed
-    except ValueError:
-        # Ignore errors from already closed file
-        pass
+@pytest.fixture
+def handler():
+    """Create a new IORedirectionHandler instance for each test."""
+    return IORedirectionHandler()
 
 
 @pytest.mark.asyncio
-async def test_redirection_append_mode(temp_test_dir):
-    """Test output redirection in append mode"""
-    executor = ShellExecutor()
-
-    output_file = os.path.join(temp_test_dir, "output.txt")
+async def test_file_input_redirection(handler, mock_file):
+    """Test input redirection from a file using mocks."""
+    with (
+        patch("builtins.open", return_value=mock_file),
+        patch("os.path.exists", return_value=True),
+    ):
 
-    # Test append mode
-    redirects = {
-        "stdin": None,
-        "stdout": output_file,
-        "stdout_append": True,
-    }
-    handles = await executor._setup_redirects(redirects, temp_test_dir)
-    assert handles["stdout"].mode == "a"
-    await executor._cleanup_handles(handles)
+        redirects = {
+            "stdin": "input.txt",
+            "stdout": None,
+            "stdout_append": False,
+        }
+        handles = await handler.setup_redirects(redirects, "/mock/dir")
 
-    # Test write mode
-    redirects["stdout_append"] = False
-    handles = await executor._setup_redirects(redirects, temp_test_dir)
-    assert handles["stdout"].mode == "w"
-    await executor._cleanup_handles(handles)
+        assert "stdin" in handles
+        assert "stdin_data" in handles
+        assert handles["stdin_data"] == "test content"
+        assert isinstance(handles["stdout"], int)
+        assert isinstance(handles["stderr"], int)
 
 
 @pytest.mark.asyncio
-async def test_redirection_setup_errors(temp_test_dir):
-    """Test error cases in redirection setup"""
-    executor = ShellExecutor()
-
-    # Test non-existent input file
-    redirects = {
-        "stdin": "nonexistent.txt",
-        "stdout": None,
-        "stdout_append": False,
-    }
-    with pytest.raises(ValueError, match="Failed to open input file"):
-        await executor._setup_redirects(redirects, temp_test_dir)
-
-    # Test error in output file creation
-    os.chmod(temp_test_dir, 0o444)  # Make directory read-only
-    try:
+async def test_file_output_redirection(handler, mock_file):
+    """Test output redirection to a file using mocks."""
+    with patch("builtins.open", return_value=mock_file):
         redirects = {
             "stdin": None,
             "stdout": "output.txt",
             "stdout_append": False,
         }
-        with pytest.raises(ValueError, match="Failed to open output file"):
-            await executor._setup_redirects(redirects, temp_test_dir)
-    finally:
-        os.chmod(temp_test_dir, 0o755)  # Reset permissions
+        handles = await handler.setup_redirects(redirects, "/mock/dir")
 
+        assert "stdout" in handles
+        assert not handles["stdout"].closed
+        await handler.cleanup_handles(handles)
+        mock_file.close.assert_called_once()
 
-@pytest.mark.asyncio
-async def test_invalid_redirection_paths():
-    """Test invalid redirection path scenarios"""
-    executor = ShellExecutor()
 
-    # Test missing path for output redirection
-    with pytest.raises(ValueError, match="Missing path for output redirection"):
-        executor._parse_command(["echo", "test", ">"])
+@pytest.mark.asyncio
+async def test_append_mode(handler, mock_file):
+    """Test output redirection in append mode using mocks."""
+    with patch("builtins.open", return_value=mock_file):
+        # Test append mode
+        redirects = {
+            "stdin": None,
+            "stdout": "output.txt",
+            "stdout_append": True,
+        }
+        mock_file.mode = "a"  # Set the expected mode
+        handles = await handler.setup_redirects(redirects, "/mock/dir")
+        assert handles["stdout"].mode == "a"
+        await handler.cleanup_handles(handles)
+        mock_file.close.assert_called_once()
+
+        # Reset mock and test write mode
+        mock_file.reset_mock()
+        mock_file.mode = "w"  # Set the expected mode for write mode
+        redirects["stdout_append"] = False
+        handles = await handler.setup_redirects(redirects, "/mock/dir")
+        assert handles["stdout"].mode == "w"
+        await handler.cleanup_handles(handles)
+        mock_file.close.assert_called_once()
+
+
+def test_validate_redirection_syntax(handler):
+    """Test validation of redirection syntax."""
+    # Valid cases
+    handler.validate_redirection_syntax(["echo", "hello", ">", "output.txt"])
+    handler.validate_redirection_syntax(["cat", "<", "input.txt", ">", "output.txt"])
+    handler.validate_redirection_syntax(["echo", "hello", ">>", "output.txt"])
+
+    # Invalid cases
+    with pytest.raises(ValueError, match="consecutive operators"):
+        handler.validate_redirection_syntax(["echo", ">", ">", "output.txt"])
+
+    with pytest.raises(ValueError, match="consecutive operators"):
+        handler.validate_redirection_syntax(["cat", "<", ">", "output.txt"])
+
+
+def test_process_redirections(handler):
+    """Test processing of redirection operators."""
+    # Input redirection
+    cmd, redirects = handler.process_redirections(["cat", "<", "input.txt"])
+    assert cmd == ["cat"]
+    assert redirects["stdin"] == "input.txt"
+    assert redirects["stdout"] is None
+
+    # Output redirection
+    cmd, redirects = handler.process_redirections(["echo", "test", ">", "output.txt"])
+    assert cmd == ["echo", "test"]
+    assert redirects["stdout"] == "output.txt"
+    assert not redirects["stdout_append"]
+
+    # Combined redirections
+    cmd, redirects = handler.process_redirections(
+        ["cat", "<", "in.txt", ">", "out.txt"]
+    )
+    assert cmd == ["cat"]
+    assert redirects["stdin"] == "in.txt"
+    assert redirects["stdout"] == "out.txt"
 
-    # Test invalid redirection target (operator found)
-    with pytest.raises(ValueError, match="Invalid redirection target: operator found"):
-        executor._parse_command(["echo", "test", ">", ">"])
 
-    # Test missing path for input redirection
-    with pytest.raises(ValueError, match="Missing path for input redirection"):
-        executor._parse_command(["cat", "<"])
+@pytest.mark.asyncio
+async def test_setup_errors(handler, mock_file):
+    """Test error cases in redirection setup using mocks."""
+    # Test non-existent input file
+    with patch("os.path.exists", return_value=False):
+        redirects = {
+            "stdin": "nonexistent.txt",
+            "stdout": None,
+            "stdout_append": False,
+        }
+        with pytest.raises(ValueError, match="Failed to open input file"):
+            await handler.setup_redirects(redirects, "/mock/dir")
 
-    # Test missing path for output redirection
-    with pytest.raises(ValueError, match="Missing path for output redirection"):
-        executor._parse_command(["echo", "test", ">"])
+    # Test error in output file creation
+    # Mock builtins.open to raise PermissionError
+    mock_open = MagicMock(side_effect=PermissionError("Permission denied"))
+    with patch("builtins.open", mock_open):
+        redirects = {
+            "stdin": None,
+            "stdout": "output.txt",
+            "stdout_append": False,
+        }
+        with pytest.raises(ValueError, match="Failed to open output file"):
+            await handler.setup_redirects(redirects, "/mock/dir")
 
-    # Test invalid redirection target: operator found for output
-    with pytest.raises(ValueError, match="Invalid redirection target: operator found"):
-        executor._parse_command(["echo", "test", ">", ">"])
+    mock_file.close.assert_not_called()

tests/test_shell_executor_new_tests.py

@@ -0,0 +1,132 @@
+import os
+import tempfile
+from io import TextIOWrapper
+
+import pytest
+
+from mcp_shell_server.shell_executor import ShellExecutor
+
+
+@pytest.fixture
+def temp_test_dir():
+    """Create a temporary directory for testing"""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        yield os.path.realpath(tmpdirname)
+
+
+@pytest.mark.asyncio
+async def test_redirection_setup(temp_test_dir):
+    """Test setup of redirections with files"""
+    executor = ShellExecutor()
+
+    # Create a test input file
+    with open(os.path.join(temp_test_dir, "input.txt"), "w") as f:
+        f.write("test content")
+
+    # Test input redirection setup
+    redirects = {
+        "stdin": "input.txt",
+        "stdout": None,
+        "stdout_append": False,
+    }
+    handles = await executor.io_handler.setup_redirects(redirects, temp_test_dir)
+    assert "stdin" in handles
+    assert "stdin_data" in handles
+    assert handles["stdin_data"] == "test content"
+    assert isinstance(handles["stdout"], int)
+    assert isinstance(handles["stderr"], int)
+
+    # Test output redirection setup
+    output_file = os.path.join(temp_test_dir, "output.txt")
+    redirects = {
+        "stdin": None,
+        "stdout": output_file,
+        "stdout_append": False,
+    }
+    handles = await executor.io_handler.setup_redirects(redirects, temp_test_dir)
+    assert isinstance(handles["stdout"], TextIOWrapper)
+    assert not handles["stdout"].closed
+    await executor.io_handler.cleanup_handles(handles)
+    try:
+        assert handles["stdout"].closed
+    except ValueError:
+        # Ignore errors from already closed file
+        pass
+
+
+@pytest.mark.asyncio
+async def test_redirection_append_mode(temp_test_dir):
+    """Test output redirection in append mode"""
+    executor = ShellExecutor()
+
+    output_file = os.path.join(temp_test_dir, "output.txt")
+
+    # Test append mode
+    redirects = {
+        "stdin": None,
+        "stdout": output_file,
+        "stdout_append": True,
+    }
+    handles = await executor.io_handler.setup_redirects(redirects, temp_test_dir)
+    assert handles["stdout"].mode == "a"
+    await executor.io_handler.cleanup_handles(handles)
+
+    # Test write mode
+    redirects["stdout_append"] = False
+    handles = await executor.io_handler.setup_redirects(redirects, temp_test_dir)
+    assert handles["stdout"].mode == "w"
+    await executor.io_handler.cleanup_handles(handles)
+
+
+@pytest.mark.asyncio
+async def test_redirection_setup_errors(temp_test_dir):
+    """Test error cases in redirection setup"""
+    executor = ShellExecutor()
+
+    # Test non-existent input file
+    redirects = {
+        "stdin": "nonexistent.txt",
+        "stdout": None,
+        "stdout_append": False,
+    }
+    with pytest.raises(ValueError, match="Failed to open input file"):
+        await executor.io_handler.setup_redirects(redirects, temp_test_dir)
+
+    # Test error in output file creation
+    os.chmod(temp_test_dir, 0o444)  # Make directory read-only
+    try:
+        redirects = {
+            "stdin": None,
+            "stdout": "output.txt",
+            "stdout_append": False,
+        }
+        with pytest.raises(ValueError, match="Failed to open output file"):
+            await executor.io_handler.setup_redirects(redirects, temp_test_dir)
+    finally:
+        os.chmod(temp_test_dir, 0o755)  # Reset permissions
+
+
+@pytest.mark.asyncio
+async def test_invalid_redirection_paths():
+    """Test invalid redirection path scenarios"""
+    executor = ShellExecutor()
+
+    # Test missing path for output redirection
+    with pytest.raises(ValueError, match="Missing path for output redirection"):
+        executor.io_handler.process_redirections(["echo", "test", ">"])
+
+    # Test invalid redirection target (operator found)
+    with pytest.raises(ValueError, match="Invalid redirection target: operator found"):
+        executor.io_handler.process_redirections(["echo", "test", ">", ">"])
+
+    # Test missing path for input redirection
+    with pytest.raises(ValueError, match="Missing path for input redirection"):
+        executor.io_handler.process_redirections(["cat", "<"])
+
+    # Test missing path for output redirection
+    with pytest.raises(ValueError, match="Missing path for output redirection"):
+        executor.io_handler.process_redirections(["echo", "test", ">"])
+
+    # Test invalid redirection target: operator found for output
+    with pytest.raises(ValueError, match="Invalid redirection target: operator found"):
+        executor.io_handler.process_redirections(["echo", "test", ">", ">"])