Skip to content

Commit c0539e4

Browse files
committed
test: Add redirection append test and coverage pragmas
1 parent 519db08 commit c0539e4

File tree

3 files changed

+32
-8
lines changed

3 files changed

+32
-8
lines changed

src/mcp_shell_server/shell_executor.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ def __init__(self):
1515
"""
1616
Initialize the executor.
1717
"""
18-
pass
18+
pass # pragma: no cover
1919

2020
def _get_allowed_commands(self) -> set[str]:
2121
"""Get the set of allowed commands from environment variables"""

tests/test_shell_executor.py

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -617,3 +617,28 @@ async def execute_with_keyboard_interrupt():
617617
assert result["error"] == "Command timed out after 1 seconds"
618618
assert result["status"] == -1
619619
assert "execution_time" in result
620+
621+
622+
@pytest.mark.asyncio
623+
async def test_output_redirection_with_append(executor, temp_test_dir, monkeypatch):
624+
"""Test output redirection with append mode"""
625+
clear_env(monkeypatch)
626+
monkeypatch.setenv("ALLOW_COMMANDS", "echo,cat")
627+
output_file = os.path.join(temp_test_dir, "test.txt")
628+
629+
# Write initial content
630+
await executor.execute(["echo", "hello", ">", output_file], directory=temp_test_dir)
631+
632+
# Append content
633+
result = await executor.execute(
634+
["echo", "world", ">>", output_file], directory=temp_test_dir
635+
)
636+
assert result["error"] is None
637+
assert result["status"] == 0
638+
639+
# Verify contents
640+
result = await executor.execute(["cat", output_file], directory=temp_test_dir)
641+
lines = result["stdout"].strip().split("\n")
642+
assert len(lines) == 2
643+
assert lines[0] == "hello"
644+
assert lines[1] == "world"

tests/test_shell_executor_pipeline.py

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
"""Test pipeline execution and cleanup scenarios."""
2+
23
import os
34
import tempfile
45

@@ -61,7 +62,9 @@ async def test_pipeline_execution_success(executor, temp_test_dir, monkeypatch):
6162
monkeypatch.setenv("ALLOWED_COMMANDS", "echo,grep")
6263

6364
result = await executor.execute(
64-
["echo", "hello world", "|", "grep", "world"], directory=temp_test_dir, timeout=5
65+
["echo", "hello world", "|", "grep", "world"],
66+
directory=temp_test_dir,
67+
timeout=5,
6568
)
6669

6770
assert result["error"] is None
@@ -84,16 +87,12 @@ async def test_pipeline_cleanup_and_timeouts(executor, temp_test_dir, monkeypatc
8487
result = await executor.execute(
8588
["cat", test_file, "|", "tr", "[:lower:]", "[:upper:]", "|", "head", "-n", "1"],
8689
temp_test_dir,
87-
timeout=2
90+
timeout=2,
8891
)
8992
assert result["status"] == 0
9093
assert result["stdout"].strip() == "TEST"
9194

9295
# Test timeout handling in pipeline
93-
result = await executor.execute(
94-
["sleep", "5"],
95-
temp_test_dir,
96-
timeout=1
97-
)
96+
result = await executor.execute(["sleep", "5"], temp_test_dir, timeout=1)
9897
assert result["status"] == -1
9998
assert "timed out" in result["error"].lower() # タイムアウトエラーの確認

0 commit comments

Comments
 (0)