forked from HAL9000/cleveragents-core
183 lines
6.9 KiB
Python
183 lines
6.9 KiB
Python
"""Step definitions for tui_shell_exec_coverage.feature.
|
|
|
|
These steps target specific uncovered lines in tui/input/shell_exec.py:
|
|
- Lines 48-49: empty command returns ShellResult with exit_code=2
|
|
- Lines 52-56: CLEVERAGENTS_DISABLE_SHELL_MODE env var disables shell mode
|
|
- Line 61: confirm_dangerous callback is invoked for dangerous commands
|
|
- Lines 77-82: subprocess.TimeoutExpired is caught and returns exit_code=124
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.tui.input.shell_exec import (
|
|
ShellResult,
|
|
looks_dangerous,
|
|
run_shell_command,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
@given("the shell_exec module is imported")
|
|
def step_shell_exec_imported(context):
|
|
"""Verify the module is importable and key symbols exist."""
|
|
assert run_shell_command is not None
|
|
assert ShellResult is not None
|
|
assert looks_dangerous is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Empty / whitespace command (lines 48-49)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I run a shell command with an empty string")
|
|
def step_run_empty_command(context):
|
|
"""Run run_shell_command with an empty string."""
|
|
context.shell_result = run_shell_command("")
|
|
|
|
|
|
@when('I run a shell command with only whitespace "{whitespace}"')
|
|
def step_run_whitespace_command(context, whitespace):
|
|
"""Run run_shell_command with whitespace-only input."""
|
|
context.shell_result = run_shell_command(whitespace)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shell mode disabled via env var (lines 52-56)
|
|
# ---------------------------------------------------------------------------
|
|
@given('the environment variable CLEVERAGENTS_DISABLE_SHELL_MODE is set to "{value}"')
|
|
def step_set_disable_env_var(context, value):
|
|
"""Set the CLEVERAGENTS_DISABLE_SHELL_MODE environment variable."""
|
|
context.original_env_value = os.environ.get("CLEVERAGENTS_DISABLE_SHELL_MODE")
|
|
os.environ["CLEVERAGENTS_DISABLE_SHELL_MODE"] = value
|
|
|
|
def restore_env():
|
|
if context.original_env_value is None:
|
|
os.environ.pop("CLEVERAGENTS_DISABLE_SHELL_MODE", None)
|
|
else:
|
|
os.environ["CLEVERAGENTS_DISABLE_SHELL_MODE"] = context.original_env_value
|
|
|
|
context.add_cleanup(restore_env)
|
|
|
|
|
|
@when('I run a shell command "{command}"')
|
|
def step_run_shell_command(context, command):
|
|
"""Run run_shell_command with the given command string."""
|
|
context.shell_result = run_shell_command(command)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dangerous command with confirm callback (line 61)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a confirm_dangerous callback that returns True")
|
|
def step_confirm_callback_true(context):
|
|
"""Create a confirm_dangerous callback that approves dangerous commands."""
|
|
context.confirm_called = False
|
|
|
|
def confirm_true(cmd):
|
|
context.confirm_called = True
|
|
return True
|
|
|
|
context.confirm_callback = confirm_true
|
|
|
|
|
|
@given("a confirm_dangerous callback that returns False")
|
|
def step_confirm_callback_false(context):
|
|
"""Create a confirm_dangerous callback that rejects dangerous commands."""
|
|
context.confirm_called = False
|
|
|
|
def confirm_false(cmd):
|
|
context.confirm_called = True
|
|
return False
|
|
|
|
context.confirm_callback = confirm_false
|
|
|
|
|
|
@when('I run a dangerous command "{command}" with the callback')
|
|
def step_run_dangerous_with_callback(context, command):
|
|
"""Run a dangerous command with the confirm_dangerous callback.
|
|
|
|
We patch subprocess.run to avoid actually executing dangerous commands
|
|
when the callback returns True.
|
|
"""
|
|
fake_proc = subprocess.CompletedProcess(
|
|
args=command, returncode=0, stdout="mocked output", stderr=""
|
|
)
|
|
with patch(
|
|
"cleveragents.tui.input.shell_exec.subprocess.run", return_value=fake_proc
|
|
):
|
|
context.shell_result = run_shell_command(
|
|
command, confirm_dangerous=context.confirm_callback
|
|
)
|
|
|
|
|
|
@then("the dangerous command should have been confirmed")
|
|
def step_verify_confirm_called(context):
|
|
"""Verify that the confirm_dangerous callback was invoked."""
|
|
assert context.confirm_called is True, "confirm_dangerous callback was not called"
|
|
|
|
|
|
@then("the shell result should reflect the executed command")
|
|
def step_verify_executed_result(context):
|
|
"""Verify the result reflects a successfully executed (mocked) command."""
|
|
assert context.shell_result.exit_code == 0
|
|
assert context.shell_result.stdout == "mocked output"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Timeout (lines 77-82)
|
|
# ---------------------------------------------------------------------------
|
|
@given("subprocess run is mocked to raise TimeoutExpired")
|
|
def step_mock_timeout(context):
|
|
"""Prepare a mock that raises subprocess.TimeoutExpired."""
|
|
context.timeout_mock_active = True
|
|
|
|
|
|
@when('I run a shell command "{command}" with a {seconds:d} second timeout')
|
|
def step_run_with_timeout(context, command, seconds):
|
|
"""Run a command with a short timeout, mocking TimeoutExpired."""
|
|
with patch(
|
|
"cleveragents.tui.input.shell_exec.subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(cmd=command, timeout=seconds),
|
|
):
|
|
context.shell_result = run_shell_command(command, timeout_seconds=seconds)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Common assertions
|
|
# ---------------------------------------------------------------------------
|
|
@then("the shell result exit code should be {code:d}")
|
|
def step_verify_exit_code(context, code):
|
|
"""Verify the exit code of the shell result."""
|
|
assert context.shell_result.exit_code == code, (
|
|
f"Expected exit_code={code}, got {context.shell_result.exit_code}"
|
|
)
|
|
|
|
|
|
@then('the shell result stderr should be "{expected}"')
|
|
def step_verify_stderr(context, expected):
|
|
"""Verify the stderr of the shell result matches exactly."""
|
|
assert context.shell_result.stderr == expected, (
|
|
f"Expected stderr={expected!r}, got {context.shell_result.stderr!r}"
|
|
)
|
|
|
|
|
|
@then("the shell result stdout should be empty")
|
|
def step_verify_stdout_empty(context):
|
|
"""Verify the stdout of the shell result is empty."""
|
|
assert context.shell_result.stdout == "", (
|
|
f"Expected stdout='', got {context.shell_result.stdout!r}"
|
|
)
|
|
|
|
|
|
@then('the shell result stderr should contain "{fragment}"')
|
|
def step_verify_stderr_contains(context, fragment):
|
|
"""Verify the stderr of the shell result contains the given fragment."""
|
|
assert fragment in context.shell_result.stderr, (
|
|
f"Expected stderr to contain {fragment!r}, got {context.shell_result.stderr!r}"
|
|
)
|