Files
cleveragents-core/features/steps/repl_input_modes_steps.py
T
aditya d9e51d98f8 fix(tui,cli,tests): harden persona/input modes and stabilize parallel test execution
- Add shell safety controls for REPL/TUI (`looks_dangerous`, confirmation gate, timeout handling, and env-based shell disable guard).
- Secure persona workflows with strict name/path validation, safe import/export resolution, atomic+locked registry writes, and malformed YAML resilience.
- Unify persona models and wiring by reusing canonical TUI schema/registry, adding DI providers, and lazy-loading TUI exports to avoid circular imports.
- Improve reference discovery with ignored-directory filtering, symlink-safe walking, and TTL caching for CLI/TUI reference catalogs.
- Expand Behave/Robot coverage for safety/error paths and parallel-isolation behavior; add shared `features/mocks/fake_repl_input.py` helper.
- Fix parallel-run flakiness via deterministic cleanup/reload patterns and watchdog polling fallback when inotify limits are reached.

ISSUES CLOSED: #695
2026-03-18 10:09:06 +00:00

158 lines
5.9 KiB
Python

"""Behave steps for REPL input mode behavior."""
from __future__ import annotations
import os
import shutil
import tempfile
from pathlib import Path
from unittest.mock import patch
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.cli.commands.repl import _expand_references, run_repl
from cleveragents.cli.persona import PersonaConfig
from features.mocks.fake_repl_input import FakeReplInput
def _run_repl_capture(lines: list[str], config_dir: Path) -> tuple[int, str]:
from io import StringIO
from rich.console import Console
output = StringIO()
console = Console(file=output, no_color=True, width=180)
fake_input = FakeReplInput(lines)
with (
patch("cleveragents.cli.commands.repl._console", console),
patch("builtins.input", fake_input),
patch.dict(os.environ, {"CLEVERAGENTS_CONFIG_DIR": str(config_dir)}),
):
code = run_repl(no_history=True)
return code, output.getvalue()
@given("a temporary REPL config directory")
def step_temp_repl_config(context: Context) -> None:
config_dir = Path(tempfile.mkdtemp())
context.repl_mode_config_dir = config_dir
context.add_cleanup(lambda: shutil.rmtree(str(config_dir), ignore_errors=True))
@given("a deterministic REPL reference catalog fixture")
def step_repl_reference_fixture(context: Context) -> None:
fixture_dir = Path(tempfile.mkdtemp())
(fixture_dir / "README.md").write_text("# fixture\n", encoding="utf-8")
context.repl_reference_fixture_dir = fixture_dir
context.add_cleanup(lambda: shutil.rmtree(str(fixture_dir), ignore_errors=True))
@given("a temporary persona export path")
def step_temp_persona_export_path(context: Context) -> None:
export_path = Path("tmp-persona-export.yaml")
context.repl_mode_export_path = export_path
context.add_cleanup(lambda: export_path.unlink(missing_ok=True))
@given("an absolute persona export path")
def step_absolute_persona_export_path(context: Context) -> None:
export_path = Path(tempfile.mkdtemp()) / "persona-export.yaml"
context.repl_mode_export_path = export_path
context.add_cleanup(
lambda: shutil.rmtree(str(export_path.parent), ignore_errors=True)
)
@given("an invalid persona import file path")
def step_invalid_persona_import_file(context: Context) -> None:
invalid_path = Path("tmp-invalid-persona.yaml")
invalid_path.write_text("- bad\n- yaml\n", encoding="utf-8")
context.repl_mode_invalid_import_path = invalid_path
context.add_cleanup(lambda: invalid_path.unlink(missing_ok=True))
@given("an absolute persona import file path")
def step_absolute_persona_import_path(context: Context) -> None:
temp_dir = Path(tempfile.mkdtemp())
input_path = temp_dir / "persona-import.yaml"
input_path.write_text("name: x\nactor: local/mock-default\n", encoding="utf-8")
context.repl_mode_absolute_import_path = input_path
context.add_cleanup(lambda: shutil.rmtree(str(temp_dir), ignore_errors=True))
@when("I run the REPL with input lines")
def step_run_repl_lines(context: Context) -> None:
replacements: dict[str, str] = {}
export_path = getattr(context, "repl_mode_export_path", None)
if export_path is not None:
replacements["export_path"] = str(export_path)
invalid_import_path = getattr(context, "repl_mode_invalid_import_path", None)
if invalid_import_path is not None:
replacements["invalid_import_path"] = str(invalid_import_path)
absolute_import_path = getattr(context, "repl_mode_absolute_import_path", None)
if absolute_import_path is not None:
replacements["absolute_import_path"] = str(absolute_import_path)
lines = []
for row in context.table:
line = row["line"]
if replacements:
line = line.format(**replacements)
lines.append(line)
code, output = _run_repl_capture(lines, context.repl_mode_config_dir)
context.repl_mode_code = code
context.repl_mode_output = output
@then('the REPL mode output should contain "{text}"')
def step_output_contains(context: Context, text: str) -> None:
assert text in context.repl_mode_output, (
f"Expected {text!r} in output, got:\n{context.repl_mode_output}"
)
@when('I expand REPL references for "{line}"')
def step_expand_references(context: Context, line: str) -> None:
fixture_dir = getattr(context, "repl_reference_fixture_dir", None)
if fixture_dir is None:
context.expanded_repl_line = _expand_references(line)
return
with patch("cleveragents.cli.commands.repl.Path.cwd", return_value=fixture_dir):
context.expanded_repl_line = _expand_references(line)
@then('the expanded REPL line should contain "{text}"')
def step_expanded_contains(context: Context, text: str) -> None:
assert text in context.expanded_repl_line, (
f"Expected {text!r} in expanded line: {context.expanded_repl_line}"
)
@when('I validate CLI persona name "{name}"')
def step_validate_cli_persona_name(context: Context, name: str) -> None:
decoded = name.encode("utf-8").decode("unicode_escape")
context.repl_cli_persona_error = None
try:
PersonaConfig(name=decoded, actor="local/mock-default")
except ValidationError as exc:
context.repl_cli_persona_error = exc
@then('CLI persona validation should fail with "{message}"')
def step_cli_persona_validation_fails(context: Context, message: str) -> None:
assert context.repl_cli_persona_error is not None
assert message in str(context.repl_cli_persona_error)
@when('I validate CLI persona payload with color "{color}"')
def step_validate_cli_persona_payload_with_color(context: Context, color: str) -> None:
context.repl_cli_persona_payload = PersonaConfig.model_validate(
{"name": "colorful", "actor": "local/mock-default", "color": color}
)
@then('CLI persona color should be "{color}"')
def step_cli_persona_color(context: Context, color: str) -> None:
assert context.repl_cli_persona_payload.color == color