andrewyng--aisuite
Rohit C Prasad
41b1591f6d
Mechanical 'black .' over root package and platform/ (109 files; the Lint workflow checks the whole repo and has been red since platform landed unformatted, with a few more violations from community merges whose branches never ran CI). No functional change: full platform suite (264) and aisuite suites verified unchanged after formatting.
95 行
2.9 KiB
Python
95 行
2.9 KiB
Python
"""P3 gate tests — persistent shell executor.
|
|
|
|
The executor drives the OS-native shell (bash on POSIX, PowerShell on Windows), so the
|
|
command strings here are parameterized per-OS. The behavior under test (cwd/env persistence,
|
|
exit codes, timeout-and-recover, truncation) is identical across both.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from coworker.permissions import PermissionEngine
|
|
from coworker.tools import ToolRegistry
|
|
from coworker.tools.shell import LocalExecutor, shell_tools
|
|
|
|
_WIN = sys.platform == "win32"
|
|
|
|
# Per-OS command snippets exercising the same behavior in the native shell.
|
|
SET_ENV = "$env:GREETING='hello_world'" if _WIN else "export GREETING=hello_world"
|
|
ECHO_ENV = "echo $env:GREETING" if _WIN else "echo $GREETING"
|
|
EXIT_OK = "cmd /c exit 0" if _WIN else "true"
|
|
EXIT_FAIL = "cmd /c exit 1" if _WIN else "false"
|
|
SLEEP_5 = "Start-Sleep -Seconds 5" if _WIN else "sleep 5"
|
|
PRINT_1000 = (
|
|
'foreach ($i in 1..1000) { "line$i" }'
|
|
if _WIN
|
|
else "for i in $(seq 1 1000); do echo line$i; done"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def executor(tmp_path):
|
|
ex = LocalExecutor(cwd=tmp_path, default_timeout=10)
|
|
yield ex
|
|
ex.close()
|
|
|
|
|
|
def test_cwd_persists_across_calls(executor, tmp_path):
|
|
(tmp_path / "sub").mkdir()
|
|
executor.run("cd sub")
|
|
result = executor.run("pwd")
|
|
assert result["exit_code"] == 0
|
|
assert "sub" in result["output"]
|
|
assert executor.cwd.endswith("sub")
|
|
|
|
|
|
def test_env_persists_across_calls(executor):
|
|
executor.run(SET_ENV)
|
|
result = executor.run(ECHO_ENV)
|
|
assert "hello_world" in result["output"]
|
|
|
|
|
|
def test_exit_code_captured(executor):
|
|
assert executor.run(EXIT_OK)["exit_code"] == 0
|
|
assert executor.run(EXIT_FAIL)["exit_code"] == 1
|
|
|
|
|
|
def test_timeout_kills_command(executor):
|
|
start = time.monotonic()
|
|
result = executor.run(SLEEP_5, timeout=1)
|
|
elapsed = time.monotonic() - start
|
|
assert result["timed_out"] is True
|
|
assert elapsed < 4.0 # did not block for the full sleep
|
|
# session survives the timeout — still usable (POSIX keeps the shell; Windows respawns)
|
|
assert executor.run("echo alive")["output"].strip().endswith("alive")
|
|
|
|
|
|
def test_large_output_truncated(tmp_path):
|
|
ex = LocalExecutor(cwd=tmp_path, max_output_chars=200, default_timeout=10)
|
|
try:
|
|
result = ex.run(PRINT_1000)
|
|
assert result["truncated"] is True
|
|
assert len(result["output"]) <= 200
|
|
finally:
|
|
ex.close()
|
|
|
|
|
|
def test_shell_tool_integration(executor, tmp_path):
|
|
reg = ToolRegistry()
|
|
reg.register_all(shell_tools(executor))
|
|
assert "run_shell" in reg.names()
|
|
|
|
spec = reg.get("run_shell")
|
|
assert spec.metadata.requires_approval is True
|
|
|
|
eng = PermissionEngine(workspace_root=tmp_path)
|
|
decision = eng.evaluate("run_shell", {"command": "echo hi"}, spec.metadata)
|
|
assert not decision.allowed and decision.needs_user # high-risk → asks
|
|
|
|
out = reg.execute("run_shell", {"command": "echo hi"})
|
|
assert "hi" in out["output"]
|