"""Step definitions for system_diagnostics_coverage.feature. Covers uncovered paths in cli/commands/system.py: - _check_file_permissions: writable-but-not-readable (line 344) - _check_stale_locks: locks table exists paths (lines 367-388) - _check_async_worker_health: async enabled path (lines 405-417) - _check_error_patterns: exception path (lines 438-442) All step patterns use the 'sysdiag_' prefix to avoid AmbiguousStep collisions with existing steps in cli_core_steps.py and coverage_security_template_boost_steps.py. """ from __future__ import annotations import tempfile from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch from behave import given, then, when from behave.runner import Context __all__: list[str] = [] # =========================================================================== # Helpers # =========================================================================== def _sysdiag_mock_settings(**overrides: Any) -> Any: """Create mock settings with sensible defaults for diagnostic 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.log_dir = overrides.get("log_dir", Path(tmpdir) / "logs") s.default_automation_profile = "auto" s.debug_enabled = False s.async_enabled = overrides.get("async_enabled", False) s.async_max_workers = overrides.get("async_max_workers", 4) s.async_poll_interval = overrides.get("async_poll_interval", 1.0) s.has_provider_configured = MagicMock(return_value=False) s.configured_providers = [] return s # =========================================================================== # Given steps # =========================================================================== @given("a sysdiag test environment") def step_sysdiag_env(context: Context) -> None: context.sysdiag_result = {} context.sysdiag_tmpdir = tempfile.mkdtemp() @given("a data dir that exists") def step_sysdiag_data_dir_exists(context: Context) -> None: data_dir = Path(context.sysdiag_tmpdir) / "data" data_dir.mkdir(parents=True, exist_ok=True) context.sysdiag_data_dir = data_dir # =========================================================================== # When steps — _check_file_permissions edge cases # =========================================================================== @when("I run the file permissions check with read access denied") def step_sysdiag_perms_no_read(context: Context) -> None: from cleveragents.cli.commands.system import _check_file_permissions ms = _sysdiag_mock_settings(data_dir=context.sysdiag_data_dir) def fake_access(path: Any, mode: int) -> bool: """Simulate writable-but-not-readable directory.""" import os as _os if mode == _os.R_OK: return False return mode == _os.W_OK with ( patch("cleveragents.config.settings.get_settings", return_value=ms), patch("cleveragents.cli.commands.system.os.access", side_effect=fake_access), ): context.sysdiag_result = _check_file_permissions() @when("I run the file permissions check with no access at all") def step_sysdiag_perms_no_access(context: Context) -> None: from cleveragents.cli.commands.system import _check_file_permissions ms = _sysdiag_mock_settings(data_dir=context.sysdiag_data_dir) with ( patch("cleveragents.config.settings.get_settings", return_value=ms), patch("cleveragents.cli.commands.system.os.access", return_value=False), ): context.sysdiag_result = _check_file_permissions() # =========================================================================== # When steps — _check_stale_locks edge cases # =========================================================================== @when("I run the stale locks check with {count:d} stale locks in the table") def step_sysdiag_stale_locks_count(context: Context, count: int) -> None: from cleveragents.cli.commands.system import _check_stale_locks mock_engine = MagicMock() 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 = count with ( patch( "cleveragents.cli.commands.system.get_database_url", return_value="sqlite:///test.db", ), patch( "cleveragents.cli.commands.system.create_engine", return_value=mock_engine, ), patch( "cleveragents.cli.commands.system.sa_inspect", return_value=mock_inspector, ), patch( "cleveragents.cli.commands.system.LockService", return_value=mock_lock_service, ), ): context.sysdiag_result = _check_stale_locks() @when("I run the stale locks check with a database error") def step_sysdiag_stale_locks_error(context: Context) -> None: from cleveragents.cli.commands.system import _check_stale_locks with patch( "cleveragents.cli.commands.system.get_database_url", side_effect=RuntimeError("database unreachable"), ): context.sysdiag_result = _check_stale_locks() # =========================================================================== # When steps — _check_async_worker_health edge cases # =========================================================================== @when("I run the async worker health check with async enabled") def step_sysdiag_async_enabled(context: Context) -> None: from cleveragents.cli.commands.system import _check_async_worker_health ms = _sysdiag_mock_settings( async_enabled=True, async_max_workers=8, async_poll_interval=2.5, ) with patch("cleveragents.config.settings.get_settings", return_value=ms): context.sysdiag_result = _check_async_worker_health() @when("I run the async worker health check with a settings error") def step_sysdiag_async_error(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.sysdiag_result = _check_async_worker_health() # =========================================================================== # When steps — _check_error_patterns exception path # =========================================================================== @when("I run the error patterns check with a service error") def step_sysdiag_error_patterns_error(context: Context) -> None: from cleveragents.cli.commands.system import _check_error_patterns with patch( "cleveragents.application.services.error_pattern_service.ErrorPatternService", side_effect=RuntimeError("service init failed"), ): context.sysdiag_result = _check_error_patterns() # =========================================================================== # Then steps # =========================================================================== @then('the sysdiag check status should be "{expected}"') def step_sysdiag_status(context: Context, expected: str) -> None: from cleveragents.cli.commands.system import CheckStatus status_map = { "ok": CheckStatus.OK, "warn": CheckStatus.WARN, "error": CheckStatus.ERROR, } actual = context.sysdiag_result["status"] assert actual == status_map[expected], ( f"Expected status '{expected}', got '{actual}'" ) @then('the sysdiag check details should be "{expected}"') def step_sysdiag_details_exact(context: Context, expected: str) -> None: actual = context.sysdiag_result["details"] assert actual == expected, f"Expected details '{expected}', got '{actual}'" @then('the sysdiag check details should contain "{text}"') def step_sysdiag_details_contains(context: Context, text: str) -> None: details = context.sysdiag_result["details"] assert text in details, f"Expected '{text}' in details: '{details}'" @then("the sysdiag check should have a recommendation") def step_sysdiag_has_recommendation(context: Context) -> None: rec = context.sysdiag_result.get("recommendation") assert rec is not None and len(rec) > 0, ( f"Expected a non-empty recommendation, got: {rec!r}" )