REFACTOR handle user defined test cmd (#59)
This commit is contained in:
Jose M Leon
GitHub
parent
d89f8990a3
commit
90b8875a73
|
|
@ -1,6 +1,7 @@
|
|||
"""Utilities for executing and managing user-defined test commands."""
|
||||
|
||||
from typing import Dict, Any, Tuple, Optional
|
||||
from typing import Dict, Any, Tuple
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from rich.console import Console
|
||||
from rich.markdown import Markdown
|
||||
|
|
@ -20,105 +21,146 @@ class TestState:
|
|||
auto_test: bool
|
||||
should_break: bool = False
|
||||
|
||||
def display_test_failure(attempts: int, max_retries: int) -> None:
|
||||
"""Display test failure message.
|
||||
class TestCommandExecutor:
|
||||
"""Class for executing and managing test commands."""
|
||||
|
||||
Args:
|
||||
attempts: Current number of attempts
|
||||
max_retries: Maximum allowed retries
|
||||
"""
|
||||
console.print(
|
||||
Panel(
|
||||
Markdown(f"Test failed. Attempt number {attempts} of {max_retries}. Retrying and informing of failure output"),
|
||||
title="🔎 User Defined Test",
|
||||
border_style="red bold"
|
||||
def __init__(self, config: Dict[str, Any], original_prompt: str, test_attempts: int = 0, auto_test: bool = False):
|
||||
"""Initialize the test command executor.
|
||||
|
||||
Args:
|
||||
config: Configuration dictionary containing test settings
|
||||
original_prompt: The original prompt to append errors to
|
||||
test_attempts: Current number of test attempts
|
||||
auto_test: Whether auto-test mode is enabled
|
||||
"""
|
||||
self.config = config
|
||||
self.state = TestState(
|
||||
prompt=original_prompt,
|
||||
test_attempts=test_attempts,
|
||||
auto_test=auto_test,
|
||||
should_break=False
|
||||
)
|
||||
)
|
||||
self.max_retries = config.get("max_test_cmd_retries", 5)
|
||||
|
||||
def handle_test_failure(state: TestState, original_prompt: str, test_result: Dict[str, Any]) -> TestState:
|
||||
"""Handle test command failure.
|
||||
|
||||
Args:
|
||||
state: Current test state
|
||||
original_prompt: Original prompt text
|
||||
test_result: Test command result
|
||||
|
||||
Returns:
|
||||
Updated test state
|
||||
"""
|
||||
state.prompt = f"{original_prompt}. Previous attempt failed with: <test_cmd_stdout>{test_result['output']}</test_cmd_stdout>"
|
||||
display_test_failure(state.test_attempts, 5) # Default max retries
|
||||
state.should_break = False
|
||||
return state
|
||||
def display_test_failure(self) -> None:
|
||||
"""Display test failure message."""
|
||||
console.print(
|
||||
Panel(
|
||||
Markdown(f"Test failed. Attempt number {self.state.test_attempts} of {self.max_retries}. Retrying and informing of failure output"),
|
||||
title="🔎 User Defined Test",
|
||||
border_style="red bold"
|
||||
)
|
||||
)
|
||||
|
||||
def run_test_command(cmd: str, state: TestState, original_prompt: str) -> TestState:
|
||||
"""Run test command and handle result.
|
||||
|
||||
Args:
|
||||
cmd: Test command to execute
|
||||
state: Current test state
|
||||
original_prompt: Original prompt text
|
||||
def handle_test_failure(self, original_prompt: str, test_result: Dict[str, Any]) -> None:
|
||||
"""Handle test command failure.
|
||||
|
||||
Returns:
|
||||
Updated test state
|
||||
"""
|
||||
try:
|
||||
test_result = run_shell_command(cmd)
|
||||
state.test_attempts += 1
|
||||
Args:
|
||||
original_prompt: Original prompt text
|
||||
test_result: Test command result
|
||||
"""
|
||||
self.state.prompt = f"{original_prompt}. Previous attempt failed with: <test_cmd_stdout>{test_result['output']}</test_cmd_stdout>"
|
||||
self.display_test_failure()
|
||||
self.state.should_break = False
|
||||
|
||||
def run_test_command(self, cmd: str, original_prompt: str) -> None:
|
||||
"""Run test command and handle result.
|
||||
|
||||
if not test_result["success"]:
|
||||
return handle_test_failure(state, original_prompt, test_result)
|
||||
Args:
|
||||
cmd: Test command to execute
|
||||
original_prompt: Original prompt text
|
||||
"""
|
||||
timeout = self.config.get('timeout', 30)
|
||||
try:
|
||||
logger.info(f"Executing test command: {cmd} with timeout {timeout}s")
|
||||
test_result = run_shell_command(cmd, timeout=timeout)
|
||||
self.state.test_attempts += 1
|
||||
|
||||
state.should_break = True
|
||||
return state
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Test command execution failed: {str(e)}")
|
||||
state.test_attempts += 1
|
||||
state.should_break = True
|
||||
return state
|
||||
if not test_result["success"]:
|
||||
self.handle_test_failure(original_prompt, test_result)
|
||||
return
|
||||
|
||||
self.state.should_break = True
|
||||
logger.info("Test command executed successfully")
|
||||
|
||||
except subprocess.TimeoutExpired as e:
|
||||
logger.warning(f"Test command timed out after {timeout}s: {cmd}")
|
||||
self.state.test_attempts += 1
|
||||
self.state.prompt = f"{original_prompt}. Previous attempt timed out after {timeout} seconds"
|
||||
self.display_test_failure()
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
logger.error(f"Test command failed with exit code {e.returncode}: {cmd}\nOutput: {e.output}")
|
||||
self.state.test_attempts += 1
|
||||
self.state.prompt = f"{original_prompt}. Previous attempt failed with exit code {e.returncode}: {e.output}"
|
||||
self.display_test_failure()
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Test command execution failed: {str(e)}")
|
||||
self.state.test_attempts += 1
|
||||
self.state.should_break = True
|
||||
|
||||
def handle_user_response(response: str, state: TestState, cmd: str, original_prompt: str) -> TestState:
|
||||
"""Handle user's response to test prompt.
|
||||
|
||||
Args:
|
||||
response: User's response (y/n/a)
|
||||
state: Current test state
|
||||
cmd: Test command
|
||||
original_prompt: Original prompt text
|
||||
def handle_user_response(self, response: str, cmd: str, original_prompt: str) -> None:
|
||||
"""Handle user's response to test prompt.
|
||||
Args:
|
||||
response: User's response (y/n/a)
|
||||
cmd: Test command
|
||||
original_prompt: Original prompt text
|
||||
"""
|
||||
response = response.strip().lower()
|
||||
|
||||
Returns:
|
||||
Updated test state
|
||||
"""
|
||||
response = response.strip().lower()
|
||||
|
||||
if response == "n":
|
||||
state.should_break = True
|
||||
return state
|
||||
|
||||
if response == "a":
|
||||
state.auto_test = True
|
||||
return run_test_command(cmd, state, original_prompt)
|
||||
|
||||
if response == "y":
|
||||
return run_test_command(cmd, state, original_prompt)
|
||||
|
||||
return state
|
||||
if response == "n":
|
||||
self.state.should_break = True
|
||||
return
|
||||
|
||||
if response == "a":
|
||||
self.state.auto_test = True
|
||||
self.run_test_command(cmd, original_prompt)
|
||||
return
|
||||
|
||||
if response == "y":
|
||||
self.run_test_command(cmd, original_prompt)
|
||||
|
||||
def check_max_retries(attempts: int, max_retries: int) -> bool:
|
||||
"""Check if max retries reached.
|
||||
|
||||
Args:
|
||||
attempts: Current number of attempts
|
||||
max_retries: Maximum allowed retries
|
||||
def check_max_retries(self) -> bool:
|
||||
"""Check if max retries reached.
|
||||
|
||||
Returns:
|
||||
True if max retries reached
|
||||
"""
|
||||
if attempts >= max_retries:
|
||||
logger.warning("Max test retries reached")
|
||||
return True
|
||||
return False
|
||||
Returns:
|
||||
True if max retries reached
|
||||
"""
|
||||
if self.state.test_attempts >= self.max_retries:
|
||||
logger.warning("Max test retries reached")
|
||||
return True
|
||||
return False
|
||||
|
||||
def execute(self) -> Tuple[bool, str, bool, int]:
|
||||
"""Execute test command and handle retries.
|
||||
|
||||
Returns:
|
||||
Tuple containing:
|
||||
- bool: Whether to break the retry loop
|
||||
- str: Updated prompt
|
||||
- bool: Updated auto_test flag
|
||||
- int: Updated test_attempts count
|
||||
"""
|
||||
if not self.config.get("test_cmd"):
|
||||
self.state.should_break = True
|
||||
return self.state.should_break, self.state.prompt, self.state.auto_test, self.state.test_attempts
|
||||
|
||||
cmd = self.config["test_cmd"]
|
||||
|
||||
if not self.state.auto_test:
|
||||
print()
|
||||
response = ask_human.invoke({"question": "Would you like to run the test command? (y=yes, n=no, a=enable auto-test)"})
|
||||
self.handle_user_response(response, cmd, self.state.prompt)
|
||||
else:
|
||||
if self.check_max_retries():
|
||||
logger.error(f"Maximum number of test retries ({self.max_retries}) reached. Stopping test execution.")
|
||||
console.print(Panel(f"Maximum retries ({self.max_retries}) reached. Test execution stopped.", title="⚠️ Test Execution", border_style="yellow bold"))
|
||||
self.state.should_break = True
|
||||
else:
|
||||
self.run_test_command(cmd, self.state.prompt)
|
||||
|
||||
return self.state.should_break, self.state.prompt, self.state.auto_test, self.state.test_attempts
|
||||
|
||||
def execute_test_command(
|
||||
config: Dict[str, Any],
|
||||
|
|
@ -127,13 +169,13 @@ def execute_test_command(
|
|||
auto_test: bool = False,
|
||||
) -> Tuple[bool, str, bool, int]:
|
||||
"""Execute a test command and handle retries.
|
||||
|
||||
|
||||
Args:
|
||||
config: Configuration dictionary containing test settings
|
||||
original_prompt: The original prompt to append errors to
|
||||
test_attempts: Current number of test attempts
|
||||
auto_test: Whether auto-test mode is enabled
|
||||
|
||||
|
||||
Returns:
|
||||
Tuple containing:
|
||||
- bool: Whether to break the retry loop
|
||||
|
|
@ -141,27 +183,5 @@ def execute_test_command(
|
|||
- bool: Updated auto_test flag
|
||||
- int: Updated test_attempts count
|
||||
"""
|
||||
state = TestState(
|
||||
prompt=original_prompt,
|
||||
test_attempts=test_attempts,
|
||||
auto_test=auto_test
|
||||
)
|
||||
|
||||
if not config.get("test_cmd"):
|
||||
state.should_break = True
|
||||
return state.should_break, state.prompt, state.auto_test, state.test_attempts
|
||||
|
||||
max_retries = config.get("max_test_cmd_retries", 5)
|
||||
cmd = config["test_cmd"]
|
||||
|
||||
if not auto_test:
|
||||
print()
|
||||
response = ask_human.invoke({"question": "Would you like to run the test command? (y=yes, n=no, a=enable auto-test)"})
|
||||
state = handle_user_response(response, state, cmd, original_prompt)
|
||||
else:
|
||||
if check_max_retries(test_attempts, max_retries):
|
||||
state.should_break = True
|
||||
else:
|
||||
state = run_test_command(cmd, state, original_prompt)
|
||||
|
||||
return state.should_break, state.prompt, state.auto_test, state.test_attempts
|
||||
executor = TestCommandExecutor(config, original_prompt, test_attempts, auto_test)
|
||||
return executor.execute()
|
||||
|
|
@ -190,7 +190,8 @@ def test_execute_test_command(
|
|||
|
||||
if auto_test and test_attempts < config.get("max_test_cmd_retries", 5):
|
||||
if config.get("test_cmd"):
|
||||
mock_run_cmd.assert_called_once_with(config["test_cmd"])
|
||||
# Verify run_shell_command called with command and default timeout
|
||||
mock_run_cmd.assert_called_once_with(config["test_cmd"], timeout=config.get('timeout', 30))
|
||||
|
||||
# Verify logging for max retries
|
||||
if test_attempts >= config.get("max_test_cmd_retries", 5):
|
||||
|
|
|
|||
|
|
@ -2,13 +2,11 @@
|
|||
|
||||
import pytest
|
||||
from unittest.mock import patch, Mock
|
||||
import subprocess
|
||||
from ra_aid.tools.handle_user_defined_test_cmd_execution import (
|
||||
TestState,
|
||||
execute_test_command,
|
||||
handle_test_failure,
|
||||
run_test_command,
|
||||
handle_user_response,
|
||||
check_max_retries
|
||||
TestCommandExecutor,
|
||||
execute_test_command
|
||||
)
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -17,93 +15,170 @@ def test_state():
|
|||
return TestState(
|
||||
prompt="test prompt",
|
||||
test_attempts=0,
|
||||
auto_test=False
|
||||
auto_test=False,
|
||||
should_break=False
|
||||
)
|
||||
|
||||
def test_check_max_retries():
|
||||
"""Test max retries check."""
|
||||
assert not check_max_retries(2, 3)
|
||||
assert check_max_retries(3, 3)
|
||||
assert check_max_retries(4, 3)
|
||||
@pytest.fixture
|
||||
def test_executor():
|
||||
"""Create a test executor fixture."""
|
||||
config = {"test_cmd": "test", "max_test_cmd_retries": 3}
|
||||
return TestCommandExecutor(config, "test prompt")
|
||||
|
||||
def test_handle_test_failure(test_state):
|
||||
def test_check_max_retries(test_executor):
|
||||
"""Test max retries check."""
|
||||
test_executor.state.test_attempts = 2
|
||||
assert not test_executor.check_max_retries()
|
||||
|
||||
test_executor.state.test_attempts = 3
|
||||
assert test_executor.check_max_retries()
|
||||
|
||||
test_executor.state.test_attempts = 4
|
||||
assert test_executor.check_max_retries()
|
||||
|
||||
def test_handle_test_failure(test_executor):
|
||||
"""Test handling of test failures."""
|
||||
test_result = {"output": "error message"}
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.display_test_failure"):
|
||||
state = handle_test_failure(test_state, "original", test_result)
|
||||
assert not state.should_break
|
||||
assert "error message" in state.prompt
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.console.print"):
|
||||
test_executor.handle_test_failure("original", test_result)
|
||||
assert not test_executor.state.should_break
|
||||
assert "error message" in test_executor.state.prompt
|
||||
|
||||
def test_run_test_command_success(test_state):
|
||||
def test_run_test_command_success(test_executor):
|
||||
"""Test successful test command execution."""
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.run_shell_command") as mock_run:
|
||||
mock_run.return_value = {"success": True, "output": ""}
|
||||
state = run_test_command("test", test_state, "original")
|
||||
assert state.should_break
|
||||
assert state.test_attempts == 1
|
||||
test_executor.run_test_command("test", "original")
|
||||
assert test_executor.state.should_break
|
||||
assert test_executor.state.test_attempts == 1
|
||||
|
||||
def test_run_test_command_failure(test_state):
|
||||
def test_run_test_command_failure(test_executor):
|
||||
"""Test failed test command execution."""
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.run_shell_command") as mock_run:
|
||||
mock_run.return_value = {"success": False, "output": "error"}
|
||||
state = run_test_command("test", test_state, "original")
|
||||
assert not state.should_break
|
||||
assert state.test_attempts == 1
|
||||
assert "error" in state.prompt
|
||||
test_executor.run_test_command("test", "original")
|
||||
assert not test_executor.state.should_break
|
||||
assert test_executor.state.test_attempts == 1
|
||||
assert "error" in test_executor.state.prompt
|
||||
|
||||
def test_run_test_command_error(test_state):
|
||||
def test_run_test_command_error(test_executor):
|
||||
"""Test test command execution error."""
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.run_shell_command") as mock_run:
|
||||
mock_run.side_effect = Exception("Command failed")
|
||||
state = run_test_command("test", test_state, "original")
|
||||
assert state.should_break
|
||||
assert state.test_attempts == 1
|
||||
mock_run.side_effect = Exception("Generic error")
|
||||
test_executor.run_test_command("test", "original")
|
||||
assert test_executor.state.should_break
|
||||
assert test_executor.state.test_attempts == 1
|
||||
|
||||
def test_handle_user_response_no(test_state):
|
||||
def test_run_test_command_timeout(test_executor):
|
||||
"""Test test command timeout handling."""
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.run_shell_command") as mock_run,\
|
||||
patch("ra_aid.tools.handle_user_defined_test_cmd_execution.logger.warning") as mock_logger:
|
||||
|
||||
# Create a TimeoutExpired exception
|
||||
timeout_exc = subprocess.TimeoutExpired(cmd="test", timeout=30)
|
||||
mock_run.side_effect = timeout_exc
|
||||
|
||||
test_executor.run_test_command("test", "original")
|
||||
|
||||
# Verify state updates
|
||||
assert not test_executor.state.should_break
|
||||
assert test_executor.state.test_attempts == 1
|
||||
assert "timed out after 30 seconds" in test_executor.state.prompt
|
||||
|
||||
# Verify logging
|
||||
mock_logger.assert_called_once()
|
||||
|
||||
def test_run_test_command_called_process_error(test_executor):
|
||||
"""Test handling of CalledProcessError exception."""
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.run_shell_command") as mock_run,\
|
||||
patch("ra_aid.tools.handle_user_defined_test_cmd_execution.logger.error") as mock_logger:
|
||||
|
||||
# Create a CalledProcessError exception
|
||||
process_error = subprocess.CalledProcessError(
|
||||
returncode=1,
|
||||
cmd="test",
|
||||
output="Command failed output"
|
||||
)
|
||||
mock_run.side_effect = process_error
|
||||
|
||||
test_executor.run_test_command("test", "original")
|
||||
|
||||
# Verify state updates
|
||||
assert not test_executor.state.should_break
|
||||
assert test_executor.state.test_attempts == 1
|
||||
assert "failed with exit code 1" in test_executor.state.prompt
|
||||
|
||||
# Verify logging
|
||||
mock_logger.assert_called_once()
|
||||
|
||||
def test_handle_user_response_no(test_executor):
|
||||
"""Test handling of 'no' response."""
|
||||
state = handle_user_response("n", test_state, "test", "original")
|
||||
assert state.should_break
|
||||
assert not state.auto_test
|
||||
test_executor.handle_user_response("n", "test", "original")
|
||||
assert test_executor.state.should_break
|
||||
assert not test_executor.state.auto_test
|
||||
|
||||
def test_handle_user_response_auto(test_state):
|
||||
def test_handle_user_response_auto(test_executor):
|
||||
"""Test handling of 'auto' response."""
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.run_test_command") as mock_run:
|
||||
mock_state = TestState("prompt", 1, True, True)
|
||||
mock_run.return_value = mock_state
|
||||
state = handle_user_response("a", test_state, "test", "original")
|
||||
assert state.auto_test
|
||||
mock_run.assert_called_once()
|
||||
with patch.object(test_executor, "run_test_command") as mock_run:
|
||||
test_executor.handle_user_response("a", "test", "original")
|
||||
assert test_executor.state.auto_test
|
||||
mock_run.assert_called_once_with("test", "original")
|
||||
|
||||
def test_handle_user_response_yes(test_state):
|
||||
def test_handle_user_response_yes(test_executor):
|
||||
"""Test handling of 'yes' response."""
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.run_test_command") as mock_run:
|
||||
mock_state = TestState("prompt", 1, False, True)
|
||||
mock_run.return_value = mock_state
|
||||
state = handle_user_response("y", test_state, "test", "original")
|
||||
assert not state.auto_test
|
||||
mock_run.assert_called_once()
|
||||
with patch.object(test_executor, "run_test_command") as mock_run:
|
||||
test_executor.handle_user_response("y", "test", "original")
|
||||
assert not test_executor.state.auto_test
|
||||
mock_run.assert_called_once_with("test", "original")
|
||||
|
||||
def test_execute_test_command_no_cmd():
|
||||
def test_execute_no_cmd():
|
||||
"""Test execution with no test command."""
|
||||
result = execute_test_command({}, "prompt")
|
||||
executor = TestCommandExecutor({}, "prompt")
|
||||
result = executor.execute()
|
||||
assert result == (True, "prompt", False, 0)
|
||||
|
||||
def test_execute_test_command_manual():
|
||||
def test_execute_manual():
|
||||
"""Test manual test execution."""
|
||||
config = {"test_cmd": "test"}
|
||||
executor = TestCommandExecutor(config, "prompt")
|
||||
|
||||
def mock_handle_response(response, cmd, prompt):
|
||||
# Simulate the behavior of handle_user_response and run_test_command
|
||||
executor.state.should_break = True
|
||||
executor.state.test_attempts = 1
|
||||
executor.state.prompt = "new prompt"
|
||||
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.ask_human") as mock_ask, \
|
||||
patch("ra_aid.tools.handle_user_defined_test_cmd_execution.handle_user_response") as mock_handle:
|
||||
patch.object(executor, "handle_user_response", side_effect=mock_handle_response) as mock_handle:
|
||||
mock_ask.invoke.return_value = "y"
|
||||
mock_state = TestState("new prompt", 1, False, True)
|
||||
mock_handle.return_value = mock_state
|
||||
result = execute_test_command(config, "prompt")
|
||||
|
||||
result = executor.execute()
|
||||
mock_handle.assert_called_once_with("y", "test", "prompt")
|
||||
assert result == (True, "new prompt", False, 1)
|
||||
|
||||
def test_execute_test_command_auto():
|
||||
def test_execute_auto():
|
||||
"""Test auto test execution."""
|
||||
config = {"test_cmd": "test", "max_test_cmd_retries": 3}
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.run_test_command") as mock_run:
|
||||
mock_state = TestState("new prompt", 1, True, True)
|
||||
mock_run.return_value = mock_state
|
||||
executor = TestCommandExecutor(config, "prompt", auto_test=True)
|
||||
|
||||
# Set up state before creating mock
|
||||
executor.state.test_attempts = 1
|
||||
executor.state.should_break = True
|
||||
|
||||
with patch.object(executor, "run_test_command") as mock_run:
|
||||
result = executor.execute()
|
||||
assert result == (True, "prompt", True, 1)
|
||||
mock_run.assert_called_once_with("test", "prompt")
|
||||
|
||||
def test_execute_test_command_function():
|
||||
"""Test the execute_test_command function."""
|
||||
config = {"test_cmd": "test"}
|
||||
with patch("ra_aid.tools.handle_user_defined_test_cmd_execution.TestCommandExecutor") as mock_executor_class:
|
||||
mock_executor = Mock()
|
||||
mock_executor.execute.return_value = (True, "new prompt", True, 1)
|
||||
mock_executor_class.return_value = mock_executor
|
||||
|
||||
result = execute_test_command(config, "prompt", auto_test=True)
|
||||
assert result == (True, "new prompt", True, 1)
|
||||
assert result == (True, "new prompt", True, 1)
|
||||
mock_executor_class.assert_called_once_with(config, "prompt", 0, True)
|
||||
mock_executor.execute.assert_called_once()
|
||||
Loading…
Reference in New Issue