test(coverage): add 32 BDD scenarios to boost coverage from 97.0% to 97.2%
Add coverage boost tests targeting low-coverage modules: - skills/registry.py: validate_skill error paths - cli/formatting.py: format_output_session, _format_table branches - cli/commands/system.py: health check functions - cli/commands/session.py: CRUD command edge cases - cli/commands/cleanup.py: error handling paths - cli/commands/audit.py: audit service creation - application/container.py: container init/reset All 32 new scenarios pass alongside existing 29 security template scenarios. Full nox -s coverage_report confirmed at 97.2%.
This commit is contained in:
CoreRasurae
@@ -0,0 +1,225 @@
|
||||
@coverage_boost
|
||||
Feature: Coverage boost for security template branch
|
||||
As a developer ensuring 97%+ test coverage
|
||||
I want to exercise uncovered code paths across the codebase
|
||||
So that the coverage threshold is comfortably met
|
||||
|
||||
# =========================================================================
|
||||
# skills/registry.py — validate_skill with tool registry, inline tools,
|
||||
# and includes (lines 200-213)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Validate skill with tool registry reports missing tool refs
|
||||
Given a skill registry with a mock tool registry
|
||||
And a skill definition with tool ref "nonexistent-tool"
|
||||
When I validate the skill definition via the registry
|
||||
Then the skill validation errors should mention "not found in tool registry"
|
||||
|
||||
Scenario: Validate skill with inline tool missing description
|
||||
Given a skill registry with no tool registry
|
||||
And a skill definition with an inline tool missing its description
|
||||
When I validate the skill definition via the registry
|
||||
Then the skill validation errors should mention "missing a description"
|
||||
|
||||
Scenario: Validate skill referencing unknown included skill
|
||||
Given a skill registry with no tool registry
|
||||
And a skill definition that includes "unknown/skill"
|
||||
When I validate the skill definition via the registry
|
||||
Then the skill validation errors should mention "is not registered"
|
||||
|
||||
Scenario: Validate skill with all valid references
|
||||
Given a skill registry with no tool registry
|
||||
And a skill definition with no tool refs or includes
|
||||
When I validate the skill definition via the registry
|
||||
Then the skill validation errors should be empty
|
||||
|
||||
# =========================================================================
|
||||
# cli/formatting.py — format_output_session, _format_table edge cases
|
||||
# (lines 121, 128, 197, 231-236)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Format output with color format type
|
||||
Given sample formatting data with key "status" and value "ok"
|
||||
When I format the data with format type "color"
|
||||
Then the formatted output should contain "status"
|
||||
|
||||
Scenario: Format output session with list of dicts
|
||||
Given sample formatting data as a list of two items
|
||||
When I format the data via format_output_session with format "plain"
|
||||
Then the formatted session output should not be empty
|
||||
|
||||
Scenario: Format output session with dict input
|
||||
Given sample formatting data with key "name" and value "test"
|
||||
When I format the data via format_output_session with format "json"
|
||||
Then the formatted session output should contain "name"
|
||||
|
||||
Scenario: Format table with rows having extra keys
|
||||
Given sample formatting data as a list with mismatched keys
|
||||
When I format the data with format type "table"
|
||||
Then the formatted output should contain "extra_col"
|
||||
|
||||
Scenario: Format table with empty list
|
||||
Given sample formatting data as an empty list
|
||||
When I format the data with format type "table"
|
||||
Then the formatted output should be "(empty)"
|
||||
|
||||
# =========================================================================
|
||||
# cli/commands/system.py — health check functions
|
||||
# (lines 108-118, 146-147, 174-175, 193-194, 258-278, 325-334)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Check config file when it exists and is readable
|
||||
Given a system health check environment
|
||||
And a config file that exists and is readable
|
||||
When I run the config file health check
|
||||
Then the health check status should be "ok"
|
||||
And the health check details should be "readable"
|
||||
|
||||
Scenario: Check data dir when it does not exist
|
||||
Given a system health check environment
|
||||
And a data dir that does not exist
|
||||
When I run the data dir health check
|
||||
Then the health check status should be "warn"
|
||||
And the health check details should contain "missing"
|
||||
|
||||
Scenario: Check disk space with plenty of space
|
||||
Given a system health check environment
|
||||
When I run the disk space health check
|
||||
Then the health check status should be "ok"
|
||||
|
||||
Scenario: Check git availability
|
||||
Given a system health check environment
|
||||
When I run the git availability health check
|
||||
Then the health check status should be "ok"
|
||||
|
||||
Scenario: Check file permissions on writable data dir
|
||||
Given a system health check environment
|
||||
And a writable data dir
|
||||
When I run the file permissions health check
|
||||
Then the health check status should be "ok"
|
||||
And the health check details should be "data dir r/w"
|
||||
|
||||
Scenario: Check file permissions on nonexistent data dir
|
||||
Given a system health check environment
|
||||
And a data dir that does not exist
|
||||
When I run the file permissions health check
|
||||
Then the health check status should be "warn"
|
||||
And the health check details should contain "does not exist"
|
||||
|
||||
Scenario: Build info data with existing database file
|
||||
Given a system health check environment
|
||||
And a database file that exists
|
||||
When I build info data
|
||||
Then the info data should contain a db_size value
|
||||
|
||||
Scenario: Build info data with log directory
|
||||
Given a system health check environment
|
||||
And a log directory with files
|
||||
When I build info data
|
||||
Then the info data should contain a logs value
|
||||
|
||||
# =========================================================================
|
||||
# cli/commands/session.py — session CLI commands
|
||||
# (lines 56-69, 75, 157-159, 267, 277-278, 320-323, 381-382, 482-485)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Get session service lazily initialises from container
|
||||
Given a session CLI test environment
|
||||
When I call get_session_service
|
||||
Then a session service should be returned
|
||||
|
||||
Scenario: Reset session service clears the cached instance
|
||||
Given a session CLI test environment
|
||||
When I call reset_session_service
|
||||
Then the session service cache should be cleared
|
||||
|
||||
Scenario: Session create with non-rich format outputs formatted data
|
||||
Given a session CLI test environment
|
||||
And a mock session service that returns a created session
|
||||
When I invoke session create with format "json"
|
||||
Then the covboost session output should contain "session_id"
|
||||
|
||||
Scenario: Session create error shows session not found message
|
||||
Given a session CLI test environment
|
||||
And a mock session service that raises SessionNotFoundError on create
|
||||
When I invoke session create expecting an error
|
||||
Then the session CLI should have exited with error
|
||||
|
||||
Scenario: Session list with no sessions shows empty message
|
||||
Given a session CLI test environment
|
||||
And a mock session service that returns no sessions
|
||||
When I invoke session list with format "rich"
|
||||
Then the covboost session output should contain "No sessions found"
|
||||
|
||||
Scenario: Session delete without confirmation aborts
|
||||
Given a session CLI test environment
|
||||
And a mock session service that returns a session
|
||||
When I invoke session delete without confirming
|
||||
Then the session CLI should have been aborted
|
||||
|
||||
Scenario: Session delete with yes flag succeeds
|
||||
Given a session CLI test environment
|
||||
And a mock session service that returns a session
|
||||
When I invoke session delete with yes flag
|
||||
Then the covboost session output should contain "deleted"
|
||||
|
||||
Scenario: Session export to stdout outputs JSON
|
||||
Given a session CLI test environment
|
||||
And a mock session service that can export
|
||||
When I invoke session export to stdout
|
||||
Then the covboost session output should contain "session_id"
|
||||
|
||||
Scenario: Session export to file that exists without force fails
|
||||
Given a session CLI test environment
|
||||
And a mock session service that can export
|
||||
And a temporary output file that already exists
|
||||
When I invoke session export to that file without force
|
||||
Then the session CLI should have exited with error
|
||||
|
||||
# =========================================================================
|
||||
# cli/commands/cleanup.py — error handling paths
|
||||
# (lines 49-58, 94, 129-131)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Active plan detection with OSError returns empty set
|
||||
Given a cleanup CLI test environment
|
||||
And a cleanup service that raises OSError on sandbox scan
|
||||
When I detect active plans
|
||||
Then the active plans set should be empty
|
||||
|
||||
Scenario: Cleanup prune command runs successfully
|
||||
Given a cleanup CLI test environment
|
||||
And a mock cleanup service
|
||||
When I invoke cleanup prune
|
||||
Then the cleanup should have run without error
|
||||
|
||||
# =========================================================================
|
||||
# cli/commands/audit.py — audit service creation
|
||||
# (lines 29-32, 110-111)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Audit service is created from settings
|
||||
Given an audit CLI test environment
|
||||
When I create the audit service
|
||||
Then an audit service instance should be returned
|
||||
|
||||
Scenario: Audit list command with no entries
|
||||
Given an audit CLI test environment
|
||||
And a mock audit service with no entries
|
||||
When I invoke audit list
|
||||
Then the covboost audit output should contain "No audit"
|
||||
|
||||
# =========================================================================
|
||||
# application/container.py — container init paths
|
||||
# (lines 66-69, 125-130)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Container initialises database from settings
|
||||
Given a container test environment
|
||||
When I get the application container
|
||||
Then the container should provide a database session
|
||||
|
||||
Scenario: Container clears its singleton on reset
|
||||
Given a container test environment
|
||||
When I reset the container singleton
|
||||
Then the container cache should be cleared
|
||||
@@ -0,0 +1,855 @@
|
||||
"""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 cleveragents.cli.formatting import format_output
|
||||
|
||||
context.cov_format_output = format_output(context.cov_format_data, fmt)
|
||||
|
||||
|
||||
@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 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))):
|
||||
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
|
||||
Reference in New Issue
Block a user