581 lines
18 KiB
Python
581 lines
18 KiB
Python
"""Tests for runner.py - Tool execution engine."""
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from cmdforge.runner import (
|
|
substitute_variables,
|
|
execute_prompt_step,
|
|
execute_code_step,
|
|
run_tool,
|
|
create_argument_parser
|
|
)
|
|
from cmdforge.tool import Tool, ToolArgument, PromptStep, CodeStep
|
|
from cmdforge.providers import ProviderResult
|
|
|
|
|
|
class TestSubstituteVariables:
|
|
"""Tests for variable substitution."""
|
|
|
|
def test_simple_substitution(self):
|
|
result = substitute_variables("Hello {name}", {"name": "World"})
|
|
assert result == "Hello World"
|
|
|
|
def test_multiple_variables(self):
|
|
result = substitute_variables(
|
|
"{greeting}, {name}!",
|
|
{"greeting": "Hello", "name": "Alice"}
|
|
)
|
|
assert result == "Hello, Alice!"
|
|
|
|
def test_same_variable_multiple_times(self):
|
|
result = substitute_variables(
|
|
"{x} + {x} = {y}",
|
|
{"x": "1", "y": "2"}
|
|
)
|
|
assert result == "1 + 1 = 2"
|
|
|
|
def test_missing_variable_unchanged(self):
|
|
result = substitute_variables("Hello {name}", {"other": "value"})
|
|
assert result == "Hello {name}"
|
|
|
|
def test_empty_value(self):
|
|
result = substitute_variables("Value: {x}", {"x": ""})
|
|
assert result == "Value: "
|
|
|
|
def test_none_value(self):
|
|
result = substitute_variables("Value: {x}", {"x": None})
|
|
assert result == "Value: "
|
|
|
|
def test_escaped_braces_double_to_single(self):
|
|
"""{{x}} should become {x} (literal braces)."""
|
|
result = substitute_variables("Use {{braces}}", {"braces": "nope"})
|
|
assert result == "Use {braces}"
|
|
|
|
def test_escaped_braces_not_substituted(self):
|
|
"""Escaped braces should not be treated as variables."""
|
|
result = substitute_variables(
|
|
"Format: {{name}} is {name}",
|
|
{"name": "Alice"}
|
|
)
|
|
assert result == "Format: {name} is Alice"
|
|
|
|
def test_escaped_and_normal_mixed(self):
|
|
result = substitute_variables(
|
|
"{{literal}} and {variable}",
|
|
{"variable": "substituted", "literal": "ignored"}
|
|
)
|
|
assert result == "{literal} and substituted"
|
|
|
|
def test_nested_braces(self):
|
|
"""Edge case: nested braces.
|
|
|
|
{{{x}}} has overlapping escape sequences:
|
|
- First {{ is escaped
|
|
- Then }} at end overlaps with the closing } of {x}
|
|
- This breaks the {x} placeholder, leaving it as literal text
|
|
"""
|
|
result = substitute_variables("{{{x}}}", {"x": "val"})
|
|
# The }} at end captures part of {x}}, breaking the placeholder
|
|
assert result == "{{x}}"
|
|
|
|
def test_multiline_template(self):
|
|
template = """Line 1: {var1}
|
|
Line 2: {var2}
|
|
Line 3: {var1} again"""
|
|
result = substitute_variables(template, {"var1": "A", "var2": "B"})
|
|
assert "Line 1: A" in result
|
|
assert "Line 2: B" in result
|
|
assert "Line 3: A again" in result
|
|
|
|
def test_numeric_value(self):
|
|
result = substitute_variables("Count: {n}", {"n": 42})
|
|
assert result == "Count: 42"
|
|
|
|
def test_zero_value_preserved(self):
|
|
"""Zero should be preserved, not converted to empty string."""
|
|
result = substitute_variables("Count: {n}", {"n": 0})
|
|
assert result == "Count: 0"
|
|
|
|
def test_false_value_preserved(self):
|
|
"""False should be preserved, not converted to empty string."""
|
|
result = substitute_variables("Enabled: {flag}", {"flag": False})
|
|
assert result == "Enabled: False"
|
|
|
|
def test_true_value_preserved(self):
|
|
"""True should be preserved as 'True'."""
|
|
result = substitute_variables("Enabled: {flag}", {"flag": True})
|
|
assert result == "Enabled: True"
|
|
|
|
|
|
class TestExecutePromptStep:
|
|
"""Tests for prompt step execution."""
|
|
|
|
@patch('cmdforge.runner.call_provider')
|
|
def test_successful_prompt(self, mock_call):
|
|
mock_call.return_value = ProviderResult(
|
|
text="This is the response",
|
|
success=True
|
|
)
|
|
|
|
step = PromptStep(
|
|
prompt="Summarize: {input}",
|
|
provider="claude",
|
|
output_var="summary"
|
|
)
|
|
variables = {"input": "Some text to summarize"}
|
|
|
|
output, success = execute_prompt_step(step, variables)
|
|
|
|
assert success is True
|
|
assert output == "This is the response"
|
|
mock_call.assert_called_once()
|
|
# Verify the prompt was substituted
|
|
call_args = mock_call.call_args
|
|
assert "Some text to summarize" in call_args[0][1]
|
|
|
|
@patch('cmdforge.runner.call_provider')
|
|
def test_failed_prompt(self, mock_call):
|
|
mock_call.return_value = ProviderResult(
|
|
text="",
|
|
success=False,
|
|
error="Provider error"
|
|
)
|
|
|
|
step = PromptStep(prompt="Test", provider="claude", output_var="out")
|
|
output, success = execute_prompt_step(step, {"input": ""})
|
|
|
|
assert success is False
|
|
assert output == ""
|
|
|
|
@patch('cmdforge.runner.mock_provider')
|
|
def test_mock_provider_used(self, mock_mock):
|
|
mock_mock.return_value = ProviderResult(text="mock response", success=True)
|
|
|
|
step = PromptStep(prompt="Test", provider="mock", output_var="out")
|
|
output, success = execute_prompt_step(step, {"input": ""})
|
|
|
|
assert success is True
|
|
mock_mock.assert_called_once()
|
|
|
|
@patch('cmdforge.runner.call_provider')
|
|
def test_provider_override(self, mock_call):
|
|
mock_call.return_value = ProviderResult(text="response", success=True)
|
|
|
|
step = PromptStep(prompt="Test", provider="claude", output_var="out")
|
|
execute_prompt_step(step, {"input": ""}, provider_override="gpt4")
|
|
|
|
# Should use override, not step's provider
|
|
assert mock_call.call_args[0][0] == "gpt4"
|
|
|
|
|
|
class TestExecuteCodeStep:
|
|
"""Tests for code step execution."""
|
|
|
|
def test_simple_code(self):
|
|
step = CodeStep(
|
|
code="result = input.upper()",
|
|
output_var="result"
|
|
)
|
|
variables = {"input": "hello"}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is True
|
|
assert outputs["result"] == "HELLO"
|
|
|
|
def test_multiple_output_vars(self):
|
|
step = CodeStep(
|
|
code="a = 1\nb = 2\nc = a + b",
|
|
output_var="a, b, c"
|
|
)
|
|
variables = {"input": ""}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is True
|
|
# Code step outputs preserve original types (integers here)
|
|
assert outputs["a"] == 1
|
|
assert outputs["b"] == 2
|
|
assert outputs["c"] == 3
|
|
|
|
def test_code_uses_variables(self):
|
|
step = CodeStep(
|
|
code="result = f'{prefix}: {input}'",
|
|
output_var="result"
|
|
)
|
|
variables = {"input": "content", "prefix": "Data"}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is True
|
|
assert outputs["result"] == "Data: content"
|
|
|
|
def test_code_with_variable_substitution(self):
|
|
"""Variables in code are substituted before exec."""
|
|
step = CodeStep(
|
|
code="filename = '{outputfile}'",
|
|
output_var="filename"
|
|
)
|
|
variables = {"outputfile": "/tmp/test.txt"}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is True
|
|
assert outputs["filename"] == "/tmp/test.txt"
|
|
|
|
def test_code_error(self):
|
|
step = CodeStep(
|
|
code="result = undefined_variable",
|
|
output_var="result"
|
|
)
|
|
variables = {"input": ""}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is False
|
|
assert outputs == {}
|
|
|
|
def test_code_syntax_error(self):
|
|
step = CodeStep(
|
|
code="this is not valid python",
|
|
output_var="result"
|
|
)
|
|
variables = {"input": ""}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is False
|
|
|
|
def test_code_can_use_builtins(self):
|
|
"""Code should have access to Python builtins."""
|
|
step = CodeStep(
|
|
code="result = len(input.split())",
|
|
output_var="result"
|
|
)
|
|
variables = {"input": "one two three"}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is True
|
|
# Code step outputs preserve original types (integer here)
|
|
assert outputs["result"] == 3
|
|
|
|
def test_code_preserves_complex_types(self):
|
|
"""Code step outputs preserve lists/dicts for code→code workflows."""
|
|
step = CodeStep(
|
|
code="data = [1, 2, 3]\ninfo = {'key': 'value'}",
|
|
output_var="data, info"
|
|
)
|
|
variables = {"input": ""}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is True
|
|
assert outputs["data"] == [1, 2, 3]
|
|
assert outputs["info"] == {"key": "value"}
|
|
|
|
def test_code_to_code_workflow(self):
|
|
"""Test that complex types can flow between code steps."""
|
|
# First step creates a list
|
|
step1 = CodeStep(
|
|
code="numbers = [1, 2, 3, 4, 5]",
|
|
output_var="numbers"
|
|
)
|
|
variables = {"input": ""}
|
|
outputs1, _ = execute_code_step(step1, variables)
|
|
variables.update(outputs1)
|
|
|
|
# Second step uses the list directly (no re-parsing needed)
|
|
step2 = CodeStep(
|
|
code="total = sum(numbers)",
|
|
output_var="total"
|
|
)
|
|
outputs2, success = execute_code_step(step2, variables)
|
|
|
|
assert success is True
|
|
assert outputs2["total"] == 15
|
|
|
|
def test_nested_function_can_access_imports(self):
|
|
"""Nested functions should be able to access module-level imports.
|
|
|
|
This tests a fix for a scoping bug where exec() with separate
|
|
globals/locals dicts caused nested functions to fail when accessing
|
|
imported modules. The fix uses the same dict for both globals and
|
|
locals so imports are visible to nested function closures.
|
|
"""
|
|
step = CodeStep(
|
|
code="""
|
|
import re
|
|
|
|
def parse_numbers(text):
|
|
# This should be able to access 're' from the outer scope
|
|
matches = re.findall(r'\\d+', text)
|
|
return [int(m) for m in matches]
|
|
|
|
result = parse_numbers(input)
|
|
""",
|
|
output_var="result"
|
|
)
|
|
variables = {"input": "abc 123 def 456 ghi"}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is True
|
|
assert outputs["result"] == [123, 456]
|
|
|
|
def test_nested_function_can_access_outer_variables(self):
|
|
"""Nested functions should access variables defined in outer scope."""
|
|
step = CodeStep(
|
|
code="""
|
|
multiplier = 10
|
|
|
|
def multiply_all(numbers):
|
|
return [n * multiplier for n in numbers]
|
|
|
|
result = multiply_all([1, 2, 3])
|
|
""",
|
|
output_var="result"
|
|
)
|
|
variables = {"input": ""}
|
|
|
|
outputs, success = execute_code_step(step, variables)
|
|
|
|
assert success is True
|
|
assert outputs["result"] == [10, 20, 30]
|
|
|
|
|
|
class TestRunTool:
|
|
"""Tests for run_tool function."""
|
|
|
|
def test_tool_with_no_steps(self):
|
|
"""Tool with no steps just substitutes output template."""
|
|
tool = Tool(
|
|
name="echo",
|
|
output="You said: {input}"
|
|
)
|
|
|
|
output, exit_code = run_tool(tool, "hello", {})
|
|
|
|
assert exit_code == 0
|
|
assert output == "You said: hello"
|
|
|
|
def test_tool_with_arguments(self):
|
|
tool = Tool(
|
|
name="greet",
|
|
arguments=[
|
|
ToolArgument(flag="--name", variable="name", default="World")
|
|
],
|
|
output="Hello, {name}!"
|
|
)
|
|
|
|
# With default
|
|
output, exit_code = run_tool(tool, "", {})
|
|
assert output == "Hello, World!"
|
|
|
|
# With custom value
|
|
output, exit_code = run_tool(tool, "", {"name": "Alice"})
|
|
assert output == "Hello, Alice!"
|
|
|
|
@patch('cmdforge.runner.call_provider')
|
|
def test_tool_with_prompt_step(self, mock_call):
|
|
mock_call.return_value = ProviderResult(text="Summarized!", success=True)
|
|
|
|
tool = Tool(
|
|
name="summarize",
|
|
steps=[
|
|
PromptStep(
|
|
prompt="Summarize: {input}",
|
|
provider="claude",
|
|
output_var="summary"
|
|
)
|
|
],
|
|
output="{summary}"
|
|
)
|
|
|
|
output, exit_code = run_tool(tool, "Long text here", {})
|
|
|
|
assert exit_code == 0
|
|
assert output == "Summarized!"
|
|
|
|
def test_tool_with_code_step(self):
|
|
tool = Tool(
|
|
name="word-count",
|
|
steps=[
|
|
CodeStep(
|
|
code="count = len(input.split())",
|
|
output_var="count"
|
|
)
|
|
],
|
|
output="Words: {count}"
|
|
)
|
|
|
|
output, exit_code = run_tool(tool, "one two three four", {})
|
|
|
|
assert exit_code == 0
|
|
assert output == "Words: 4"
|
|
|
|
@patch('cmdforge.runner.call_provider')
|
|
def test_tool_with_multiple_steps(self, mock_call):
|
|
mock_call.return_value = ProviderResult(text="AI response", success=True)
|
|
|
|
tool = Tool(
|
|
name="multi-step",
|
|
steps=[
|
|
CodeStep(code="preprocessed = input.strip().upper()", output_var="preprocessed"),
|
|
PromptStep(prompt="Process: {preprocessed}", provider="claude", output_var="response"),
|
|
CodeStep(code="final = f'Result: {response}'", output_var="final")
|
|
],
|
|
output="{final}"
|
|
)
|
|
|
|
output, exit_code = run_tool(tool, " test ", {})
|
|
|
|
assert exit_code == 0
|
|
assert "Result: AI response" in output
|
|
|
|
@patch('cmdforge.runner.call_provider')
|
|
def test_tool_dry_run(self, mock_call):
|
|
tool = Tool(
|
|
name="test",
|
|
steps=[
|
|
PromptStep(prompt="Test", provider="claude", output_var="response")
|
|
],
|
|
output="{response}"
|
|
)
|
|
|
|
output, exit_code = run_tool(tool, "input", {}, dry_run=True)
|
|
|
|
# Provider should not be called
|
|
mock_call.assert_not_called()
|
|
assert exit_code == 0
|
|
assert "DRY RUN" in output
|
|
|
|
@patch('cmdforge.runner.call_provider')
|
|
def test_tool_provider_override(self, mock_call):
|
|
mock_call.return_value = ProviderResult(text="response", success=True)
|
|
|
|
tool = Tool(
|
|
name="test",
|
|
steps=[
|
|
PromptStep(prompt="Test", provider="claude", output_var="response")
|
|
],
|
|
output="{response}"
|
|
)
|
|
|
|
run_tool(tool, "input", {}, provider_override="gpt4")
|
|
|
|
# Should use override
|
|
assert mock_call.call_args[0][0] == "gpt4"
|
|
|
|
@patch('cmdforge.runner.call_provider')
|
|
def test_tool_prompt_failure(self, mock_call):
|
|
mock_call.return_value = ProviderResult(text="", success=False, error="API error")
|
|
|
|
tool = Tool(
|
|
name="test",
|
|
steps=[
|
|
PromptStep(prompt="Test", provider="claude", output_var="response")
|
|
],
|
|
output="{response}"
|
|
)
|
|
|
|
output, exit_code = run_tool(tool, "input", {})
|
|
|
|
assert exit_code == 2
|
|
assert output == ""
|
|
|
|
def test_tool_code_failure(self):
|
|
tool = Tool(
|
|
name="test",
|
|
steps=[
|
|
CodeStep(code="raise ValueError('fail')", output_var="result")
|
|
],
|
|
output="{result}"
|
|
)
|
|
|
|
output, exit_code = run_tool(tool, "input", {})
|
|
|
|
assert exit_code == 1
|
|
assert output == ""
|
|
|
|
def test_variables_flow_between_steps(self):
|
|
"""Variables from earlier steps should be available in later steps."""
|
|
tool = Tool(
|
|
name="flow",
|
|
arguments=[
|
|
ToolArgument(flag="--prefix", variable="prefix", default=">>")
|
|
],
|
|
steps=[
|
|
CodeStep(code="step1 = input.upper()", output_var="step1"),
|
|
CodeStep(code="step2 = f'{prefix} {step1}'", output_var="step2")
|
|
],
|
|
output="{step2}"
|
|
)
|
|
|
|
output, exit_code = run_tool(tool, "hello", {"prefix": ">>"})
|
|
|
|
assert exit_code == 0
|
|
assert output == ">> HELLO"
|
|
|
|
|
|
class TestCreateArgumentParser:
|
|
"""Tests for argument parser creation."""
|
|
|
|
def test_basic_parser(self):
|
|
tool = Tool(name="test", description="Test tool")
|
|
parser = create_argument_parser(tool)
|
|
|
|
assert parser.prog == "test"
|
|
assert "Test tool" in parser.description
|
|
|
|
def test_parser_with_universal_flags(self):
|
|
tool = Tool(name="test")
|
|
parser = create_argument_parser(tool)
|
|
|
|
# Parse with universal flags
|
|
args = parser.parse_args(["--dry-run", "--verbose", "-p", "mock"])
|
|
|
|
assert args.dry_run is True
|
|
assert args.verbose is True
|
|
assert args.provider == "mock"
|
|
|
|
def test_parser_with_tool_arguments(self):
|
|
tool = Tool(
|
|
name="test",
|
|
arguments=[
|
|
ToolArgument(flag="--max", variable="max_size", default="100"),
|
|
ToolArgument(flag="--format", variable="format", description="Output format")
|
|
]
|
|
)
|
|
parser = create_argument_parser(tool)
|
|
|
|
# Parse with custom flags
|
|
args = parser.parse_args(["--max", "50", "--format", "json"])
|
|
|
|
assert args.max_size == "50"
|
|
assert args.format == "json"
|
|
|
|
def test_parser_default_values(self):
|
|
tool = Tool(
|
|
name="test",
|
|
arguments=[
|
|
ToolArgument(flag="--count", variable="count", default="10")
|
|
]
|
|
)
|
|
parser = create_argument_parser(tool)
|
|
|
|
# Parse without providing the flag
|
|
args = parser.parse_args([])
|
|
|
|
assert args.count == "10"
|
|
|
|
def test_parser_input_output_flags(self):
|
|
tool = Tool(name="test")
|
|
parser = create_argument_parser(tool)
|
|
|
|
args = parser.parse_args(["-i", "input.txt", "-o", "output.txt"])
|
|
|
|
assert args.input_file == "input.txt"
|
|
assert args.output_file == "output.txt"
|