|
| 1 | +import asyncio |
| 2 | + |
| 3 | +import pytest |
| 4 | + |
| 5 | +from mcp_shell_server.shell_executor import ShellExecutor |
| 6 | + |
| 7 | + |
| 8 | +@pytest.mark.asyncio |
| 9 | +async def test_empty_command_validation(): |
| 10 | + """Test validation of empty commands""" |
| 11 | + executor = ShellExecutor() |
| 12 | + |
| 13 | + # 空のコマンドのテスト |
| 14 | + with pytest.raises(ValueError, match="Empty command"): |
| 15 | + executor._validate_command([]) |
| 16 | + |
| 17 | + |
| 18 | +@pytest.mark.asyncio |
| 19 | +async def test_no_allowed_commands_validation(monkeypatch): |
| 20 | + """Test validation when no commands are allowed""" |
| 21 | + # ALLOW_COMMANDSを削除 |
| 22 | + monkeypatch.delenv("ALLOW_COMMANDS", raising=False) |
| 23 | + monkeypatch.delenv("ALLOWED_COMMANDS", raising=False) |
| 24 | + |
| 25 | + executor = ShellExecutor() |
| 26 | + with pytest.raises( |
| 27 | + ValueError, |
| 28 | + match="No commands are allowed. Please set ALLOW_COMMANDS environment variable.", |
| 29 | + ): |
| 30 | + executor._validate_command(["any_command"]) |
| 31 | + |
| 32 | + |
| 33 | +@pytest.mark.asyncio |
| 34 | +async def test_shell_operator_validation(): |
| 35 | + """Test validation of shell operators""" |
| 36 | + executor = ShellExecutor() |
| 37 | + |
| 38 | + operators = [";" "&&", "||", "|"] |
| 39 | + for op in operators: |
| 40 | + # シェル操作子の検証 |
| 41 | + with pytest.raises(ValueError, match=f"Unexpected shell operator: {op}"): |
| 42 | + executor._validate_no_shell_operators(op) |
| 43 | + |
| 44 | + |
| 45 | +@pytest.mark.asyncio |
| 46 | +async def test_process_execution_timeout(monkeypatch): |
| 47 | + """Test process execution timeout handling""" |
| 48 | + monkeypatch.setenv("ALLOW_COMMANDS", "sleep") |
| 49 | + executor = ShellExecutor() |
| 50 | + |
| 51 | + # プロセスのタイムアウトをテスト |
| 52 | + command = ["sleep", "5"] |
| 53 | + with pytest.raises(asyncio.TimeoutError): |
| 54 | + await asyncio.wait_for(executor.execute(command), timeout=0.1) |
| 55 | + |
| 56 | + |
| 57 | +@pytest.mark.asyncio |
| 58 | +async def test_process_failure(monkeypatch): |
| 59 | + """Test handling of process execution failure""" |
| 60 | + monkeypatch.setenv("ALLOW_COMMANDS", "false") |
| 61 | + executor = ShellExecutor() |
| 62 | + |
| 63 | + # falseコマンドは常に終了コード1を返す |
| 64 | + result = await executor.execute(["false"]) |
| 65 | + assert result["status"] == 1 |
0 commit comments