|
| 1 | +import pytest |
| 2 | + |
| 3 | +from mcp_shell_server.shell_executor import ShellExecutor |
| 4 | + |
| 5 | + |
| 6 | +@pytest.mark.asyncio |
| 7 | +async def test_pipeline_split(): |
| 8 | + """Test pipeline command splitting functionality""" |
| 9 | + executor = ShellExecutor() |
| 10 | + |
| 11 | + # Test basic pipe command |
| 12 | + commands = executor._split_pipe_commands(["echo", "hello", "|", "grep", "h"]) |
| 13 | + assert len(commands) == 2 |
| 14 | + assert commands[0] == ["echo", "hello"] |
| 15 | + assert commands[1] == ["grep", "h"] |
| 16 | + |
| 17 | + # Test empty pipe sections |
| 18 | + commands = executor._split_pipe_commands(["|", "grep", "pattern"]) |
| 19 | + assert len(commands) == 2 |
| 20 | + assert commands[0] == [] |
| 21 | + assert commands[1] == ["grep", "pattern"] |
| 22 | + |
| 23 | + # Test multiple pipes |
| 24 | + commands = executor._split_pipe_commands( |
| 25 | + ["cat", "file.txt", "|", "grep", "pattern", "|", "wc", "-l"] |
| 26 | + ) |
| 27 | + assert len(commands) == 3 |
| 28 | + assert commands[0] == ["cat", "file.txt"] |
| 29 | + assert commands[1] == ["grep", "pattern"] |
| 30 | + assert commands[2] == ["wc", "-l"] |
| 31 | + |
| 32 | + # Test trailing pipe |
| 33 | + commands = executor._split_pipe_commands(["echo", "hello", "|"]) |
| 34 | + assert len(commands) == 2 |
| 35 | + assert commands[0] == ["echo", "hello"] |
| 36 | + assert commands[1] == [] |
0 commit comments