forked from HAL9000/cleveragents-core
freemo
31472b5413
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
370 lines
12 KiB
Python
370 lines
12 KiB
Python
# pyright: reportRedeclaration=false
|
|
"""Step definitions for actor_context_coverage_r3.feature.
|
|
|
|
Covers uncovered lines in actor_context.py: 41, 47, 53-58, 151-152,
|
|
155-160, 201-202, 209-210, 386, 390, 392, 395-396, 402.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import yaml
|
|
from behave import given, then, when
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.actor_context import (
|
|
_context_size_kb,
|
|
_default_context_base,
|
|
_list_context_names,
|
|
)
|
|
from cleveragents.cli.commands.actor_context import (
|
|
app as actor_context_app,
|
|
)
|
|
from cleveragents.reactive.context_manager import ContextManager
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("accov3 a temporary context directory")
|
|
def step_accov3_temp_dir(context):
|
|
context.accov3_tmp = Path(tempfile.mkdtemp())
|
|
context.accov3_ctx_dir = context.accov3_tmp / "contexts"
|
|
context.accov3_ctx_dir.mkdir(parents=True, exist_ok=True)
|
|
context.accov3_runner = CliRunner()
|
|
context.accov3_result = None
|
|
context.accov3_helper_result = None
|
|
context.accov3_import_file = None
|
|
context.add_cleanup(lambda: shutil.rmtree(context.accov3_tmp, ignore_errors=True))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Givens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('accov3 a context named "{name}" exists')
|
|
def step_accov3_context_exists(context, name):
|
|
mgr = ContextManager(name, context.accov3_ctx_dir)
|
|
mgr.add_message("system", f"Seed message for {name}")
|
|
|
|
|
|
@given('accov3 a context named "{name}" with some file content')
|
|
def step_accov3_context_with_content(context, name):
|
|
mgr = ContextManager(name, context.accov3_ctx_dir)
|
|
mgr.add_message("user", "Hello, this is filler content for size testing.")
|
|
mgr.add_message("assistant", "Acknowledged. More filler to ensure non-zero size.")
|
|
# Store the manager for later retrieval
|
|
context.accov3_size_mgr = mgr
|
|
|
|
|
|
@given('accov3 a valid YAML context file named "{filename}"')
|
|
def step_accov3_yaml_file(context, filename):
|
|
"""Create a .yaml file containing JSON-compatible data.
|
|
|
|
The file uses JSON syntax which is valid YAML. This triggers the
|
|
YAML-suffix branch (line 386) in context_import, and also allows
|
|
ContextManager.import_context (which uses json.load) to succeed.
|
|
"""
|
|
data = {
|
|
"context_name": "yamlctx",
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": "yaml test message",
|
|
"timestamp": "2026-01-01T00:00:00",
|
|
"metadata": {},
|
|
},
|
|
],
|
|
"metadata": {"context_name": "yamlctx"},
|
|
"state": {},
|
|
"global_context": {},
|
|
}
|
|
context.accov3_import_file = context.accov3_tmp / filename
|
|
# Write as JSON (which is valid YAML) so import_context can also read it
|
|
context.accov3_import_file.write_text(
|
|
json.dumps(data, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
@given("accov3 a .json file containing valid YAML but invalid JSON")
|
|
def step_accov3_bad_json_yaml_content(context):
|
|
"""Create a .json file with YAML-only syntax.
|
|
|
|
The import command will fail to json.loads() this (line 389-390),
|
|
then fallback to yaml.safe_load() at line 392. We mock
|
|
ContextManager.import_context so it doesn't also try json.load.
|
|
"""
|
|
yaml_content = (
|
|
"context_name: fallback\n"
|
|
"messages:\n"
|
|
" - role: user\n"
|
|
" content: yaml-only message\n"
|
|
" timestamp: '2026-01-01T00:00:00'\n"
|
|
" metadata: {}\n"
|
|
"metadata:\n"
|
|
" context_name: fallback\n"
|
|
"state: {}\n"
|
|
"global_context: {}\n"
|
|
)
|
|
context.accov3_import_file = context.accov3_tmp / "bad.json"
|
|
context.accov3_import_file.write_text(yaml_content, encoding="utf-8")
|
|
|
|
|
|
@given("accov3 a .json file containing a JSON list")
|
|
def step_accov3_json_list_file(context):
|
|
context.accov3_import_file = context.accov3_tmp / "listdata.json"
|
|
context.accov3_import_file.write_text(
|
|
json.dumps([1, 2, 3]),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
@given('accov3 a .json file named "{filename}" without context_name key')
|
|
def step_accov3_json_no_context_name(context, filename):
|
|
data = {
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": "no name test",
|
|
"timestamp": "2026-01-01T00:00:00",
|
|
"metadata": {},
|
|
},
|
|
],
|
|
"metadata": {},
|
|
"state": {},
|
|
"global_context": {},
|
|
}
|
|
context.accov3_import_file = context.accov3_tmp / filename
|
|
context.accov3_import_file.write_text(
|
|
json.dumps(data, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When — helpers called directly
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("accov3 I call _default_context_base with None")
|
|
def step_accov3_default_base_none(context):
|
|
try:
|
|
context.accov3_helper_result = _default_context_base(None)
|
|
except Exception as exc:
|
|
context.accov3_error = exc
|
|
|
|
|
|
@when("accov3 I call _list_context_names with a non-existent path")
|
|
def step_accov3_list_nonexistent(context):
|
|
fake_path = context.accov3_tmp / "does_not_exist"
|
|
try:
|
|
context.accov3_helper_result = _list_context_names(fake_path)
|
|
except Exception as exc:
|
|
context.accov3_error = exc
|
|
|
|
|
|
@when('accov3 I call _context_size_kb for "{name}"')
|
|
def step_accov3_context_size(context, name):
|
|
mgr = context.accov3_size_mgr
|
|
try:
|
|
context.accov3_helper_result = _context_size_kb(mgr)
|
|
except Exception as exc:
|
|
context.accov3_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When — CLI remove
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("accov3 I run remove --all --yes on empty context dir")
|
|
def step_accov3_remove_all_empty(context):
|
|
context.accov3_result = context.accov3_runner.invoke(
|
|
actor_context_app,
|
|
["remove", "--all", "--yes", "--context-dir", str(context.accov3_ctx_dir)],
|
|
)
|
|
|
|
|
|
@when("accov3 I run remove --all without --yes and answer no")
|
|
def step_accov3_remove_all_cancel(context):
|
|
context.accov3_result = context.accov3_runner.invoke(
|
|
actor_context_app,
|
|
["remove", "--all", "--context-dir", str(context.accov3_ctx_dir)],
|
|
input="n\n",
|
|
)
|
|
|
|
|
|
@when('accov3 I run remove "{name}" without --yes and answer no')
|
|
def step_accov3_remove_single_cancel(context, name):
|
|
context.accov3_result = context.accov3_runner.invoke(
|
|
actor_context_app,
|
|
["remove", name, "--context-dir", str(context.accov3_ctx_dir)],
|
|
input="n\n",
|
|
)
|
|
|
|
|
|
@when('accov3 I run remove "{name}" with --yes')
|
|
def step_accov3_remove_single_yes(context, name):
|
|
context.accov3_result = context.accov3_runner.invoke(
|
|
actor_context_app,
|
|
["remove", name, "--yes", "--context-dir", str(context.accov3_ctx_dir)],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When — CLI import
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('accov3 I run import "{name}" from the YAML file')
|
|
def step_accov3_import_yaml(context, name):
|
|
context.accov3_result = context.accov3_runner.invoke(
|
|
actor_context_app,
|
|
[
|
|
"import",
|
|
name,
|
|
"--input",
|
|
str(context.accov3_import_file),
|
|
"--context-dir",
|
|
str(context.accov3_ctx_dir),
|
|
],
|
|
)
|
|
|
|
|
|
@when('accov3 I run import "{name}" from the bad-json file')
|
|
def step_accov3_import_bad_json(context, name):
|
|
"""Import a .json file whose content is YAML-only (not valid JSON).
|
|
|
|
We mock ContextManager.import_context so it handles the YAML file
|
|
properly — the real import_context hard-codes json.load. The lines
|
|
under test (390, 392) are in the CLI command's file parsing, not in
|
|
ContextManager.import_context.
|
|
"""
|
|
|
|
def _yaml_import_context(self, context_file):
|
|
"""A replacement import_context that uses yaml.safe_load."""
|
|
text = context_file.read_text(encoding="utf-8")
|
|
data = yaml.safe_load(text) or {}
|
|
self.messages = data.get("messages", [])
|
|
self.metadata = data.get("metadata", {})
|
|
self.state = data.get("state", {})
|
|
self.global_context = data.get("global_context", {})
|
|
self.save()
|
|
|
|
patcher = patch.object(ContextManager, "import_context", _yaml_import_context)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
context.accov3_result = context.accov3_runner.invoke(
|
|
actor_context_app,
|
|
[
|
|
"import",
|
|
name,
|
|
"--input",
|
|
str(context.accov3_import_file),
|
|
"--context-dir",
|
|
str(context.accov3_ctx_dir),
|
|
],
|
|
)
|
|
|
|
|
|
@when('accov3 I run import "{name}" from the list file')
|
|
def step_accov3_import_list(context, name):
|
|
context.accov3_result = context.accov3_runner.invoke(
|
|
actor_context_app,
|
|
[
|
|
"import",
|
|
name,
|
|
"--input",
|
|
str(context.accov3_import_file),
|
|
"--context-dir",
|
|
str(context.accov3_ctx_dir),
|
|
],
|
|
)
|
|
|
|
|
|
@when("accov3 I run import without name from the stemname file")
|
|
def step_accov3_import_stemname(context):
|
|
context.accov3_result = context.accov3_runner.invoke(
|
|
actor_context_app,
|
|
[
|
|
"import",
|
|
"--input",
|
|
str(context.accov3_import_file),
|
|
"--context-dir",
|
|
str(context.accov3_ctx_dir),
|
|
],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then — helper assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("accov3 the result should be the home cleveragents context path")
|
|
def step_accov3_check_home_path(context):
|
|
expected = Path.home() / ".cleveragents" / "context"
|
|
assert context.accov3_helper_result == expected, (
|
|
f"Expected {expected}, got {context.accov3_helper_result}"
|
|
)
|
|
|
|
|
|
@then("accov3 the result should be an empty list")
|
|
def step_accov3_check_empty_list(context):
|
|
assert context.accov3_helper_result == [], (
|
|
f"Expected [], got {context.accov3_helper_result}"
|
|
)
|
|
|
|
|
|
@then("accov3 the size should be a non-negative number")
|
|
def step_accov3_check_size(context):
|
|
result = context.accov3_helper_result
|
|
assert isinstance(result, (int, float)), f"Expected number, got {type(result)}"
|
|
assert result >= 0, f"Expected non-negative, got {result}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then — CLI result assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("accov3 the result exit code should be 0")
|
|
def step_accov3_exit_0(context):
|
|
r = context.accov3_result
|
|
assert r.exit_code == 0, (
|
|
f"Expected exit 0, got {r.exit_code}.\n"
|
|
f"output: {r.output}\n"
|
|
f"exception: {r.exception!r}"
|
|
)
|
|
|
|
|
|
@then("accov3 the result exit code should be 1")
|
|
def step_accov3_exit_1(context):
|
|
r = context.accov3_result
|
|
assert r.exit_code == 1, (
|
|
f"Expected exit 1, got {r.exit_code}.\n"
|
|
f"output: {r.output}\n"
|
|
f"exception: {r.exception!r}"
|
|
)
|
|
|
|
|
|
@then('accov3 the output should contain "{text}"')
|
|
def step_accov3_output_contains(context, text):
|
|
r = context.accov3_result
|
|
assert text in r.output, f"Expected output to contain {text!r}, got:\n{r.output}"
|
|
|
|
|
|
@then('accov3 the context "{name}" should still exist')
|
|
def step_accov3_context_still_exists(context, name):
|
|
ctx_path = context.accov3_ctx_dir / name
|
|
assert ctx_path.exists(), f"Context dir {ctx_path} does not exist"
|