From d6c58344d100984b05464d983f420d798b184764 Mon Sep 17 00:00:00 2001 From: CoreRasurae Date: Tue, 24 Feb 2026 19:43:27 +0000 Subject: [PATCH] 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%. --- .../security_template_coverage_boost.feature | 225 +++++ .../coverage_security_template_boost_steps.py | 855 ++++++++++++++++++ 2 files changed, 1080 insertions(+) create mode 100644 features/security_template_coverage_boost.feature create mode 100644 features/steps/coverage_security_template_boost_steps.py diff --git a/features/security_template_coverage_boost.feature b/features/security_template_coverage_boost.feature new file mode 100644 index 000000000..f25d51b53 --- /dev/null +++ b/features/security_template_coverage_boost.feature @@ -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 diff --git a/features/steps/coverage_security_template_boost_steps.py b/features/steps/coverage_security_template_boost_steps.py new file mode 100644 index 000000000..5bd7a2e67 --- /dev/null +++ b/features/steps/coverage_security_template_boost_steps.py @@ -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