Coverage for little_loops / cli / loop / testing.py: 46%
154 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""ll-loop testing subcommands: test, simulate."""
3from __future__ import annotations
5import argparse
6from pathlib import Path
8from little_loops.cli.loop._helpers import load_loop
9from little_loops.logger import Logger
12def cmd_test(
13 loop_name: str,
14 args: argparse.Namespace,
15 loops_dir: Path,
16 logger: Logger,
17) -> int:
18 """Run a single test iteration to verify loop configuration.
20 Executes the target state's action and evaluation, then reports
21 what the loop would do without actually transitioning further.
22 """
23 from little_loops.fsm.evaluators import EvaluationResult, evaluate, evaluate_exit_code
24 from little_loops.fsm.executor import DefaultActionRunner
25 from little_loops.fsm.interpolation import InterpolationContext
27 try:
28 fsm = load_loop(loop_name, loops_dir, logger)
29 except FileNotFoundError as e:
30 logger.error(str(e))
31 return 1
32 except ValueError as e:
33 logger.error(f"Validation error: {e}")
34 return 1
36 # Determine target state
37 target = args.state if args.state else fsm.initial
38 if target not in fsm.states:
39 logger.error(f"State '{target}' not found. Available: {', '.join(fsm.states)}")
40 return 1
41 state_config = fsm.states[target]
43 print(f"## Test Iteration: {loop_name}")
44 print()
45 print(f"State: {target}")
47 # If no action, report and exit
48 if not state_config.action:
49 print(f"State '{target}' has no action to test")
50 print()
51 print("\u2713 Loop structure is valid (no check action to execute)")
52 return 0
54 action = state_config.action
55 is_slash = action.startswith("/") or state_config.action_type in (
56 "prompt",
57 "slash_command",
58 )
60 print(f"Action: {action}")
61 print()
63 if is_slash:
64 from little_loops.fsm.executor import ActionResult, SimulationActionRunner
66 exit_code_arg = getattr(args, "exit_code", None)
67 if exit_code_arg is not None:
68 sim_exit_code = exit_code_arg
69 print(f"[SIMULATED] Using --exit-code {sim_exit_code}")
70 else:
71 sim_runner = SimulationActionRunner()
72 sim_result = sim_runner.run(action, timeout=120, is_slash_command=True)
73 sim_exit_code = sim_result.exit_code
74 print()
75 result = ActionResult(
76 output=f"[simulated output for: {action}]",
77 stderr="",
78 exit_code=sim_exit_code,
79 duration_ms=0,
80 )
81 else:
82 # Run the action
83 runner = DefaultActionRunner()
84 timeout = state_config.timeout or 120
85 result = runner.run(action, timeout=timeout, is_slash_command=False)
87 print(f"Exit code: {result.exit_code}")
89 # Truncate output for display
90 output_lines = result.output.strip().split("\n")
91 if len(output_lines) > 10:
92 extra = len(output_lines) - 10
93 output_preview = "\n".join(output_lines[:10]) + f"\n... ({extra} more lines)"
94 elif len(result.output) > 500:
95 output_preview = result.output[:500] + "..."
96 else:
97 output_preview = result.output.strip() if result.output.strip() else "(empty)"
99 print(f"Output:\n{output_preview}")
101 if result.stderr:
102 stderr_lines = result.stderr.strip().split("\n")
103 if len(stderr_lines) > 5:
104 extra = len(stderr_lines) - 5
105 stderr_preview = "\n".join(stderr_lines[:5]) + f"\n... ({extra} more lines)"
106 else:
107 stderr_preview = result.stderr.strip()
108 print(f"Stderr:\n{stderr_preview}")
110 print()
112 # Evaluate
113 ctx = InterpolationContext()
114 eval_result: EvaluationResult
116 if state_config.evaluate:
117 eval_result = evaluate(
118 config=state_config.evaluate,
119 output=result.output,
120 exit_code=result.exit_code,
121 context=ctx,
122 )
123 evaluator_type: str = state_config.evaluate.type
124 else:
125 # Default to exit_code evaluation
126 eval_result = evaluate_exit_code(result.exit_code)
127 evaluator_type = "exit_code (default)"
129 print(f"Evaluator: {evaluator_type}")
130 print(f"Verdict: {eval_result.verdict.upper()}")
132 if eval_result.details:
133 for key, value in eval_result.details.items():
134 if key != "exit_code" or evaluator_type != "exit_code (default)":
135 print(f" {key}: {value}")
137 # Determine next state based on verdict
138 verdict = eval_result.verdict
139 next_state = None
141 if state_config.route:
142 routes = state_config.route.routes
143 if verdict in routes:
144 next_state = routes[verdict]
145 elif state_config.route.default:
146 next_state = state_config.route.default
147 else:
148 if verdict == "yes" and state_config.on_yes:
149 next_state = state_config.on_yes
150 elif verdict == "no" and state_config.on_no:
151 next_state = state_config.on_no
152 elif verdict == "error" and state_config.on_error:
153 next_state = state_config.on_error
154 elif verdict in state_config.extra_routes:
155 next_state = state_config.extra_routes[verdict]
157 print()
158 if next_state:
159 print(f"Would transition: {target} \u2192 {next_state}")
160 else:
161 print(f"Would transition: {target} \u2192 (no route for '{verdict}')")
163 # Summary
164 print()
165 has_error = eval_result.verdict == "error" or "error" in eval_result.details
166 if has_error:
167 print("\u26a0 Loop has issues - review the error details above")
168 return 1
169 else:
170 print("\u2713 Loop appears to be configured correctly")
171 return 0
174def cmd_simulate(
175 loop_name: str,
176 args: argparse.Namespace,
177 loops_dir: Path,
178 logger: Logger,
179) -> int:
180 """Run interactive simulation of loop execution.
182 Traces through loop logic without executing commands, allowing users
183 to verify state transitions and understand loop behavior.
184 """
185 from little_loops.fsm.executor import FSMExecutor, SimulationActionRunner
187 try:
188 fsm = load_loop(loop_name, loops_dir, logger)
189 except FileNotFoundError as e:
190 logger.error(str(e))
191 return 1
192 except ValueError as e:
193 logger.error(f"Validation error: {e}")
194 return 1
196 # Apply CLI overrides
197 if args.max_iterations:
198 fsm.max_iterations = args.max_iterations
199 else:
200 # Limit iterations for simulation safety (cap at 20 unless overridden)
201 if fsm.max_iterations > 20:
202 logger.info(f"Limiting simulation to 20 iterations (loop config: {fsm.max_iterations})")
203 fsm.max_iterations = 20
205 # Create simulation runner
206 sim_runner = SimulationActionRunner(scenario=args.scenario)
208 # Track simulation state
209 states_visited: list[str] = []
211 def simulation_callback(event: dict) -> None:
212 """Display simulation progress."""
213 event_type = event.get("event")
215 if event_type == "state_enter":
216 iteration = event.get("iteration", 0)
217 state = event.get("state", "")
218 states_visited.append(state)
219 print()
220 print(f"[{iteration}] State: {state}")
222 elif event_type == "action_start":
223 action = event.get("action", "")
224 action_display = action[:70] + "..." if len(action) > 70 else action
225 print(f" Action: {action_display}")
227 elif event_type == "evaluate":
228 evaluator = event.get("type", "exit_code")
229 verdict = event.get("verdict", "")
230 print(f" Evaluator: {evaluator}")
231 print(f" Result: {verdict.upper()}")
233 elif event_type == "route":
234 from_state = event.get("from", "")
235 to_state = event.get("to", "")
236 print(f" Transition: {from_state} \u2192 {to_state}")
238 # Print header
239 mode_str = f"scenario={args.scenario}" if args.scenario else "interactive"
240 print(f"=== SIMULATION: {fsm.name} ({mode_str}) ===")
242 # Run simulation
243 executor = FSMExecutor(
244 fsm,
245 event_callback=simulation_callback,
246 action_runner=sim_runner,
247 )
248 result = executor.run()
250 # Print summary
251 print()
252 print("=== Summary ===")
253 arrow = " \u2192 "
254 print(f"States visited: {arrow.join(states_visited)}")
255 print(f"Iterations: {result.iterations}")
256 print(f"Would have executed {len(sim_runner.calls)} commands")
257 print(f"Terminated by: {result.terminated_by}")
259 return 0