|
| 1 | +import asyncio |
| 2 | +import os |
| 3 | +import tempfile |
| 4 | + |
1 | 5 | import pytest
|
2 | 6 |
|
3 | 7 | from mcp_shell_server.shell_executor import ShellExecutor
|
@@ -30,3 +34,43 @@ async def test_process_redirections_invalid_output(executor):
|
30 | 34 | with pytest.raises(ValueError) as exc:
|
31 | 35 | executor._process_redirections(["echo", "hello", ">"])
|
32 | 36 | assert str(exc.value) == "Missing path for output redirection"
|
| 37 | + |
| 38 | + |
| 39 | +@pytest.mark.asyncio |
| 40 | +async def test_setup_redirects_stdin(executor): |
| 41 | + """Test _setup_redirects method with stdin redirection""" |
| 42 | + # Create a temporary file with test content |
| 43 | + with tempfile.NamedTemporaryFile(mode="w", delete=False) as f: |
| 44 | + f.write("test content\n") |
| 45 | + input_file = f.name |
| 46 | + |
| 47 | + try: |
| 48 | + redirects = {"stdin": input_file, "stdout": None, "stdout_append": False} |
| 49 | + handles = await executor._setup_redirects(redirects) |
| 50 | + |
| 51 | + assert handles["stdin"] == asyncio.subprocess.PIPE |
| 52 | + assert "stdin_data" in handles |
| 53 | + assert handles["stdin_data"] == "test content\n" |
| 54 | + assert handles["stdout"] == asyncio.subprocess.PIPE |
| 55 | + assert handles["stderr"] == asyncio.subprocess.PIPE |
| 56 | + |
| 57 | + await executor._cleanup_handles(handles) |
| 58 | + finally: |
| 59 | + os.unlink(input_file) |
| 60 | + |
| 61 | + |
| 62 | +@pytest.mark.asyncio |
| 63 | +async def test_setup_redirects_stdout_error(executor): |
| 64 | + """Test _setup_redirects with output file open error""" |
| 65 | + # Create a read-only directory |
| 66 | + with tempfile.TemporaryDirectory() as tmpdirname: |
| 67 | + output_file = os.path.join(tmpdirname, "out.txt") |
| 68 | + os.chmod(tmpdirname, 0o555) # Read and execute only |
| 69 | + |
| 70 | + try: |
| 71 | + redirects = {"stdin": None, "stdout": output_file, "stdout_append": False} |
| 72 | + with pytest.raises(ValueError) as exc: |
| 73 | + await executor._setup_redirects(redirects) |
| 74 | + assert "Failed to open output file" in str(exc.value) |
| 75 | + finally: |
| 76 | + os.chmod(tmpdirname, 0o755) # Restore permissions |
0 commit comments