test: improve test coverage by fixing pipeline timeout tests

Author: tumf

tests/test_shell_executor.py

@@ -610,11 +610,7 @@ async def test_command_cleanup_on_error(executor, temp_test_dir, monkeypatch):
 
     async def execute_with_keyboard_interrupt():
         # Simulate keyboard interrupt during execution
-        result = await executor.execute(
-            ["sleep", "5"],
-            temp_test_dir,
-            timeout=1
-        )
+        result = await executor.execute(["sleep", "5"], temp_test_dir, timeout=1)
         return result
 
     result = await execute_with_keyboard_interrupt()

tests/test_shell_executor_pipeline.py

@@ -1,13 +1,33 @@
+"""Test pipeline execution and cleanup scenarios."""
+import os
+import tempfile
+
 import pytest
 
 from mcp_shell_server.shell_executor import ShellExecutor
 
 
+def clear_env(monkeypatch):
+    monkeypatch.delenv("ALLOW_COMMANDS", raising=False)
+    monkeypatch.delenv("ALLOWED_COMMANDS", raising=False)
+
+
+@pytest.fixture
+def executor():
+    return ShellExecutor()
+
+
+@pytest.fixture
+def temp_test_dir():
+    """Create a temporary directory for testing"""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        # Return the real path to handle macOS /private/tmp symlink
+        yield os.path.realpath(tmpdirname)
+
+
 @pytest.mark.asyncio
-async def test_pipeline_split():
+async def test_pipeline_split(executor):
     """Test pipeline command splitting functionality"""
-    executor = ShellExecutor()
-
     # Test basic pipe command
     commands = executor._split_pipe_commands(["echo", "hello", "|", "grep", "h"])
     assert len(commands) == 2
@@ -35,19 +55,45 @@ async def test_pipeline_split():
 
 
 @pytest.mark.asyncio
-async def test_pipeline_execution_success():
+async def test_pipeline_execution_success(executor, temp_test_dir, monkeypatch):
     """Test successful pipeline execution with proper return value"""
-    executor = ShellExecutor()
-    import os
-
-    os.environ["ALLOWED_COMMANDS"] = "echo,grep"
+    clear_env(monkeypatch)
+    monkeypatch.setenv("ALLOWED_COMMANDS", "echo,grep")
 
     result = await executor.execute(
-        ["echo", "hello world", "|", "grep", "world"], directory="/tmp", timeout=5
+        ["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 "execution_time" in result
-    assert result["directory"] == "/tmp"
+
+
+@pytest.mark.asyncio
+async def test_pipeline_cleanup_and_timeouts(executor, temp_test_dir, 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)
+
+    result = await executor.execute(
+        ["cat", test_file, "|", "tr", "[:lower:]", "[:upper:]", "|", "head", "-n", "1"],
+        temp_test_dir,
+        timeout=2
+    )
+    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()  # タイムアウトエラーの確認