From fa49ca08b445b50fbcc4aa81b0bbcce3e31f2a75 Mon Sep 17 00:00:00 2001 From: oleks-dev <1228369+oleks-dev@users.noreply.github.com> Date: Sun, 12 Oct 2025 14:34:49 +0200 Subject: [PATCH] add debug feature --- prich/cli/dynamic_command_group.py | 1 + prich/constants.py | 2 +- prich/core/engine.py | 381 ++++++++++++++++++++--------- prich/core/utils.py | 27 +- prich/models/template.py | 14 ++ tests/test_engine.py | 2 +- tests/test_run_template.py | 34 +++ 7 files changed, 335 insertions(+), 126 deletions(-) diff --git a/prich/cli/dynamic_command_group.py b/prich/cli/dynamic_command_group.py index 8f3dbcf..b84f999 100644 --- a/prich/cli/dynamic_command_group.py +++ b/prich/cli/dynamic_command_group.py @@ -78,6 +78,7 @@ def create_dynamic_command(config, template: TemplateModel) -> click.Command: click.Option(["-p", "--provider"], type=click.Choice(config.providers.keys()), show_default=True, help="Override LLM provider"), click.Option(["-v", "--verbose"], is_flag=True, default=False, help="Verbose mode"), + click.Option(["-d", "--debug"], is_flag=True, default=False, help="Debug mode"), click.Option(["-q", "--quiet"], is_flag=True, default=False, help="Suppress all output"), click.Option(["-f", "--only-final-output"], is_flag=True, default=False, help="Suppress output and show only the last step output") diff --git a/prich/constants.py b/prich/constants.py index f177501..00399dd 100644 --- a/prich/constants.py +++ b/prich/constants.py @@ -2,7 +2,7 @@ # and are not allowed to be user/defined in the template variables cli_option RESERVED_RUN_TEMPLATE_CLI_OPTIONS = [ "-g", "--global", "-q", "--quiet", "-o", "--output", "-p", "--provider", - "-f", "--only-final-output", "-v", "--verbose" + "-f", "--only-final-output", "-v", "--verbose", "-d", "--debug" ] # .prich folder name diff --git a/prich/core/engine.py b/prich/core/engine.py index 70fe2b6..769e4e1 100644 --- a/prich/core/engine.py +++ b/prich/core/engine.py @@ -1,6 +1,7 @@ import click +import yaml from typing import Dict - +from prich.core.utils import console_print_debug from prich.core.template_utils import should_run_step from prich.core.steps.step_render_template import render_template from prich.core.steps.step_run_command import run_command_step @@ -9,7 +10,7 @@ from prich.models.template import LLMStep, PythonStep, RenderStep, \ CommandStep, ValidateStepOutput from prich.core.utils import console_print, is_quiet, is_only_final_output, \ - is_verbose + is_verbose, models_equal from prich.core.loaders import get_env_vars from prich.core.variable_utils import replace_env_vars, expand_vars @@ -38,140 +39,278 @@ def validate_step_exit_code(validate_step: ValidateStepOutput, exit_code: int, v raise click.ClickException(f"Failed to validate step exit code: {str(e)}") return True +def run_step(step_idx, config, provider, template, variables): + """ Run step during run_template """ + step = template.steps[step_idx-1] + step_return_exit_code = None # Used only for subprocess execute commands + if not is_verbose() and step.name == template.steps[-1].name and step.output_console is None: + # show final step output when non-verbose execution + step.output_console = True + + if is_verbose(): + step_brief = f"\n• Step #{step_idx}: {step.name}" + else: + step_brief = f"• {step.name}" + + should_run = should_run_step(step.when, variables) + if (not should_run and is_verbose()) or should_run: + when_expression = f" (\"when\" expression \"{step.when}\" is {should_run})" if step.when else "" + console_print( + f"[dim]{step_brief}{' - Skipped' if not should_run else ''}{when_expression if is_verbose() else ''}[/dim]") + + # Set output variable to None + if step.output_variable: + variables[step.output_variable] = None + + if not should_run: + return None, None + + output_var = step.output_variable + if isinstance(step, (PythonStep, CommandStep)): + step_output, step_return_exit_code = run_command_step(template, step, variables) + elif isinstance(step, RenderStep): + step_output = render_template(step, variables) + elif isinstance(step, LLMStep): + step_output = send_to_llm(template, step, provider, config, variables) + else: + raise click.ClickException(f"Step {step.type} type is not supported.") + + if is_verbose(): + if step.extract_variables or step.filter: + console_print(f"[dim]Output:\n{step_output}[/dim]") + step.postprocess_extract_vars(out=step_output, variables=variables) + step_output = step.postprocess_filter(out=step_output) + if is_verbose(): + if step.extract_variables: + for spec in step.extract_variables: + console_print( + f"""[dim]Inject {repr(spec.regex)} {f'({len(variables.get(spec.variable))} matches) ' if spec.multiple else ''}→ {spec.variable}: {f'{repr(variables.get(spec.variable))}' if isinstance(variables.get(spec.variable), str) else variables.get(spec.variable)}[/dim]""") + if step.filter: + if step.filter.strip is not None: + console_print(f"[dim]Strip output spaces: {step.filter.strip}[/dim]") + if step.filter.strip_prefix: + console_print(f"[dim]Strip output prefix: \"{step.filter.strip_prefix}\"[/dim]") + if step.filter.slice_start or step.filter.slice_end: + console_print( + f"[dim]Slice output text{f' from {step.filter.slice_start}' if step.filter.slice_start else ''}{f' to {step.filter.slice_end}' if step.filter.slice_end else ''}[/dim]") + if step.filter.regex_extract: + console_print(f"[dim]Apply regex: '{step.filter.regex_extract}'[/dim]") + if step.filter.regex_replace: + replace_details = '\n '.join( + [f"'{x}' → '{y}'" for x, y in step.filter.regex_replace]) + console_print(f"[dim]Apply regex replace: {replace_details}[/dim]") + + # Store last output + last_output = step_output + + if output_var: + variables[output_var] = step_output + if step.output_file: + save_to_file = step.output_file.name + try: + write_mode = step.output_file.mode[:1] if step.output_file.mode else 'w' + with open(save_to_file, write_mode) as step_output_file: + if is_verbose(): + console_print( + f"[dim]{'Save' if write_mode == 'w' else 'Append'} output to file: {save_to_file}[/dim]") + step_output_file.write(step_output) + except Exception as e: + raise click.ClickException(f"Failed to save output to file {save_to_file}: {e}") + # Print step output to console + if ((step.output_console or is_verbose()) and not ( + isinstance(step, LLMStep))) and not is_only_final_output() and not is_quiet(): + console_print(step_output, markup=False) + return last_output, step_return_exit_code + +def run_step_validation(step, step_output, step_return_exit_code, variables): + # Validate step output, used in run_template + skip_following_steps = False + if step.validate_: + if isinstance(step.validate_, ValidateStepOutput): + step_validate = [step.validate_] + else: + step_validate = step.validate_ + idx = 0 + for validate in step_validate: + idx += 1 + validated = True + if isinstance(step, (PythonStep, CommandStep)) and ( + validate.match_exit_code is not None or validate.not_match_exit_code is not None): + validated = validate_step_exit_code(validate, step_return_exit_code, variables) + elif validate.match_exit_code is not None or validate.not_match_exit_code is not None: + raise click.ClickException( + "Step validation using 'match_exitcode' and/or 'not_match_exitcode' supported only in 'python' and 'command' step types.") + if validated: + validated = validate_step_output(validate, step_output, variables) + if not validated: + action = validate.on_fail + failure_msg = validate.message or "Validation failed for step output" + if action == "warn": + console_print(f"[yellow]Warning: {failure_msg}![/yellow]") + elif action == "error": + raise click.ClickException(failure_msg) + elif action == "skip": + console_print(f"[yellow]{failure_msg} – skipping next steps.[/yellow]") + skip_following_steps = True + break + elif action == "continue": + console_print(f"[yellow]{failure_msg} – continue.[/yellow]") + pass + else: + raise click.ClickException(f"Validation type {action} is not supported.") + else: + if is_verbose() and len(step_validate) > 1: + console_print(f"[dim]Validation #{idx} [green]passed[/green].[/dim]") + if is_verbose(): + console_print(f"[dim][green]Validation{'s' if len(step_validate) > 1 else ''} completed.[/green][/dim]") + return skip_following_steps + def run_template(template_id, **kwargs): - from prich.core.loaders import get_loaded_config, get_loaded_template + from pathlib import Path + from pydantic import ValidationError as PydanticValidationError + from prich.core.loaders import get_loaded_config, get_loaded_template, load_template_model, _load_yaml + from prich.cli.validate import template_model_doctor config, _ = get_loaded_config() provider = kwargs.get('provider') output_file = kwargs.get('output') + debug = kwargs.get('debug') template = get_loaded_template(template_id) - variables = {} - for var in template.variables: - cli_option = var.cli_option - if cli_option: - option_name = cli_option.lstrip("-").replace("-", "_") - variables[var.name] = replace_env_vars(kwargs.get(option_name, var.default), get_env_vars()) - else: - variables[var.name] = replace_env_vars(kwargs.get(var.name, var.default), get_env_vars()) - if var.required and variables.get(var.name) is None: - raise click.ClickException(f"Missing required variable {var.name}") + variables = template.prepare_variables(cli_options=kwargs, env_vars=get_env_vars()) + + variables_stash = {} + if debug: + console_print_debug("[dim]Initial variables:[/dim]") + for k, v in variables.items(): + if k != "builtin": + console_print_debug(f"[green]{k}[/green] = [cyan]{v}[/cyan]") if template.steps: - step_return_exit_code = None # Used only for subprocess execute commands - step_idx = 0 + step_idx = 1 last_output = "" + last_step_return_exit_code = None skip_following_steps = False # used with validate - for step in template.steps: - step_idx += 1 - - if not is_verbose() and step.name == template.steps[-1].name and step.output_console is None: - # show final step output when non-verbose execution - step.output_console = True - - if is_verbose(): - step_brief = f"\n• Step #{step_idx}: {step.name}" - else: - step_brief = f"• {step.name}" - should_run = should_run_step(step.when, variables) - if (not should_run and is_verbose()) or should_run: - when_expression = f" (\"when\" expression \"{step.when}\" is {should_run})" if step.when else "" - console_print(f"[dim]{step_brief}{' - Skipped' if not should_run else ''}{when_expression if is_verbose() else ''}[/dim]") - if not should_run: - continue - - # Set output variable to None - if step.output_variable: - variables[step.output_variable] = None - - output_var = step.output_variable - if isinstance(step, (PythonStep, CommandStep)): - step_output, step_return_exit_code = run_command_step(template, step, variables) - elif isinstance(step, RenderStep): - step_output = render_template(step, variables) - elif isinstance(step, LLMStep): - step_output = send_to_llm(template, step, provider, config, variables) - else: - raise click.ClickException(f"Step {step.type} type is not supported.") - - if is_verbose(): - if step.extract_variables or step.filter: - console_print(f"[dim]Output:\n{step_output}[/dim]") - step.postprocess_extract_vars(out=step_output, variables=variables) - step_output = step.postprocess_filter(out=step_output) - if is_verbose(): - if step.extract_variables: - for spec in step.extract_variables: - console_print(f"""[dim]Inject {repr(spec.regex)} {f'({len(variables.get(spec.variable))} matches) ' if spec.multiple else ''}→ {spec.variable}: {f'{repr(variables.get(spec.variable))}' if isinstance(variables.get(spec.variable), str) else variables.get(spec.variable)}[/dim]""") - if step.filter: - if step.filter.strip is not None: - console_print(f"[dim]Strip output spaces: {step.filter.strip}[/dim]") - if step.filter.strip_prefix: - console_print(f"[dim]Strip output prefix: \"{step.filter.strip_prefix}\"[/dim]") - if step.filter.slice_start or step.filter.slice_end: - console_print(f"[dim]Slice output text{f' from {step.filter.slice_start}' if step.filter.slice_start else ''}{f' to {step.filter.slice_end}' if step.filter.slice_end else ''}[/dim]") - if step.filter.regex_extract: - console_print(f"[dim]Apply regex: '{step.filter.regex_extract}'[/dim]") - if step.filter.regex_replace: - replace_details = '\n '.join([f"'{x}' → '{y}'" for x,y in step.filter.regex_replace]) - console_print(f"[dim]Apply regex replace: {replace_details}[/dim]") - - # Store last output - last_output = step_output - - if output_var: - variables[output_var] = step_output - if step.output_file: - save_to_file = step.output_file.name + debug_user_selected_action = None # used with debug + while step_idx <= len(template.steps): + if debug and debug_user_selected_action in ["r", "b", "c", "v"]: + reload_template = None try: - write_mode = step.output_file.mode[:1] if step.output_file.mode else 'w' - with open(save_to_file, write_mode) as step_output_file: - if is_verbose(): - console_print(f"[dim]{'Save' if write_mode == 'w' else 'Append'} output to file: {save_to_file}[/dim]") - step_output_file.write(step_output) + reload_template = load_template_model(Path(template.file)) + except PydanticValidationError as e: + console_print_debug("[red]Exception during template load:[/red]") + found_issues = template_model_doctor(_load_yaml(Path(template.file)), e) + for found_issue in found_issues: + console_print_debug(f"[red]{found_issue}[/red]") except Exception as e: - raise click.ClickException(f"Failed to save output to file {save_to_file}: {e}") - # Print step output to console - if ((step.output_console or is_verbose()) and not (isinstance(step, LLMStep))) and not is_only_final_output() and not is_quiet(): - console_print(step_output, markup=False) - # Validate - if step.validate_: - if isinstance(step.validate_, ValidateStepOutput): - step.validate_ = [step.validate_] - idx = 0 - for validate in step.validate_: - idx += 1 - validated = True - if isinstance(step, (PythonStep, CommandStep)) and (validate.match_exit_code is not None or validate.not_match_exit_code is not None): - validated = validate_step_exit_code(validate, step_return_exit_code, variables) - elif validate.match_exit_code is not None or validate.not_match_exit_code is not None: - raise click.ClickException("Step validation using 'match_exitcode' and/or 'not_match_exitcode' supported only in 'python' and 'command' step types.") - if validated: - validated = validate_step_output(validate, step_output, variables) - if not validated: - action = validate.on_fail - failure_msg = validate.message or "Validation failed for step output" - if action == "warn": - console_print(f"[yellow]Warning: {failure_msg}![/yellow]") - elif action == "error": - raise click.ClickException(failure_msg) - elif action == "skip": - console_print(f"[yellow]{failure_msg} – skipping next steps.[/yellow]") - skip_following_steps = True - break - elif action == "continue": - console_print(f"[yellow]{failure_msg} – continue.[/yellow]") - pass + console_print_debug("[red]Exception during template load:[/red]") + console_print_debug(f"[red]{str(e)}[/red]") + if reload_template is not None and not models_equal(reload_template, template): + console_print_debug("[yellow]Template change detected[/yellow] [dim]- reloading[/dim]") + reload_variables = reload_template.prepare_variables(cli_options=kwargs, env_vars=get_env_vars()) + console_print_debug("[dim]Updating stashed variables[/dim]") + for k,v in variables_stash.items(): + v.update(reload_variables) + console_print_debug("[dim]Updating current variables[/dim]") + variables.update(reload_variables) + console_print_debug("[dim]Updating template[/dim]") + template = reload_template + if debug and debug_user_selected_action in ["r", "b"]: + console_print_debug(f"[dim]Restoring variables from before step {step_idx}[/dim]") + variables = dict(variables_stash[step_idx]) + elif debug and debug_user_selected_action != "v": + console_print_debug(f"[dim]Stash variables before step {step_idx}[/dim]") + variables_stash[step_idx] = dict(variables) + + if not debug or (debug and debug_user_selected_action != "v"): + if debug: + console_print_debug(f"[dim]--- Step {template.steps[step_idx-1].type} [/dim][cyan]{template.steps[step_idx-1].name}[/cyan] #{step_idx}") + output, last_step_return_exit_code = run_step( + step_idx=step_idx, + config=config, + provider=provider, + template=template, + variables=variables + ) + if output is not None: + last_output = output + + try: + skip_following_steps = run_step_validation( + step=template.steps[step_idx-1], + step_output=last_output, + step_return_exit_code=last_step_return_exit_code, + variables=variables + ) + except Exception as e: + if not debug: + raise + else: + console_print_debug(f"[red]Exception[/red] during validation, would be terminated here in non-debug mode: [red]{e}[/red]") + if debug: + debug_user_selected_action = None + while not debug_user_selected_action: + console_print_debug( + f"[white]([cyan]c[/cyan])ontinue, ([cyan]r[/cyan])epeat step{', repeat ([cyan]v[/cyan])alidate' if template.steps[step_idx-1].validate_ else ''}, ([cyan]t[/cyan])erminate{', step ([cyan]b[/cyan])ack' if step_idx > 1 else ''}; show ([cyan]s[/cyan])tep, variab([cyan]l[/cyan])es, variables stas([cyan]h[/cyan]):[/white] ", + end="") + try: + user_input = None + accepted_keys = ['c', 'r', 't', 'l', 's', 'h'] + if step_idx > 1: + accepted_keys.append('b') + if template.steps[step_idx-1].validate_: + accepted_keys.append('v') + while user_input not in accepted_keys: + user_input = click.getchar() # returns a 1-char string + # user_input = input() # used for debug + except KeyboardInterrupt: + user_input = "t" + console_print(f"[cyan]{user_input}[/cyan]") + if user_input in ["t", "terminate", "quit", "exit"]: + console_print("[red]Terminated during debug pause.[/red]") + exit(1) + elif step_idx > 1 and user_input == "b": + console_print_debug(f"[dim]Drop variables stash from step {step_idx}[/dim]") + variables_stash.pop(step_idx) + step_idx -= 1 + console_print_debug(f"[dim]Return to step {step_idx}[/dim]") + debug_user_selected_action = "b" + elif user_input == "c": + debug_user_selected_action = "c" + elif user_input == "s": + console_print_debug(f"[dim]Step #{step_idx}:[/dim]") + tpl_model_dump = template.steps[step_idx-1].model_dump(exclude_none=True) + if tpl_model_dump.get("validate_"): + tpl_model_dump["validate"] = tpl_model_dump.get("validate_") + tpl_model_dump.pop("validate_") + console_print(f"[cyan]{yaml.safe_dump(tpl_model_dump, sort_keys=False)}[/cyan]") + elif user_input == "r": + debug_user_selected_action = "r" + elif user_input == "v": + debug_user_selected_action = "v" + elif user_input == "l": + if not variables: + console_print_debug("[green]no variables[/green]") else: - raise click.ClickException(f"Validation type {action} is not supported.") - else: - if is_verbose() and len(step.validate_) > 1: - console_print(f"[dim]Validation #{idx} [green]passed[/green].[/dim]") - if is_verbose(): - console_print(f"[dim][green]Validation{'s' if len(step.validate_) > 1 else ''} completed.[/green][/dim]") - if skip_following_steps: # used with validate when skip matched - break + console_print_debug("[dim]Variables:[/dim]") + for k,v in variables.items(): + if k != "builtin": + console_print_debug(f"[green]{k}[/green] = [cyan]{v}[/cyan]") + elif user_input == "h": + console_print_debug(f"[dim]Stashed variables[/dim]") + for stash, stash_vars in variables_stash.items(): + console_print_debug(f"[dim]before step {template.steps[stash-1].type} {template.steps[stash-1].name} #{stash}:[/dim]") + for k,v in stash_vars.items(): + if k != "builtin": + console_print_debug(f"[green]{k}[/green] = [cyan]{v}[/cyan]") + + if skip_following_steps and not debug_user_selected_action in ["r", "b", "v"]: # used with validate when skip matched + break + + if debug_user_selected_action not in ["r", "b", "v"]: + step_idx += 1 + # Save last step output if output file option added if output_file: with open(output_file, 'w') as final_output_file: diff --git a/prich/core/utils.py b/prich/core/utils.py index 8f1247f..65e652f 100644 --- a/prich/core/utils.py +++ b/prich/core/utils.py @@ -63,14 +63,18 @@ def is_print_enabled() -> bool: def is_piped() -> bool: """ Check if prich executed with a piped command (should work only when not executed from pytest) """ # TODO: revisit, we need to allow executions from templates for example - return not console.is_terminal and not os.getenv("PYTEST_CURRENT_TEST") - # return False + # return not console.is_terminal and not os.getenv("PYTEST_CURRENT_TEST") + return False -def console_print(message: str = "", end: str = "\n", markup = None, flush: bool = None): +def console_print(message: str = "", end: str = "\n", markup = None): """ Print to console wrapper """ if is_print_enabled(): console.print(message, end=end, markup=markup, crop=False) +def console_print_debug(message: str = "", end: str = "\n", markup = None): + """ Print to console wrapper with debug prefix """ + console_print(f"[[yellow]debug[/yellow]] {message}", end=end, markup=markup) + def is_valid_template_id(template_id) -> bool: """ Validate Name Pattern: lowercase letters, numbers, hyphen, optional underscores, and no other characters""" pattern = r'^[a-z0-9-]+(_[a-z0-9-]+)*$' @@ -133,3 +137,20 @@ def is_just_filename(filename: Path | str): if s in ("", ".", "..") or ("/" in s) or ("\\" in s): return False return True + +def models_equal(a, b, *, + exclude_unset=False, + exclude_none=False, + by_alias=False, + strict_type=True): + if strict_type and type(a) is not type(b): + return False + return ( + a.model_dump(exclude_unset=exclude_unset, + exclude_none=exclude_none, + by_alias=by_alias) + == + b.model_dump(exclude_unset=exclude_unset, + exclude_none=exclude_none, + by_alias=by_alias) + ) diff --git a/prich/models/template.py b/prich/models/template.py index 62e1bff..7943d75 100644 --- a/prich/models/template.py +++ b/prich/models/template.py @@ -271,6 +271,20 @@ def str_presenter(dumper, data): with open(template_file, "w") as f: f.write(yaml.safe_dump(model_dict, sort_keys=False, width=float("inf"))) + def prepare_variables(self, cli_options, env_vars) -> dict: + from prich.core.variable_utils import replace_env_vars + variables = {} + for var in self.variables: + cli_option = var.cli_option + if cli_option: + option_name = cli_option.lstrip("-").replace("-", "_") + variables[var.name] = replace_env_vars(cli_options.get(option_name, var.default), env_vars) + else: + variables[var.name] = replace_env_vars(cli_options.get(var.name, var.default), env_vars) + if var.required and variables.get(var.name) is None: + raise click.ClickException(f"Missing required variable {var.name}") + return variables + def describe(self): return f""" Template: {self.id} diff --git a/tests/test_engine.py b/tests/test_engine.py index 070f17e..b7d0c97 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -445,7 +445,7 @@ def test_validate_step_output(mock_paths, basic_config, case): }, ] @pytest.mark.parametrize("case", get_run_command_step_CASES, ids=[c["id"] for c in get_run_command_step_CASES]) -def test_run_command_step(case, monkeypatch): +def test_run_command_step(case, mock_paths, monkeypatch): from prich.core.steps.step_run_command import run_command_step if case.get("mock_output"): diff --git a/tests/test_run_template.py b/tests/test_run_template.py index 9464519..184d979 100644 --- a/tests/test_run_template.py +++ b/tests/test_run_template.py @@ -928,9 +928,40 @@ def test_run_template(case, monkeypatch, basic_config): "expected_regex_output": ["^### System(?:.|\\n)+### Assistant:\\n$"]}, {"id": "run_local_template_id_quiet", "add_template": True, "args": ["template-local", "--quiet"], "expected_regex_output": ["^$"]}, + {"id": "run_local_template_id_debug_continue", "add_template": True, "args": ["template-local", "--debug"], + "expected_regex_output": [ + "\\[debug\\] Initial variables:", + "name = Assistant", + "test_output = This is my text", + "Stash variables before step 1", + "--- Step llm llm step #1", + "\\[debug\\] \\(c\\)ontinue, \\(r\\)epeat step", + ], "key_press": ["c"]}, + {"id": "run_local_template_id_debug_repeat_continue", "add_template": True, "args": ["template-local", "--debug"], + "expected_regex_output": [ + ": r\n\\[debug\\] Template change detected - reloading", + "stas\\(h\\): c" + ], "key_press": ["r", "c"]}, + {"id": "run_local_template_id_debug_list_vars", "add_template": True, "args": ["template-local", "--debug"], + "expected_regex_output": [ + "stas\\(h\\): l\n\\[debug\\] Variables:\n\\[debug\\]" + ], "key_press": ["l"]}, + {"id": "run_local_template_id_debug_list_var_hashes", "add_template": True, "args": ["template-local", "--debug"], + "expected_regex_output": [ + "stas\\(h\\): h\n\\[debug\\] Stashed variables\n\\[debug\\]" + ], "key_press": ["h"]}, + {"id": "run_local_template_id_debug_show_step", "add_template": True, "args": ["template-local", "--debug"], + "expected_regex_output": [ + "stas\\(h\\): s\n\\[debug\\] Step #1:\nname: llm step\n" + ], "key_press": ["s"]}, ] @pytest.mark.parametrize("case", get_run_template_cli_CASES, ids=[c["id"] for c in get_run_template_cli_CASES]) def test_run_template_cli(mock_paths, monkeypatch, case, template, basic_config): + def get_key_press(count): + res = case.get("key_press")[count["count"]] + count["count"] += 1 + return res + global_dir = mock_paths.home_dir local_dir = mock_paths.cwd_dir @@ -939,6 +970,9 @@ def test_run_template_cli(mock_paths, monkeypatch, case, template, basic_config) monkeypatch.setattr("prich.core.loaders.get_cwd_dir", lambda: local_dir) monkeypatch.setattr("prich.core.loaders.get_home_dir", lambda: global_dir) + if case.get("key_press"): + count = {"count": 0} + monkeypatch.setattr("click.getchar", lambda: get_key_press(count)) local_config = basic_config.model_copy(deep=True) global_config = basic_config.model_copy(deep=True)