forked from HAL9000/cleveragents-core
Luis Mendes
ab911dbdc4
The format_output() function returned a string that callers passed to Rich console.print(), which wraps long lines at terminal width. This injected literal newline characters into JSON string values (e.g. in definition_of_done fields), producing invalid JSON that downstream parsers could not decode (JSONDecodeError: Invalid control character). For machine-readable formats (json, yaml, plain), format_output() now writes the rendered output directly to sys.stdout and returns an empty string. This preserves the exact serialization from json.dumps/ yaml.dump without Rich text processing artifacts. Refs: #746
867 lines
28 KiB
Python
867 lines
28 KiB
Python
"""Step definitions for security_template_coverage_boost.feature.
|
|
|
|
Covers uncovered paths in skills/registry.py, cli/formatting.py,
|
|
cli/commands/system.py, cli/commands/session.py, cli/commands/cleanup.py,
|
|
cli/commands/audit.py, and application/container.py.
|
|
|
|
All step patterns use unique prefixes to avoid AmbiguousStep collisions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import os
|
|
import tempfile
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import typer
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
__all__: list[str] = []
|
|
|
|
|
|
# ===========================================================================
|
|
# Helpers
|
|
# ===========================================================================
|
|
|
|
|
|
def _make_mock_tool_registry(tools: dict[str, Any] | None = None) -> Any:
|
|
"""Return a mock tool registry that resolves tool names."""
|
|
reg = MagicMock()
|
|
tools = tools or {}
|
|
|
|
def _get_tool(name: str) -> Any:
|
|
return tools.get(name)
|
|
|
|
reg.get_tool = _get_tool
|
|
return reg
|
|
|
|
|
|
def _make_skill_definition(
|
|
name: str = "test/skill",
|
|
description: str = "A test skill",
|
|
tool_refs: list[str] | None = None,
|
|
inline_tools: list[Any] | None = None,
|
|
includes: list[Any] | None = None,
|
|
) -> Any:
|
|
"""Build a minimal SkillDefinition for registry validation."""
|
|
from cleveragents.domain.models.core.skill import Skill
|
|
from cleveragents.skills.protocol import SkillDefinition, SkillMetadata
|
|
|
|
skill = Skill(
|
|
name=name,
|
|
description=description,
|
|
tool_refs=tool_refs or [],
|
|
anonymous_tools=inline_tools or [],
|
|
includes=includes or [],
|
|
)
|
|
metadata = SkillMetadata(
|
|
name=name,
|
|
description=description,
|
|
tool_count=len(tool_refs or []) + len(inline_tools or []),
|
|
)
|
|
return SkillDefinition(skill=skill, metadata=metadata)
|
|
|
|
|
|
def _mock_settings(**overrides: Any) -> Any:
|
|
"""Create mock settings with sensible defaults for CLI tests."""
|
|
tmpdir = tempfile.mkdtemp()
|
|
s = MagicMock()
|
|
s.database_url = overrides.get("database_url", f"sqlite:///{tmpdir}/test.db")
|
|
s.data_dir = overrides.get("data_dir", Path(tmpdir))
|
|
s.storage_path = overrides.get("storage_path", Path(tmpdir))
|
|
s.config_path = overrides.get("config_path", Path(tmpdir) / "config.toml")
|
|
s.log_dir = overrides.get("log_dir", Path(tmpdir) / "logs")
|
|
s.default_automation_profile = "auto"
|
|
s.has_provider_configured = MagicMock(return_value=False)
|
|
s.configured_providers = []
|
|
s.debug_enabled = False
|
|
return s
|
|
|
|
|
|
# ===========================================================================
|
|
# skills/registry.py — validate_skill (lines 200-213)
|
|
# ===========================================================================
|
|
|
|
|
|
@given("a skill registry with a mock tool registry")
|
|
def step_skill_reg_with_tools(context: Context) -> None:
|
|
from cleveragents.skills.registry import SkillRegistry
|
|
|
|
context.cov_tool_registry = _make_mock_tool_registry(tools={})
|
|
context.cov_skill_registry = SkillRegistry(
|
|
tool_registry=context.cov_tool_registry,
|
|
)
|
|
context.cov_validation_errors = []
|
|
|
|
|
|
@given("a skill registry with no tool registry")
|
|
def step_skill_reg_no_tools(context: Context) -> None:
|
|
from cleveragents.skills.registry import SkillRegistry
|
|
|
|
context.cov_skill_registry = SkillRegistry(tool_registry=None)
|
|
context.cov_validation_errors = []
|
|
|
|
|
|
@given('a skill definition with tool ref "{ref}"')
|
|
def step_skill_def_tool_ref(context: Context, ref: str) -> None:
|
|
context.cov_skill_def = _make_skill_definition(tool_refs=[ref])
|
|
|
|
|
|
@given("a skill definition with an inline tool missing its description")
|
|
def step_skill_def_inline_no_desc(context: Context) -> None:
|
|
from cleveragents.domain.models.core.skill import SkillInlineTool
|
|
from cleveragents.domain.models.core.tool import ToolSource
|
|
|
|
# Use model_construct to bypass pydantic min-length validation
|
|
inline = SkillInlineTool.model_construct(
|
|
description="",
|
|
source=ToolSource.CUSTOM,
|
|
timeout=300,
|
|
resource_slots=[],
|
|
)
|
|
context.cov_skill_def = _make_skill_definition(inline_tools=[inline])
|
|
|
|
|
|
@given('a skill definition that includes "{include_name}"')
|
|
def step_skill_def_include(context: Context, include_name: str) -> None:
|
|
from cleveragents.domain.models.core.skill import SkillInclude
|
|
|
|
inc = SkillInclude(name=include_name)
|
|
context.cov_skill_def = _make_skill_definition(includes=[inc])
|
|
|
|
|
|
@given("a skill definition with no tool refs or includes")
|
|
def step_skill_def_empty(context: Context) -> None:
|
|
context.cov_skill_def = _make_skill_definition()
|
|
|
|
|
|
@when("I validate the skill definition via the registry")
|
|
def step_validate_skill(context: Context) -> None:
|
|
context.cov_validation_errors = context.cov_skill_registry.validate_skill(
|
|
context.cov_skill_def,
|
|
)
|
|
|
|
|
|
@then('the skill validation errors should mention "{text}"')
|
|
def step_skill_val_mention(context: Context, text: str) -> None:
|
|
combined = " ".join(context.cov_validation_errors)
|
|
assert text in combined, (
|
|
f"Expected '{text}' in validation errors: {context.cov_validation_errors}"
|
|
)
|
|
|
|
|
|
@then("the skill validation errors should be empty")
|
|
def step_skill_val_empty(context: Context) -> None:
|
|
assert context.cov_validation_errors == [], (
|
|
f"Expected no errors, got: {context.cov_validation_errors}"
|
|
)
|
|
|
|
|
|
# ===========================================================================
|
|
# cli/formatting.py — format_output, format_output_session (lines 121,128,197,231-236)
|
|
# ===========================================================================
|
|
|
|
|
|
@given('sample formatting data with key "{key}" and value "{value}"')
|
|
def step_fmt_data_dict(context: Context, key: str, value: str) -> None:
|
|
context.cov_format_data = {key: value}
|
|
context.cov_format_output = ""
|
|
context.cov_session_output = ""
|
|
|
|
|
|
@given("sample formatting data as a list of two items")
|
|
def step_fmt_data_list(context: Context) -> None:
|
|
context.cov_format_data = [{"a": "1", "b": "2"}, {"a": "3", "b": "4"}]
|
|
context.cov_format_output = ""
|
|
context.cov_session_output = ""
|
|
|
|
|
|
@given("sample formatting data as a list with mismatched keys")
|
|
def step_fmt_data_mismatched(context: Context) -> None:
|
|
context.cov_format_data = [
|
|
{"col_a": "val1"},
|
|
{"col_a": "val2", "extra_col": "val3"},
|
|
]
|
|
context.cov_format_output = ""
|
|
context.cov_session_output = ""
|
|
|
|
|
|
@given("sample formatting data as an empty list")
|
|
def step_fmt_data_empty(context: Context) -> None:
|
|
context.cov_format_data = []
|
|
context.cov_format_output = ""
|
|
context.cov_session_output = ""
|
|
|
|
|
|
@when('I format the data with format type "{fmt}"')
|
|
def step_fmt_format(context: Context, fmt: str) -> None:
|
|
from contextlib import redirect_stdout
|
|
from io import StringIO
|
|
|
|
from cleveragents.cli.formatting import format_output
|
|
|
|
buf = StringIO()
|
|
with redirect_stdout(buf):
|
|
result = format_output(context.cov_format_data, fmt)
|
|
context.cov_format_output = result or buf.getvalue().rstrip("\n")
|
|
|
|
|
|
@when('I format the data via format_output_session with format "{fmt}"')
|
|
def step_fmt_session(context: Context, fmt: str) -> None:
|
|
from cleveragents.cli.formatting import format_output_session
|
|
|
|
context.cov_session_output = format_output_session(context.cov_format_data, fmt)
|
|
|
|
|
|
@then("the formatted output should contain {text}")
|
|
def step_fmt_output_contains(context: Context, text: str) -> None:
|
|
text = text.strip('"')
|
|
assert text in context.cov_format_output, (
|
|
f"Expected '{text}' in output:\n{context.cov_format_output}"
|
|
)
|
|
|
|
|
|
@then('the formatted output should be "{expected}"')
|
|
def step_fmt_output_exact(context: Context, expected: str) -> None:
|
|
assert context.cov_format_output.strip() == expected, (
|
|
f"Expected '{expected}', got: '{context.cov_format_output.strip()}'"
|
|
)
|
|
|
|
|
|
@then("the formatted session output should not be empty")
|
|
def step_fmt_session_not_empty(context: Context) -> None:
|
|
assert len(context.cov_session_output.strip()) > 0
|
|
|
|
|
|
@then('the formatted session output should contain "{text}"')
|
|
def step_fmt_session_contains(context: Context, text: str) -> None:
|
|
assert text in context.cov_session_output, (
|
|
f"Expected '{text}' in session output:\n{context.cov_session_output}"
|
|
)
|
|
|
|
|
|
# ===========================================================================
|
|
# cli/commands/system.py — health check functions (lines 108-118 etc.)
|
|
# ===========================================================================
|
|
|
|
|
|
@given("a system health check environment")
|
|
def step_sys_health_env(context: Context) -> None:
|
|
context.cov_health_result = {}
|
|
context.cov_info_data = {}
|
|
context.cov_tmpdir = tempfile.mkdtemp()
|
|
|
|
|
|
@given("a config file that exists and is readable")
|
|
def step_sys_config_exists(context: Context) -> None:
|
|
config_path = Path(context.cov_tmpdir) / "config.toml"
|
|
config_path.write_text("[settings]\nkey = 'value'\n")
|
|
context.cov_config_path = config_path
|
|
|
|
|
|
@given("a data dir that does not exist")
|
|
def step_sys_data_dir_missing(context: Context) -> None:
|
|
context.cov_data_dir = Path(context.cov_tmpdir) / "nonexistent_data_dir"
|
|
|
|
|
|
@given("a writable data dir")
|
|
def step_sys_writable_data_dir(context: Context) -> None:
|
|
data_dir = Path(context.cov_tmpdir) / "data"
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
context.cov_data_dir = data_dir
|
|
|
|
|
|
@given("a database file that exists")
|
|
def step_sys_db_exists(context: Context) -> None:
|
|
db_path = Path(context.cov_tmpdir) / "test.db"
|
|
db_path.write_bytes(b"\x00" * 1024)
|
|
context.cov_db_path = db_path
|
|
|
|
|
|
@given("a log directory with files")
|
|
def step_sys_log_dir(context: Context) -> None:
|
|
log_dir = Path(context.cov_tmpdir) / "logs"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
(log_dir / "test.log").write_text("log content\n")
|
|
context.cov_log_dir = log_dir
|
|
|
|
|
|
@when("I run the config file health check")
|
|
def step_sys_check_config(context: Context) -> None:
|
|
from cleveragents.cli.commands.system import _check_config_file
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"CLEVERAGENTS_CONFIG_PATH": str(context.cov_config_path)},
|
|
):
|
|
context.cov_health_result = _check_config_file()
|
|
|
|
|
|
@when("I run the data dir health check")
|
|
def step_sys_check_data_dir(context: Context) -> None:
|
|
from cleveragents.cli.commands.system import _check_data_dir
|
|
|
|
ms = _mock_settings(data_dir=context.cov_data_dir)
|
|
with patch("cleveragents.config.settings.get_settings", return_value=ms):
|
|
context.cov_health_result = _check_data_dir()
|
|
|
|
|
|
@when("I run the disk space health check")
|
|
def step_sys_check_disk(context: Context) -> None:
|
|
from cleveragents.cli.commands.system import _check_disk_space
|
|
|
|
context.cov_health_result = _check_disk_space()
|
|
|
|
|
|
@when("I run the git availability health check")
|
|
def step_sys_check_git(context: Context) -> None:
|
|
from cleveragents.cli.commands.system import _check_git
|
|
|
|
context.cov_health_result = _check_git()
|
|
|
|
|
|
@when("I run the file permissions health check")
|
|
def step_sys_check_perms(context: Context) -> None:
|
|
from cleveragents.cli.commands.system import _check_file_permissions
|
|
|
|
data_dir = getattr(context, "cov_data_dir", Path(context.cov_tmpdir))
|
|
ms = _mock_settings(data_dir=data_dir)
|
|
with patch("cleveragents.config.settings.get_settings", return_value=ms):
|
|
context.cov_health_result = _check_file_permissions()
|
|
|
|
|
|
@when("I build info data")
|
|
def step_sys_build_info(context: Context) -> None:
|
|
from cleveragents.cli.commands.system import build_info_data
|
|
|
|
db_path = getattr(context, "cov_db_path", Path(context.cov_tmpdir) / "test.db")
|
|
log_dir = getattr(context, "cov_log_dir", Path(context.cov_tmpdir) / "logs")
|
|
|
|
ms = _mock_settings(
|
|
database_url=f"sqlite:///{db_path}",
|
|
log_dir=log_dir,
|
|
)
|
|
with patch("cleveragents.config.settings.get_settings", return_value=ms):
|
|
context.cov_info_data = build_info_data()
|
|
|
|
|
|
@then('the health check status should be "{expected}"')
|
|
def step_sys_health_status(context: Context, expected: str) -> None:
|
|
from cleveragents.cli.commands.system import CheckStatus
|
|
|
|
status_map = {
|
|
"ok": CheckStatus.OK,
|
|
"warn": CheckStatus.WARN,
|
|
"error": CheckStatus.ERROR,
|
|
}
|
|
assert context.cov_health_result["status"] == status_map[expected], (
|
|
f"Expected '{expected}', got: {context.cov_health_result['status']}"
|
|
)
|
|
|
|
|
|
@then('the health check details should be "{expected}"')
|
|
def step_sys_health_details(context: Context, expected: str) -> None:
|
|
assert context.cov_health_result["details"] == expected, (
|
|
f"Expected '{expected}', got: '{context.cov_health_result['details']}'"
|
|
)
|
|
|
|
|
|
@then('the health check details should contain "{text}"')
|
|
def step_sys_health_details_contains(context: Context, text: str) -> None:
|
|
details = context.cov_health_result["details"]
|
|
assert text in details, f"Expected '{text}' in: '{details}'"
|
|
|
|
|
|
@then("the info data should contain a db_size value")
|
|
def step_sys_info_db_size(context: Context) -> None:
|
|
assert "storage" in context.cov_info_data
|
|
assert "db_size" in context.cov_info_data["storage"]
|
|
assert context.cov_info_data["storage"]["db_size"] != "unknown"
|
|
|
|
|
|
@then("the info data should contain a logs value")
|
|
def step_sys_info_logs(context: Context) -> None:
|
|
assert "storage" in context.cov_info_data
|
|
assert "logs" in context.cov_info_data["storage"]
|
|
|
|
|
|
# ===========================================================================
|
|
# cli/commands/session.py — session CLI (lines 56-69, 75, 157-159, etc.)
|
|
# ===========================================================================
|
|
|
|
|
|
@given("a session CLI test environment")
|
|
def step_sess_env(context: Context) -> None:
|
|
context.cov_sess_output = ""
|
|
context.cov_sess_error = None
|
|
context.cov_sess_aborted = False
|
|
context.cov_sess_exit_code = 0
|
|
context.cov_sess_tmpdir = tempfile.mkdtemp()
|
|
|
|
|
|
@given("a mock session service that returns a created session")
|
|
def step_sess_mock_create(context: Context) -> None:
|
|
from datetime import UTC, datetime
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.session_id = "01TEST000000000000000000001"
|
|
mock_session.actor_name = "test-actor"
|
|
mock_session.namespace = "default"
|
|
mock_session.created_at = datetime(2026, 1, 1, tzinfo=UTC)
|
|
mock_session.updated_at = datetime(2026, 1, 1, tzinfo=UTC)
|
|
mock_session.message_count = 0
|
|
mock_session.messages = []
|
|
mock_session.linked_plan_ids = []
|
|
|
|
svc = MagicMock()
|
|
svc.create.return_value = mock_session
|
|
context.cov_mock_sess_svc = svc
|
|
|
|
|
|
@given("a mock session service that raises SessionNotFoundError on create")
|
|
def step_sess_mock_create_err(context: Context) -> None:
|
|
from cleveragents.domain.models.core.session import SessionNotFoundError
|
|
|
|
svc = MagicMock()
|
|
svc.create.side_effect = SessionNotFoundError("test-id")
|
|
context.cov_mock_sess_svc = svc
|
|
|
|
|
|
@given("a mock session service that returns no sessions")
|
|
def step_sess_mock_list_empty(context: Context) -> None:
|
|
svc = MagicMock()
|
|
svc.list.return_value = []
|
|
context.cov_mock_sess_svc = svc
|
|
|
|
|
|
@given("a mock session service that returns a session")
|
|
def step_sess_mock_get(context: Context) -> None:
|
|
from datetime import UTC, datetime
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.session_id = "01TEST000000000000000000001"
|
|
mock_session.actor_name = "test-actor"
|
|
mock_session.namespace = "default"
|
|
mock_session.created_at = datetime(2026, 1, 1, tzinfo=UTC)
|
|
mock_session.updated_at = datetime(2026, 1, 1, tzinfo=UTC)
|
|
mock_session.message_count = 0
|
|
mock_session.messages = []
|
|
mock_session.linked_plan_ids = []
|
|
|
|
svc = MagicMock()
|
|
svc.get.return_value = mock_session
|
|
svc.delete.return_value = None
|
|
context.cov_mock_sess_svc = svc
|
|
|
|
|
|
@given("a mock session service that can export")
|
|
def step_sess_mock_export(context: Context) -> None:
|
|
svc = MagicMock()
|
|
svc.export_session.return_value = {
|
|
"session_id": "01TEST000000000000000000001",
|
|
"messages": [],
|
|
}
|
|
context.cov_mock_sess_svc = svc
|
|
|
|
|
|
@given("a temporary output file that already exists")
|
|
def step_sess_temp_file(context: Context) -> None:
|
|
out_path = Path(context.cov_sess_tmpdir) / "existing_export.json"
|
|
out_path.write_text("{}")
|
|
context.cov_output_path = out_path
|
|
|
|
|
|
@when("I call get_session_service")
|
|
def step_sess_get_service(context: Context) -> None:
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
mock_container = MagicMock()
|
|
mock_db = MagicMock()
|
|
mock_container.db.return_value = mock_db
|
|
|
|
original_service = session_mod._service
|
|
try:
|
|
session_mod._service = None
|
|
with (
|
|
patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=mock_container,
|
|
),
|
|
patch(
|
|
"cleveragents.application.services.session_service.PersistentSessionService",
|
|
) as mock_cls,
|
|
):
|
|
mock_cls.return_value = MagicMock()
|
|
result = session_mod._get_session_service()
|
|
context.cov_sess_svc_result = result
|
|
finally:
|
|
session_mod._service = original_service
|
|
|
|
|
|
@when("I call reset_session_service")
|
|
def step_sess_reset_service(context: Context) -> None:
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
original = session_mod._service
|
|
session_mod._service = MagicMock() # set something
|
|
session_mod._reset_session_service()
|
|
context.cov_sess_cache_cleared = session_mod._service is None
|
|
session_mod._service = original # restore
|
|
|
|
|
|
@when('I invoke session create with format "{fmt}"')
|
|
def step_sess_create_fmt(context: Context, fmt: str) -> None:
|
|
from contextlib import redirect_stdout
|
|
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
buf = StringIO()
|
|
with patch.object(
|
|
session_mod,
|
|
"_get_session_service",
|
|
return_value=context.cov_mock_sess_svc,
|
|
):
|
|
try:
|
|
with (
|
|
patch("typer.echo", side_effect=lambda x: buf.write(str(x))),
|
|
redirect_stdout(buf),
|
|
):
|
|
session_mod.create(fmt=fmt)
|
|
except SystemExit:
|
|
pass
|
|
context.cov_sess_output = buf.getvalue()
|
|
|
|
|
|
@when("I invoke session create expecting an error")
|
|
def step_sess_create_err(context: Context) -> None:
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
with patch.object(
|
|
session_mod,
|
|
"_get_session_service",
|
|
return_value=context.cov_mock_sess_svc,
|
|
):
|
|
try:
|
|
session_mod.create(fmt="json")
|
|
except (SystemExit, Exception) as exc:
|
|
context.cov_sess_error = exc
|
|
context.cov_sess_exit_code = (
|
|
getattr(exc, "code", 1) if isinstance(exc, SystemExit) else 1
|
|
)
|
|
|
|
|
|
@when('I invoke session list with format "{fmt}"')
|
|
def step_sess_list_fmt(context: Context, fmt: str) -> None:
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
buf = StringIO()
|
|
with (
|
|
patch.object(
|
|
session_mod,
|
|
"_get_session_service",
|
|
return_value=context.cov_mock_sess_svc,
|
|
),
|
|
patch.object(
|
|
session_mod.console,
|
|
"print",
|
|
side_effect=lambda x="", **kw: buf.write(str(x) + "\n"),
|
|
),
|
|
):
|
|
session_mod.list_sessions(fmt=fmt)
|
|
context.cov_sess_output = buf.getvalue()
|
|
|
|
|
|
@when("I invoke session delete without confirming")
|
|
def step_sess_delete_no_confirm(context: Context) -> None:
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
with patch.object(
|
|
session_mod,
|
|
"_get_session_service",
|
|
return_value=context.cov_mock_sess_svc,
|
|
):
|
|
try:
|
|
with patch("typer.confirm", return_value=False):
|
|
session_mod.delete(
|
|
session_id="01TEST000000000000000000001",
|
|
yes=False,
|
|
)
|
|
except (typer.Abort, SystemExit):
|
|
context.cov_sess_aborted = True
|
|
|
|
|
|
@when("I invoke session delete with yes flag")
|
|
def step_sess_delete_yes(context: Context) -> None:
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
buf_lines: list[str] = []
|
|
with (
|
|
patch.object(
|
|
session_mod,
|
|
"_get_session_service",
|
|
return_value=context.cov_mock_sess_svc,
|
|
),
|
|
patch.object(
|
|
session_mod.console,
|
|
"print",
|
|
side_effect=lambda x="", **kw: buf_lines.append(str(x)),
|
|
),
|
|
):
|
|
session_mod.delete(
|
|
session_id="01TEST000000000000000000001",
|
|
yes=True,
|
|
)
|
|
context.cov_sess_output = "\n".join(buf_lines)
|
|
|
|
|
|
@when("I invoke session export to stdout")
|
|
def step_sess_export_stdout(context: Context) -> None:
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
buf = StringIO()
|
|
with (
|
|
patch.object(
|
|
session_mod,
|
|
"_get_session_service",
|
|
return_value=context.cov_mock_sess_svc,
|
|
),
|
|
patch("typer.echo", side_effect=lambda x: buf.write(str(x))),
|
|
):
|
|
session_mod.export_session(
|
|
session_id="01TEST000000000000000000001",
|
|
output=None,
|
|
force=False,
|
|
)
|
|
context.cov_sess_output = buf.getvalue()
|
|
|
|
|
|
@when("I invoke session export to that file without force")
|
|
def step_sess_export_no_force(context: Context) -> None:
|
|
from cleveragents.cli.commands import session as session_mod
|
|
|
|
with patch.object(
|
|
session_mod,
|
|
"_get_session_service",
|
|
return_value=context.cov_mock_sess_svc,
|
|
):
|
|
try:
|
|
session_mod.export_session(
|
|
session_id="01TEST000000000000000000001",
|
|
output=context.cov_output_path,
|
|
force=False,
|
|
)
|
|
except (SystemExit, Exception) as exc:
|
|
context.cov_sess_exit_code = getattr(exc, "code", 1)
|
|
context.cov_sess_error = exc
|
|
|
|
|
|
@then("a session service should be returned")
|
|
def step_sess_svc_returned(context: Context) -> None:
|
|
assert context.cov_sess_svc_result is not None
|
|
|
|
|
|
@then("the session service cache should be cleared")
|
|
def step_sess_cache_cleared(context: Context) -> None:
|
|
assert context.cov_sess_cache_cleared is True
|
|
|
|
|
|
@then('the covboost session output should contain "{text}"')
|
|
def step_sess_output_has(context: Context, text: str) -> None:
|
|
assert text in context.cov_sess_output, (
|
|
f"Expected '{text}' in:\n{context.cov_sess_output}"
|
|
)
|
|
|
|
|
|
@then("the session CLI should have exited with error")
|
|
def step_sess_exit_err(context: Context) -> None:
|
|
assert context.cov_sess_error is not None or context.cov_sess_exit_code != 0, (
|
|
"Expected an error exit"
|
|
)
|
|
|
|
|
|
@then("the session CLI should have been aborted")
|
|
def step_sess_aborted(context: Context) -> None:
|
|
assert context.cov_sess_aborted is True, "Expected abort"
|
|
|
|
|
|
# ===========================================================================
|
|
# cli/commands/cleanup.py — error handling (lines 49-58, 94)
|
|
# ===========================================================================
|
|
|
|
|
|
@given("a cleanup CLI test environment")
|
|
def step_clean_env(context: Context) -> None:
|
|
context.cov_clean_result = None
|
|
context.cov_active_plans = frozenset()
|
|
|
|
|
|
@given("a cleanup service that raises OSError on sandbox scan")
|
|
def step_clean_oserror(context: Context) -> None:
|
|
svc = MagicMock()
|
|
svc._get_sandbox_dirs.side_effect = OSError("Permission denied")
|
|
context.cov_clean_svc = svc
|
|
|
|
|
|
@given("a mock cleanup service")
|
|
def step_clean_mock_svc(context: Context) -> None:
|
|
svc = MagicMock()
|
|
svc._get_sandbox_dirs.return_value = []
|
|
svc.extract_plan_id_from_sandbox.return_value = None
|
|
|
|
report = MagicMock()
|
|
report.stale_items = []
|
|
report.purged_items = []
|
|
report.total_bytes_freed = 0
|
|
svc.scan.return_value = report
|
|
svc.purge.return_value = report
|
|
context.cov_clean_svc = svc
|
|
|
|
|
|
@when("I detect active plans")
|
|
def step_clean_detect(context: Context) -> None:
|
|
from cleveragents.cli.commands.cleanup import _detect_active_plan_ids
|
|
|
|
context.cov_active_plans = _detect_active_plan_ids(context.cov_clean_svc)
|
|
|
|
|
|
@when("I invoke cleanup prune")
|
|
def step_clean_prune(context: Context) -> None:
|
|
from cleveragents.cli.commands.cleanup import scan
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.cleanup._get_cleanup_service",
|
|
return_value=context.cov_clean_svc,
|
|
),
|
|
contextlib.suppress(SystemExit),
|
|
):
|
|
scan()
|
|
context.cov_clean_result = "ok"
|
|
|
|
|
|
@then("the active plans set should be empty")
|
|
def step_clean_empty(context: Context) -> None:
|
|
assert context.cov_active_plans == frozenset(), (
|
|
f"Expected empty, got: {context.cov_active_plans}"
|
|
)
|
|
|
|
|
|
@then("the cleanup should have run without error")
|
|
def step_clean_ok(context: Context) -> None:
|
|
assert context.cov_clean_result == "ok"
|
|
|
|
|
|
# ===========================================================================
|
|
# cli/commands/audit.py — audit service creation (lines 29-32, 110-111)
|
|
# ===========================================================================
|
|
|
|
|
|
@given("an audit CLI test environment")
|
|
def step_audit_env(context: Context) -> None:
|
|
context.cov_audit_svc = None
|
|
context.cov_audit_output = ""
|
|
|
|
|
|
@given("a mock audit service with no entries")
|
|
def step_audit_mock_empty(context: Context) -> None:
|
|
svc = MagicMock()
|
|
svc.list_entries.return_value = []
|
|
svc.__enter__ = MagicMock(return_value=svc)
|
|
svc.__exit__ = MagicMock(return_value=False)
|
|
context.cov_mock_audit_svc = svc
|
|
|
|
|
|
@when("I create the audit service")
|
|
def step_audit_create(context: Context) -> None:
|
|
from cleveragents.cli.commands.audit import _get_audit_service
|
|
|
|
ms = _mock_settings()
|
|
with (
|
|
patch("cleveragents.config.settings.get_settings", return_value=ms),
|
|
patch(
|
|
"cleveragents.application.services.audit_service.AuditService.__init__",
|
|
return_value=None,
|
|
),
|
|
):
|
|
context.cov_audit_svc = _get_audit_service()
|
|
|
|
|
|
@when("I invoke audit list")
|
|
def step_audit_list(context: Context) -> None:
|
|
from cleveragents.cli.commands.audit import list_entries
|
|
|
|
buf = StringIO()
|
|
console_mock = MagicMock()
|
|
console_mock.print = lambda x="", **kw: buf.write(str(x) + "\n")
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.audit._get_audit_service",
|
|
return_value=context.cov_mock_audit_svc,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.audit.get_console",
|
|
return_value=console_mock,
|
|
),
|
|
):
|
|
list_entries()
|
|
context.cov_audit_output = buf.getvalue()
|
|
|
|
|
|
@then("an audit service instance should be returned")
|
|
def step_audit_returned(context: Context) -> None:
|
|
assert context.cov_audit_svc is not None
|
|
|
|
|
|
@then('the covboost audit output should contain "{text}"')
|
|
def step_audit_output_has(context: Context, text: str) -> None:
|
|
assert text in context.cov_audit_output, (
|
|
f"Expected '{text}' in: {context.cov_audit_output}"
|
|
)
|
|
|
|
|
|
# ===========================================================================
|
|
# application/container.py — container init (lines 66-69, 125-130)
|
|
# ===========================================================================
|
|
|
|
|
|
@given("a container test environment")
|
|
def step_ctr_env(context: Context) -> None:
|
|
context.cov_container = None
|
|
context.cov_ctr_cleared = False
|
|
|
|
|
|
@when("I get the application container")
|
|
def step_ctr_get(context: Context) -> None:
|
|
from cleveragents.application.container import get_container, reset_container
|
|
|
|
reset_container()
|
|
context.cov_container = get_container()
|
|
|
|
|
|
@when("I reset the container singleton")
|
|
def step_ctr_reset(context: Context) -> None:
|
|
import cleveragents.application.container as container_mod
|
|
from cleveragents.application.container import get_container, reset_container
|
|
|
|
_ = get_container()
|
|
reset_container()
|
|
context.cov_ctr_cleared = container_mod._container is None
|
|
|
|
|
|
@then("the container should provide a database session")
|
|
def step_ctr_has_db(context: Context) -> None:
|
|
assert context.cov_container is not None
|
|
# Container has a 'db' provider attribute
|
|
assert hasattr(context.cov_container, "database_url")
|
|
|
|
|
|
@then("the container cache should be cleared")
|
|
def step_ctr_cleared(context: Context) -> None:
|
|
assert context.cov_ctr_cleared is True
|