"""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 TestNestedFieldAccess: """Tests for nested field access in variable substitution.""" def test_nested_dict_access(self): """Access nested dict field via dot notation.""" result = substitute_variables( "Answer: {result.output}", {"result": {"output": "Hello", "reasoning": "User said hi"}} ) assert result == "Answer: Hello" def test_nested_json_string_access(self): """Access field in JSON string via dot notation.""" result = substitute_variables( "Answer: {result.output}", {"result": '{"output": "Hello", "reasoning": "User said hi"}'} ) assert result == "Answer: Hello" def test_nested_deep_access(self): """Access deeply nested fields.""" result = substitute_variables( "Value: {data.nested.deep}", {"data": {"nested": {"deep": "found"}}} ) assert result == "Value: found" def test_nested_missing_field_unchanged(self): """Missing nested field leaves placeholder unchanged.""" result = substitute_variables( "Value: {result.missing}", {"result": {"output": "Hello"}} ) assert result == "Value: {result.missing}" def test_nested_on_non_dict_unchanged(self): """Nested access on non-dict value leaves placeholder unchanged.""" result = substitute_variables( "Value: {result.field}", {"result": "just a string"} ) assert result == "Value: {result.field}" def test_nested_invalid_json_unchanged(self): """Invalid JSON string leaves placeholder unchanged.""" result = substitute_variables( "Value: {result.field}", {"result": "not valid json {"} ) assert result == "Value: {result.field}" def test_nested_array_in_json(self): """Nested array values should be JSON-serialized.""" result = substitute_variables( "Items: {result.items}", {"result": '{"items": [1, 2, 3]}'} ) assert result == "Items: [1, 2, 3]" def test_nested_object_in_json(self): """Nested object values should be JSON-serialized.""" result = substitute_variables( "Data: {result.data}", {"result": '{"data": {"key": "value"}}'} ) assert result == 'Data: {"key": "value"}' def test_settings_access_unchanged(self): """Settings access should use existing special handling.""" result = substitute_variables( "Backend: {settings.backend}", {"settings": {"backend": "piper"}} ) assert result == "Backend: piper" 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 ) # Use plain_text=True to test legacy behavior (no schema enforcement) step = PromptStep( prompt="Summarize: {input}", provider="claude", output_var="summary", plain_text=True ) 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", plain_text=True) 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", plain_text=True) 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", plain_text=True) execute_prompt_step(step, {"input": ""}, provider_override="gpt4") # Should use override, not step's provider assert mock_call.call_args[0][0] == "gpt4" class TestStructuredOutput: """Tests for structured output enforcement.""" @patch('cmdforge.runner.call_provider') def test_structured_output_valid_json(self, mock_call): """Valid JSON matching default schema should succeed.""" mock_call.return_value = ProviderResult( text='{"output": "Hello, world!", "reasoning": "Greeting requested"}', success=True ) step = PromptStep( prompt="Say hello", provider="claude", output_var="result" # plain_text defaults to False - structured output enabled ) output, success = execute_prompt_step(step, {"input": ""}) assert success is True # Output should be a parsed dict (for code step compatibility) assert isinstance(output, dict) assert output["output"] == "Hello, world!" assert output["reasoning"] == "Greeting requested" @patch('cmdforge.runner.call_provider') def test_structured_output_strips_markdown_fences(self, mock_call): """Markdown code fences should be stripped before parsing.""" mock_call.return_value = ProviderResult( text='```json\n{"output": "Hello"}\n```', success=True ) step = PromptStep( prompt="Say hello", provider="claude", output_var="result" ) output, success = execute_prompt_step(step, {"input": ""}) assert success is True assert isinstance(output, dict) assert output["output"] == "Hello" @patch('cmdforge.runner.call_provider') def test_structured_output_retry_on_invalid_json(self, mock_call): """Invalid JSON should trigger retry with error feedback.""" # First call returns invalid JSON, second returns valid mock_call.side_effect = [ ProviderResult(text="Not JSON at all", success=True), ProviderResult(text='{"output": "Fixed!"}', success=True), ] step = PromptStep( prompt="Test", provider="claude", output_var="result", max_retries=1 ) output, success = execute_prompt_step(step, {"input": ""}) assert success is True assert mock_call.call_count == 2 # Second call should include error feedback second_call_prompt = mock_call.call_args_list[1][0][1] assert "failed validation" in second_call_prompt.lower() @patch('cmdforge.runner.call_provider') def test_structured_output_fails_after_retries(self, mock_call): """Should fail after exhausting retries.""" mock_call.return_value = ProviderResult( text="Still not JSON", success=True ) step = PromptStep( prompt="Test", provider="claude", output_var="result", max_retries=1 # 1 retry = 2 total attempts ) output, success = execute_prompt_step(step, {"input": ""}) assert success is False assert mock_call.call_count == 2 @patch('cmdforge.runner.call_provider') def test_structured_output_custom_schema(self, mock_call): """Custom schema should be validated.""" mock_call.return_value = ProviderResult( text='{"score": 0.8, "reasoning": "Good match"}', success=True ) custom_schema = { "type": "object", "properties": { "score": {"type": "number", "minimum": 0, "maximum": 1}, "reasoning": {"type": "string"} }, "required": ["score"] } step = PromptStep( prompt="Score this", provider="claude", output_var="result", output_schema=custom_schema ) output, success = execute_prompt_step(step, {"input": ""}) assert success is True assert isinstance(output, dict) assert output["score"] == 0.8 @patch('cmdforge.runner.call_provider') def test_structured_output_schema_violation_triggers_retry(self, mock_call): """Schema violation should trigger retry.""" # First call missing required 'output' field mock_call.side_effect = [ ProviderResult(text='{"reasoning": "thinking..."}', success=True), ProviderResult(text='{"output": "Fixed!", "reasoning": "done"}', success=True), ] step = PromptStep( prompt="Test", provider="claude", output_var="result", max_retries=1 ) output, success = execute_prompt_step(step, {"input": ""}) assert success is True assert mock_call.call_count == 2 @patch('cmdforge.runner.call_provider') def test_plain_text_bypasses_validation(self, mock_call): """plain_text=True should skip all JSON validation.""" mock_call.return_value = ProviderResult( text="Just plain text, no JSON here!", success=True ) step = PromptStep( prompt="Write prose", provider="claude", output_var="result", plain_text=True ) output, success = execute_prompt_step(step, {"input": ""}) assert success is True assert output == "Just plain text, no JSON here!" # Should only call provider once (no retries for plain text) assert mock_call.call_count == 1 class TestSchemaInstructions: """Tests for schema instruction generation.""" def test_generate_schema_example_default(self): """Default schema should generate output/reasoning example.""" from cmdforge.runner import generate_schema_example, DEFAULT_OUTPUT_SCHEMA import json example = generate_schema_example(DEFAULT_OUTPUT_SCHEMA) parsed = json.loads(example) assert "output" in parsed assert "reasoning" in parsed assert "<" in parsed["output"] # Placeholder format def test_generate_schema_example_with_enum(self): """Enum fields should use first enum value.""" from cmdforge.runner import generate_schema_example import json schema = { "type": "object", "properties": { "intent": {"type": "string", "enum": ["question", "task", "chat"]} } } example = generate_schema_example(schema) parsed = json.loads(example) assert parsed["intent"] == "question" # First enum value def test_generate_schema_example_with_array(self): """Array fields should generate array example.""" from cmdforge.runner import generate_schema_example import json schema = { "type": "object", "properties": { "items": {"type": "array", "items": {"type": "string"}} } } example = generate_schema_example(schema) parsed = json.loads(example) assert isinstance(parsed["items"], list) assert len(parsed["items"]) == 2 def test_generate_schema_example_nested_object(self): """Nested objects should be recursively generated.""" from cmdforge.runner import generate_schema_example import json schema = { "type": "object", "properties": { "scores": { "type": "object", "properties": { "accuracy": {"type": "number"}, "completeness": {"type": "number"} } } } } example = generate_schema_example(schema) parsed = json.loads(example) assert isinstance(parsed["scores"], dict) assert parsed["scores"]["accuracy"] == 0.0 assert parsed["scores"]["completeness"] == 0.0 def test_prepend_schema_instructions_format(self): """Instructions should be prepended, not appended.""" from cmdforge.runner import prepend_schema_instructions, DEFAULT_OUTPUT_SCHEMA prompt = "What is 2+2?" result = prepend_schema_instructions(prompt, DEFAULT_OUTPUT_SCHEMA) # Instructions should come before the prompt assert result.startswith("Respond with ONLY valid JSON") assert prompt in result # Prompt should be after the separator assert result.index("---") < result.index(prompt) def test_prepend_schema_instructions_includes_guidance(self): """Should include field guidance for output/reasoning schemas.""" from cmdforge.runner import prepend_schema_instructions, DEFAULT_OUTPUT_SCHEMA result = prepend_schema_instructions("Test", DEFAULT_OUTPUT_SCHEMA) assert "reasoning" in result.lower() assert "output" in result.lower() assert "Required:" in result 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", plain_text=True # Use plain text mode for legacy behavior ) ], 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", plain_text=True), 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", plain_text=True) ], 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", plain_text=True) ], 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", plain_text=True) ], 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"