"""Step definitions for system_cli_coverage_boost.feature. Targets the remaining uncovered lines in system.py: - Line 344: _check_file_permissions writable-only branch - Lines 367-376: _check_stale_locks with locks table, 0 stale - Lines 378-382: _check_stale_locks with locks table, N stale - Lines 384-388: _check_stale_locks exception path - Lines 405-410: _check_async_worker_health async enabled - Lines 413-417: _check_async_worker_health exception path """ from __future__ import annotations import os import tempfile from pathlib import Path from unittest.mock import MagicMock, patch from behave import given, then, when from behave.runner import Context def _make_settings(**overrides): """Build a mock Settings object with sensible defaults.""" 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) / "nonexistent_logs") s.default_automation_profile = "auto" s.has_provider_configured = MagicMock(return_value=False) s.configured_provider_names = MagicMock(return_value=[]) s.debug_enabled = False s.async_enabled = overrides.get("async_enabled", False) s.async_max_workers = overrides.get("async_max_workers", 2) s.async_poll_interval = overrides.get("async_poll_interval", 1) return s # --------------------------------------------------------------------------- # Given # --------------------------------------------------------------------------- @given("the system coverage boost module is loaded") def step_coverage_boost_module_loaded(context: Context) -> None: context.cov_result = None # --------------------------------------------------------------------------- # When — _check_file_permissions: writable but not readable (line 344) # --------------------------------------------------------------------------- @when("I call check_file_permissions with writable-only data dir") def step_perms_writable_only(context: Context) -> None: from cleveragents.cli.commands.system import _check_file_permissions tmpdir = tempfile.mkdtemp() data_dir = Path(tmpdir) ms = _make_settings(data_dir=data_dir) def mock_access(p, mode): if str(p) == str(data_dir): if mode == os.R_OK: return False if mode == os.W_OK: return True return True with ( patch("cleveragents.config.settings.get_settings", return_value=ms), patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access), ): context.cov_result = _check_file_permissions() # --------------------------------------------------------------------------- # When — _check_stale_locks: locks table exists, 0 stale (lines 367-376) # --------------------------------------------------------------------------- @when("I call check_stale_locks with locks table present and zero stale locks") def step_stale_locks_zero(context: Context) -> None: from cleveragents.cli.commands.system import _check_stale_locks mock_inspector = MagicMock() mock_inspector.get_table_names.return_value = ["locks", "other_table"] mock_lock_service = MagicMock() mock_lock_service.count_stale_locks.return_value = 0 with ( patch( "cleveragents.cli.commands.system.get_database_url", return_value="sqlite:///test.db", ), patch("cleveragents.cli.commands.system.create_engine"), patch( "cleveragents.cli.commands.system.sa_inspect", return_value=mock_inspector, ), patch( "cleveragents.cli.commands.system.LockService", return_value=mock_lock_service, ), ): context.cov_result = _check_stale_locks() # --------------------------------------------------------------------------- # When — _check_stale_locks: locks table exists, N stale (lines 378-382) # --------------------------------------------------------------------------- @when("I call check_stale_locks with locks table present and {count:d} stale locks") def step_stale_locks_nonzero(context: Context, count: int) -> None: from cleveragents.cli.commands.system import _check_stale_locks mock_inspector = MagicMock() mock_inspector.get_table_names.return_value = ["locks"] mock_lock_service = MagicMock() mock_lock_service.count_stale_locks.return_value = count with ( patch( "cleveragents.cli.commands.system.get_database_url", return_value="sqlite:///test.db", ), patch("cleveragents.cli.commands.system.create_engine"), patch( "cleveragents.cli.commands.system.sa_inspect", return_value=mock_inspector, ), patch( "cleveragents.cli.commands.system.LockService", return_value=mock_lock_service, ), ): context.cov_result = _check_stale_locks() # --------------------------------------------------------------------------- # When — _check_stale_locks: exception (lines 384-388) # --------------------------------------------------------------------------- @when("I call check_stale_locks and the database connection raises an exception") def step_stale_locks_exception(context: Context) -> None: from cleveragents.cli.commands.system import _check_stale_locks with patch( "cleveragents.cli.commands.system.get_database_url", side_effect=RuntimeError("connection refused"), ): context.cov_result = _check_stale_locks() # --------------------------------------------------------------------------- # When — _check_async_worker_health: async enabled (lines 405-410) # --------------------------------------------------------------------------- @when( "I call check_async_worker_health with async enabled and max_workers {max_w:d} and poll_interval {poll:d}" ) def step_async_enabled(context: Context, max_w: int, poll: int) -> None: from cleveragents.cli.commands.system import _check_async_worker_health ms = _make_settings( async_enabled=True, async_max_workers=max_w, async_poll_interval=poll, ) with patch("cleveragents.config.settings.get_settings", return_value=ms): context.cov_result = _check_async_worker_health() # --------------------------------------------------------------------------- # When — _check_async_worker_health: exception (lines 413-417) # --------------------------------------------------------------------------- @when("I call check_async_worker_health and get_settings raises an exception") def step_async_exception(context: Context) -> None: from cleveragents.cli.commands.system import _check_async_worker_health with patch( "cleveragents.config.settings.get_settings", side_effect=RuntimeError("settings unavailable"), ): context.cov_result = _check_async_worker_health() # --------------------------------------------------------------------------- # Then — assertions # --------------------------------------------------------------------------- @then('the coverage boost result status should be "{expected}"') def step_assert_status(context: Context, expected: str) -> None: status = str(context.cov_result["status"]) assert status == expected, f"Expected status '{expected}', got '{status}'" @then('the coverage boost result details should be "{expected}"') def step_assert_details_exact(context: Context, expected: str) -> None: details = context.cov_result["details"] assert details == expected, f"Expected details '{expected}', got '{details}'" @then('the coverage boost result details should contain "{substring}"') def step_assert_details_contains(context: Context, substring: str) -> None: details = context.cov_result["details"] assert substring in details, f"Expected '{substring}' in '{details}'" @then('the coverage boost result should have recommendation "{expected}"') def step_assert_recommendation(context: Context, expected: str) -> None: rec = context.cov_result.get("recommendation") assert rec == expected, f"Expected recommendation '{expected}', got '{rec}'"