aditya
b8f9da4cca
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 16s
CI / build (pull_request) Successful in 28s
CI / quality (pull_request) Successful in 35s
CI / security (pull_request) Successful in 43s
CI / typecheck (pull_request) Successful in 45s
CI / unit_tests (pull_request) Successful in 3m4s
CI / integration_tests (pull_request) Successful in 3m43s
CI / e2e_tests (pull_request) Successful in 3m48s
CI / docker (pull_request) Successful in 55s
CI / coverage (pull_request) Successful in 6m54s
CI / benchmark-regression (pull_request) Successful in 38m2s
Wrap `PersonaRegistry.delete()` with the personas file lock and use race-safe `unlink` handling for concurrent save/delete consistency.
540 lines
20 KiB
Python
540 lines
20 KiB
Python
"""Behave steps for REPL input mode behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
from rich.console import Console
|
|
|
|
from cleveragents.cli.commands.repl import (
|
|
_best_fuzzy_match,
|
|
_build_reference_catalog,
|
|
_expand_references,
|
|
_find_reference_candidates,
|
|
_handle_slash_command,
|
|
_parse_persona_create,
|
|
_reference_suggestions,
|
|
_ReplSessionState,
|
|
_run_shell_command,
|
|
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 rich REPL reference fixture")
|
|
def step_rich_repl_reference_fixture(context: Context) -> None:
|
|
fixture_dir = Path(tempfile.mkdtemp())
|
|
(fixture_dir / "README.md").write_text("# fixture\n", encoding="utf-8")
|
|
(fixture_dir / "examples" / "actors").mkdir(parents=True, exist_ok=True)
|
|
(fixture_dir / "examples" / "tools").mkdir(parents=True, exist_ok=True)
|
|
(fixture_dir / "examples" / "skills").mkdir(parents=True, exist_ok=True)
|
|
(fixture_dir / "examples" / "actors" / "review.yaml").write_text(
|
|
"name: review\n", encoding="utf-8"
|
|
)
|
|
(fixture_dir / "examples" / "tools" / "search.yml").write_text(
|
|
"name: search\n", encoding="utf-8"
|
|
)
|
|
(fixture_dir / "examples" / "skills" / "guide").mkdir(parents=True, exist_ok=True)
|
|
(fixture_dir / "examples" / "skills" / "skill.yml").write_text(
|
|
"name: skill\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
|
|
|
|
|
|
@when('I parse persona create tokens "{command}"')
|
|
def step_parse_persona_create_tokens(context: Context, command: str) -> None:
|
|
context.repl_parse_error = None
|
|
context.repl_parsed_persona = None
|
|
text = command[1:] if command.startswith("/") else command
|
|
tokens = text.split()
|
|
try:
|
|
context.repl_parsed_persona = _parse_persona_create(tokens)
|
|
except Exception as exc:
|
|
context.repl_parse_error = exc
|
|
|
|
|
|
@then('parsed persona name should be "{name}"')
|
|
def step_parsed_persona_name(context: Context, name: str) -> None:
|
|
assert context.repl_parsed_persona is not None
|
|
assert context.repl_parsed_persona.name == name
|
|
|
|
|
|
@then('parsed persona actor should be "{actor}"')
|
|
def step_parsed_persona_actor(context: Context, actor: str) -> None:
|
|
assert context.repl_parsed_persona is not None
|
|
assert context.repl_parsed_persona.actor == actor
|
|
|
|
|
|
@then("parsed persona cycle order should be {cycle:d}")
|
|
def step_parsed_persona_cycle(context: Context, cycle: int) -> None:
|
|
assert context.repl_parsed_persona is not None
|
|
assert context.repl_parsed_persona.cycle_order == cycle
|
|
|
|
|
|
@then('persona create parse should fail with "{message}"')
|
|
def step_parse_persona_create_fails(context: Context, message: str) -> None:
|
|
assert context.repl_parse_error is not None
|
|
assert message in str(context.repl_parse_error)
|
|
|
|
|
|
def _run_shell_helper_with_capture(command: str) -> tuple[int, str]:
|
|
output = StringIO()
|
|
console = Console(file=output, no_color=True, width=180)
|
|
with patch("cleveragents.cli.commands.repl._console", console):
|
|
code = _run_shell_command(command)
|
|
return code, output.getvalue()
|
|
|
|
|
|
@when('I run REPL shell helper with command "{command}"')
|
|
def step_run_shell_helper(context: Context, command: str) -> None:
|
|
code, output = _run_shell_helper_with_capture(command)
|
|
context.repl_shell_helper_code = code
|
|
context.repl_shell_helper_output = output
|
|
|
|
|
|
@when("I run REPL shell helper with empty command")
|
|
def step_run_shell_helper_empty(context: Context) -> None:
|
|
code, output = _run_shell_helper_with_capture("")
|
|
context.repl_shell_helper_code = code
|
|
context.repl_shell_helper_output = output
|
|
|
|
|
|
@when('I run REPL shell helper with command "{command}" and shell disabled')
|
|
def step_run_shell_helper_disabled(context: Context, command: str) -> None:
|
|
with patch.dict(os.environ, {"CLEVERAGENTS_DISABLE_SHELL_MODE": "true"}):
|
|
code, output = _run_shell_helper_with_capture(command)
|
|
context.repl_shell_helper_code = code
|
|
context.repl_shell_helper_output = output
|
|
|
|
|
|
@when('I run REPL shell helper with command "{command}" and subprocess timeout')
|
|
def step_run_shell_helper_timeout(context: Context, command: str) -> None:
|
|
output = StringIO()
|
|
console = Console(file=output, no_color=True, width=180)
|
|
with (
|
|
patch("cleveragents.cli.commands.repl._console", console),
|
|
patch(
|
|
"cleveragents.cli.commands.repl.subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(command, 30),
|
|
),
|
|
):
|
|
context.repl_shell_helper_code = _run_shell_command(command)
|
|
context.repl_shell_helper_output = output.getvalue()
|
|
|
|
|
|
@when(
|
|
'I run REPL shell helper with command "{command}" and subprocess stderr "{stderr}" code {code:d}'
|
|
)
|
|
def step_run_shell_helper_stderr(
|
|
context: Context, command: str, stderr: str, code: int
|
|
) -> None:
|
|
output = StringIO()
|
|
console = Console(file=output, no_color=True, width=180)
|
|
proc = subprocess.CompletedProcess(
|
|
args=command, returncode=code, stdout="", stderr=stderr
|
|
)
|
|
with (
|
|
patch("cleveragents.cli.commands.repl._console", console),
|
|
patch("cleveragents.cli.commands.repl.subprocess.run", return_value=proc),
|
|
):
|
|
context.repl_shell_helper_code = _run_shell_command(command)
|
|
context.repl_shell_helper_output = output.getvalue()
|
|
|
|
|
|
@then("REPL shell helper exit code should be {code:d}")
|
|
def step_shell_helper_exit_code(context: Context, code: int) -> None:
|
|
assert context.repl_shell_helper_code == code
|
|
|
|
|
|
@then('REPL shell helper output should contain "{text}"')
|
|
def step_shell_helper_output_contains(context: Context, text: str) -> None:
|
|
assert text in context.repl_shell_helper_output
|
|
|
|
|
|
class _SlashRegistryStub:
|
|
def __init__(self) -> None:
|
|
self.personas: dict[str, PersonaConfig] = {}
|
|
self.active_name = "default"
|
|
|
|
def list_personas(self) -> list[PersonaConfig]:
|
|
return list(self.personas.values())
|
|
|
|
def get_persona(self, _name: str) -> PersonaConfig | None:
|
|
return self.personas.get(_name)
|
|
|
|
def set_active_persona_name(self, name: str) -> None:
|
|
self.active_name = name
|
|
|
|
def save_persona(self, persona: PersonaConfig) -> None:
|
|
self.personas[persona.name] = persona
|
|
|
|
def delete_persona(self, name: str) -> bool:
|
|
return self.personas.pop(name, None) is not None
|
|
|
|
def resolve_import_path(self, path: Path) -> Path:
|
|
return path
|
|
|
|
|
|
@given("slash command test sessions")
|
|
def step_slash_test_sessions(context: Context) -> None:
|
|
context.slash_registry = _SlashRegistryStub()
|
|
context.slash_sessions = {
|
|
"default": _ReplSessionState(name="default", active_persona="default")
|
|
}
|
|
context.slash_current_session = "default"
|
|
|
|
|
|
def _handle_slash_with_capture(
|
|
slash_text: str,
|
|
*,
|
|
registry: Any,
|
|
sessions: Any,
|
|
current_session: str,
|
|
input_value: str | None = None,
|
|
) -> tuple[int, str, str]:
|
|
output = StringIO()
|
|
console = Console(file=output, no_color=True, width=180)
|
|
with patch("cleveragents.cli.commands.repl._console", console):
|
|
if input_value is None:
|
|
code, new_session = _handle_slash_command(
|
|
slash_text,
|
|
registry=registry,
|
|
sessions=sessions,
|
|
current_session=current_session,
|
|
)
|
|
else:
|
|
with patch("builtins.input", return_value=input_value):
|
|
code, new_session = _handle_slash_command(
|
|
slash_text,
|
|
registry=registry,
|
|
sessions=sessions,
|
|
current_session=current_session,
|
|
)
|
|
return code, new_session, output.getvalue()
|
|
|
|
|
|
@when('I handle slash text with parser error "{slash_text}"')
|
|
def step_handle_slash_parser_error(context: Context, slash_text: str) -> None:
|
|
code, new_session, output = _handle_slash_with_capture(
|
|
slash_text,
|
|
registry=context.slash_registry,
|
|
sessions=context.slash_sessions,
|
|
current_session=context.slash_current_session,
|
|
)
|
|
context.slash_handler_code = code
|
|
context.slash_handler_session = new_session
|
|
context.slash_handler_output = output
|
|
|
|
|
|
@when("I handle slash text that raises parser error")
|
|
def step_handle_slash_parser_error_forced(context: Context) -> None:
|
|
output = StringIO()
|
|
console = Console(file=output, no_color=True, width=180)
|
|
with (
|
|
patch("cleveragents.cli.commands.repl._console", console),
|
|
patch(
|
|
"cleveragents.cli.commands.repl.shlex.split",
|
|
side_effect=ValueError("forced parser failure"),
|
|
),
|
|
):
|
|
code, new_session = _handle_slash_command(
|
|
"persona test",
|
|
registry=context.slash_registry,
|
|
sessions=context.slash_sessions,
|
|
current_session=context.slash_current_session,
|
|
)
|
|
context.slash_handler_code = code
|
|
context.slash_handler_session = new_session
|
|
context.slash_handler_output = output.getvalue()
|
|
|
|
|
|
@when('I handle slash text "{slash_text}"')
|
|
def step_handle_slash_text(context: Context, slash_text: str) -> None:
|
|
code, new_session, output = _handle_slash_with_capture(
|
|
slash_text,
|
|
registry=context.slash_registry,
|
|
sessions=context.slash_sessions,
|
|
current_session=context.slash_current_session,
|
|
)
|
|
context.slash_handler_code = code
|
|
context.slash_handler_session = new_session
|
|
context.slash_handler_output = output
|
|
|
|
|
|
@when('I handle picker slash text "{slash_text}" using input "{input_value}"')
|
|
def step_handle_slash_text_with_input(
|
|
context: Context, slash_text: str, input_value: str
|
|
) -> None:
|
|
code, new_session, output = _handle_slash_with_capture(
|
|
slash_text,
|
|
registry=context.slash_registry,
|
|
sessions=context.slash_sessions,
|
|
current_session=context.slash_current_session,
|
|
input_value=input_value,
|
|
)
|
|
context.slash_handler_code = code
|
|
context.slash_handler_session = new_session
|
|
context.slash_handler_output = output
|
|
|
|
|
|
@when("I handle empty slash text")
|
|
def step_handle_empty_slash_text(context: Context) -> None:
|
|
code, new_session, output = _handle_slash_with_capture(
|
|
"",
|
|
registry=context.slash_registry,
|
|
sessions=context.slash_sessions,
|
|
current_session=context.slash_current_session,
|
|
)
|
|
context.slash_handler_code = code
|
|
context.slash_handler_session = new_session
|
|
context.slash_handler_output = output
|
|
|
|
|
|
@then("slash handler exit code should be {code:d}")
|
|
def step_slash_handler_code(context: Context, code: int) -> None:
|
|
assert context.slash_handler_code == code
|
|
|
|
|
|
@then('slash handler output should contain "{text}"')
|
|
def step_slash_handler_output_contains(context: Context, text: str) -> None:
|
|
assert text in context.slash_handler_output
|
|
|
|
|
|
@given("slash command test registry with personas")
|
|
def step_slash_registry_with_personas(context: Context) -> None:
|
|
context.slash_registry.save_persona(
|
|
PersonaConfig(name="dev", actor="local/mock-default")
|
|
)
|
|
context.slash_registry.save_persona(
|
|
PersonaConfig(name="reviewer", actor="local/mock-default")
|
|
)
|
|
|
|
|
|
@when("I build REPL reference catalog from fixture")
|
|
def step_build_repl_reference_catalog_fixture(context: Context) -> None:
|
|
context.repl_reference_catalog = {}
|
|
outside_root = Path("/tmp/repl-outside-root")
|
|
|
|
def fake_walk(_cwd: Path, followlinks: bool = False):
|
|
yield (
|
|
str(context.repl_reference_fixture_dir),
|
|
["examples", ".git"],
|
|
["README.md"],
|
|
)
|
|
yield str(outside_root), [], ["outside.txt"]
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.repl.Path.cwd",
|
|
return_value=context.repl_reference_fixture_dir,
|
|
),
|
|
patch("cleveragents.cli.commands.repl.os.walk", side_effect=fake_walk),
|
|
patch.dict(
|
|
os.environ,
|
|
{"CLEVERAGENTS_PROJECT": "local/proj", "CLEVERAGENTS_PLAN": "local/plan"},
|
|
clear=False,
|
|
),
|
|
):
|
|
context.repl_reference_catalog = _build_reference_catalog()
|
|
|
|
|
|
@then("REPL catalog should include fixture actor tool skill and env entries")
|
|
def step_catalog_includes_fixture_entries(context: Context) -> None:
|
|
catalog = context.repl_reference_catalog
|
|
assert "local/review" in catalog["actor"]
|
|
assert "local/search" in catalog["tool"]
|
|
assert "local/guide" in catalog["skill"]
|
|
assert "local/skill" in catalog["skill"]
|
|
assert catalog["project"] == ["local/proj"]
|
|
assert catalog["plan"] == ["local/plan"]
|
|
|
|
|
|
@then("REPL catalog should include non-relative walk file path")
|
|
def step_catalog_has_non_relative_path(context: Context) -> None:
|
|
assert any(
|
|
value.endswith("/outside.txt")
|
|
for value in context.repl_reference_catalog["file"]
|
|
)
|
|
|
|
|
|
@when("I evaluate REPL fuzzy and suggestion helpers")
|
|
def step_eval_repl_fuzzy_helpers(context: Context) -> None:
|
|
catalog = {
|
|
"file": ["README.md", "docs/guide.md"],
|
|
"actor": ["local/reviewer"],
|
|
"tool": ["local/searcher"],
|
|
"skill": ["local/guide"],
|
|
"project": ["local/proj"],
|
|
"plan": ["local/plan"],
|
|
}
|
|
context.repl_helper_checks = [
|
|
_best_fuzzy_match("", ["x"]) is None,
|
|
_best_fuzzy_match("REA", ["README.md"]) == "README.md",
|
|
_best_fuzzy_match("guide", ["docs/guide.md"]) == "docs/guide.md",
|
|
_best_fuzzy_match("reviwer", ["local/reviewer"]) == "local/reviewer",
|
|
_find_reference_candidates("", ["a", "b"], limit=1) == ["a"],
|
|
_expand_references(
|
|
"echo \\@literal @file:README @zzz", catalog=catalog
|
|
).startswith("echo @literal @file:"),
|
|
len(_reference_suggestions("inspect @file:READ", catalog=catalog)) > 0,
|
|
len(_reference_suggestions("inspect @rev", catalog=catalog)) > 0,
|
|
_reference_suggestions("inspect \\@file:README", catalog=catalog) == [],
|
|
]
|
|
|
|
|
|
@then("REPL helper evaluation should pass")
|
|
def step_repl_helper_eval_pass(context: Context) -> None:
|
|
assert all(context.repl_helper_checks)
|