"""Step definitions for system.py uncovered branches.""" 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_mock_settings(**overrides): """Create 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 return s # --------------------------------------------------------------------------- # Given # --------------------------------------------------------------------------- @given("system cli branch module is loaded") def step_system_cli_branch_module_loaded(context: Context) -> None: context.branch_result = None context.branch_exception = None # --------------------------------------------------------------------------- # When - _git_sha # --------------------------------------------------------------------------- @when("system cli branch git_sha subprocess returns non-zero exit code") def step_branch_git_sha_nonzero(context: Context) -> None: from cleveragents.cli.commands.system import _git_sha mock_result = MagicMock() mock_result.returncode = 1 mock_result.stdout = "" with patch( "cleveragents.cli.commands.system.subprocess.run", return_value=mock_result, ): context.branch_result = _git_sha() # --------------------------------------------------------------------------- # When - build_info_data (DB exists, compute size) # --------------------------------------------------------------------------- @when("system cli branch build_info_data is called with existing database file") def step_branch_info_db_exists(context: Context) -> None: from cleveragents.cli.commands.system import build_info_data tmpdir = tempfile.mkdtemp() db_file = Path(tmpdir) / "test.db" db_file.write_bytes(b"x" * 2048) # 2 KB file ms = _make_mock_settings( database_url=f"sqlite:///{db_file}", log_dir=Path(tmpdir) / "no_logs", storage_path=Path(tmpdir), ) with patch("cleveragents.config.settings.get_settings", return_value=ms): context.branch_result = build_info_data() # --------------------------------------------------------------------------- # When - build_info_data (DB stat raises exception) # --------------------------------------------------------------------------- @when("system cli branch build_info_data is called with db stat raising exception") def step_branch_info_db_exception(context: Context) -> None: from cleveragents.cli.commands.system import build_info_data tmpdir = tempfile.mkdtemp() # Create a db file that exists but whose .stat() will fail db_file = Path(tmpdir) / "test.db" db_file.write_bytes(b"data") ms = _make_mock_settings( database_url=f"sqlite:///{db_file}", log_dir=Path(tmpdir) / "no_logs", storage_path=Path(tmpdir), ) # We need to make db_path.exists() return True but db_path.stat() raise # The simplest approach: patch the stat call on the db_path original_stat = Path.stat def patched_stat(self, *args, **kwargs): if str(self) == str(db_file): raise PermissionError("mocked stat failure") return original_stat(self, *args, **kwargs) with ( patch("cleveragents.config.settings.get_settings", return_value=ms), patch.object(Path, "stat", patched_stat), ): context.branch_result = build_info_data() # --------------------------------------------------------------------------- # When - build_info_data (log_dir exists with files) # --------------------------------------------------------------------------- @when("system cli branch build_info_data is called with existing log directory") def step_branch_info_log_exists(context: Context) -> None: from cleveragents.cli.commands.system import build_info_data tmpdir = tempfile.mkdtemp() log_dir = Path(tmpdir) / "logs" log_dir.mkdir() (log_dir / "app.log").write_bytes(b"log data " * 100) ms = _make_mock_settings( database_url=f"sqlite:///{tmpdir}/nonexistent.db", log_dir=log_dir, storage_path=Path(tmpdir), ) with patch("cleveragents.config.settings.get_settings", return_value=ms): context.branch_result = build_info_data() # --------------------------------------------------------------------------- # When - _check_config_file (exists but not readable) # --------------------------------------------------------------------------- @when("system cli branch check_config_file is called with unreadable config") def step_branch_config_not_readable(context: Context) -> None: from cleveragents.cli.commands.system import _check_config_file tmpdir = tempfile.mkdtemp() config_file = Path(tmpdir) / "config.toml" config_file.write_text("[general]\n") with ( patch.dict(os.environ, {"CLEVERAGENTS_CONFIG_PATH": str(config_file)}), patch( "cleveragents.cli.commands.system.os.access", return_value=False, ), ): context.branch_result = _check_config_file() # Clean up env var os.environ.pop("CLEVERAGENTS_CONFIG_PATH", None) # --------------------------------------------------------------------------- # When - _check_data_dir (exists but not writable) # --------------------------------------------------------------------------- @when("system cli branch check_data_dir is called with non-writable directory") def step_branch_data_dir_not_writable(context: Context) -> None: from cleveragents.cli.commands.system import _check_data_dir tmpdir = tempfile.mkdtemp() data_dir = Path(tmpdir) / "data" data_dir.mkdir() ms = _make_mock_settings(data_dir=data_dir) def mock_access(p, mode): return not (str(p) == str(data_dir) and mode == os.W_OK) with ( patch("cleveragents.config.settings.get_settings", return_value=ms), patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access), ): context.branch_result = _check_data_dir() # --------------------------------------------------------------------------- # When - _check_data_dir (dir missing) # --------------------------------------------------------------------------- @when("system cli branch check_data_dir is called with missing directory") def step_branch_data_dir_missing(context: Context) -> None: from cleveragents.cli.commands.system import _check_data_dir ms = _make_mock_settings(data_dir=Path("/tmp/nonexistent_dir_xyz_99999")) with patch("cleveragents.config.settings.get_settings", return_value=ms): context.branch_result = _check_data_dir() # --------------------------------------------------------------------------- # When - _check_database (sqlite DB exists but not writable) # --------------------------------------------------------------------------- @when("system cli branch check_database is called with non-writable sqlite db") def step_branch_db_not_writable(context: Context) -> None: from cleveragents.cli.commands.system import _check_database tmpdir = tempfile.mkdtemp() db_file = Path(tmpdir) / "test.db" db_file.write_bytes(b"sqlite data") ms = _make_mock_settings(database_url=f"sqlite:///{db_file}") def mock_access(p, mode): return not (str(p) == str(db_file) and mode == os.W_OK) with ( patch("cleveragents.config.settings.get_settings", return_value=ms), patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access), ): context.branch_result = _check_database() # --------------------------------------------------------------------------- # When - _check_disk_space (< 0.5 GB free) # --------------------------------------------------------------------------- @when("system cli branch check_disk_space reports less than half GB free") def step_branch_disk_critical(context: Context) -> None: from cleveragents.cli.commands.system import _check_disk_space mock_usage = MagicMock() mock_usage.free = int(0.3 * (1024**3)) # 0.3 GB mock_usage.total = 100 * (1024**3) with patch( "cleveragents.cli.commands.system.shutil.disk_usage", return_value=mock_usage, ): context.branch_result = _check_disk_space() # --------------------------------------------------------------------------- # When - _check_disk_space (< 1.0 GB free) # --------------------------------------------------------------------------- @when("system cli branch check_disk_space reports less than one GB free") def step_branch_disk_low(context: Context) -> None: from cleveragents.cli.commands.system import _check_disk_space mock_usage = MagicMock() mock_usage.free = int(0.7 * (1024**3)) # 0.7 GB mock_usage.total = 100 * (1024**3) with patch( "cleveragents.cli.commands.system.shutil.disk_usage", return_value=mock_usage, ): context.branch_result = _check_disk_space() # --------------------------------------------------------------------------- # When - _check_disk_space (OSError) # --------------------------------------------------------------------------- @when("system cli branch check_disk_space raises OSError") def step_branch_disk_oserror(context: Context) -> None: from cleveragents.cli.commands.system import _check_disk_space with patch( "cleveragents.cli.commands.system.shutil.disk_usage", side_effect=OSError("permission denied"), ): context.branch_result = _check_disk_space() # --------------------------------------------------------------------------- # When - _check_git (non-zero exit code, not FileNotFoundError) # --------------------------------------------------------------------------- @when("system cli branch check_git subprocess returns non-zero exit code") def step_branch_git_nonzero(context: Context) -> None: from cleveragents.cli.commands.system import _check_git mock_result = MagicMock() mock_result.returncode = 127 mock_result.stdout = "" with patch( "cleveragents.cli.commands.system.subprocess.run", return_value=mock_result, ): context.branch_result = _check_git() # --------------------------------------------------------------------------- # When - _check_file_permissions (data dir missing) # --------------------------------------------------------------------------- @when("system cli branch check_file_permissions with missing data dir") def step_branch_perms_missing(context: Context) -> None: from cleveragents.cli.commands.system import _check_file_permissions ms = _make_mock_settings(data_dir=Path("/tmp/nonexistent_dir_xyz_99999")) with patch("cleveragents.config.settings.get_settings", return_value=ms): context.branch_result = _check_file_permissions() # --------------------------------------------------------------------------- # When - _check_file_permissions (readable but not writable) # --------------------------------------------------------------------------- @when("system cli branch check_file_permissions with readable not writable dir") def step_branch_perms_read_no_write(context: Context) -> None: from cleveragents.cli.commands.system import _check_file_permissions tmpdir = tempfile.mkdtemp() data_dir = Path(tmpdir) ms = _make_mock_settings(data_dir=data_dir) def mock_access(p, mode): if str(p) == str(data_dir): if mode == os.R_OK: return True if mode == os.W_OK: return False return True with ( patch("cleveragents.config.settings.get_settings", return_value=ms), patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access), ): context.branch_result = _check_file_permissions() # --------------------------------------------------------------------------- # When - _check_file_permissions (no access at all) # --------------------------------------------------------------------------- @when("system cli branch check_file_permissions with no access dir") def step_branch_perms_no_access(context: Context) -> None: from cleveragents.cli.commands.system import _check_file_permissions tmpdir = tempfile.mkdtemp() data_dir = Path(tmpdir) ms = _make_mock_settings(data_dir=data_dir) def mock_access(p, mode): return str(p) != str(data_dir) with ( patch("cleveragents.config.settings.get_settings", return_value=ms), patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access), ): context.branch_result = _check_file_permissions() # --------------------------------------------------------------------------- # When - render_version_rich (no dependencies) # --------------------------------------------------------------------------- @when("system cli branch render_version_rich is called with no dependencies") def step_branch_version_no_deps(context: Context) -> None: from cleveragents.cli.commands.system import render_version_rich data = { "version": "1.0.0", "channel": "stable", "python": "3.13.0", "build_date": "2025-01-01", "commit": "abc1234", "schema": "v3", "platform": "linux-x86_64", "dependencies": {}, } try: render_version_rich(data) context.branch_exception = None except Exception as e: context.branch_exception = e # --------------------------------------------------------------------------- # When - render_info_rich (no storage) # --------------------------------------------------------------------------- @when("system cli branch render_info_rich is called with no storage") def step_branch_info_no_storage(context: Context) -> None: from cleveragents.cli.commands.system import render_info_rich data = { "data_dir": "/tmp/data", "config_path": "/tmp/config.toml", "database": "sqlite:///test.db", "server_mode": "disabled", "platform": "Linux x86_64", "automation": "auto", "providers_configured": 0, "debug_mode": False, "storage": {}, } try: render_info_rich(data) context.branch_exception = None except Exception as e: context.branch_exception = e # --------------------------------------------------------------------------- # Then - generic assertions # --------------------------------------------------------------------------- @then('system cli branch result should equal "{expected}"') def step_branch_assert_result(context: Context, expected: str) -> None: assert context.branch_result == expected, ( f"Expected '{expected}', got '{context.branch_result}'" ) @then('system cli branch storage db_size should contain "{substring}"') def step_branch_assert_db_size_contains(context: Context, substring: str) -> None: db_size = context.branch_result["storage"]["db_size"] assert substring in db_size, f"Expected '{substring}' in '{db_size}'" @then('system cli branch storage db_size should equal "{expected}"') def step_branch_assert_db_size_eq(context: Context, expected: str) -> None: db_size = context.branch_result["storage"]["db_size"] assert db_size == expected, f"Expected '{expected}', got '{db_size}'" @then('system cli branch storage logs should contain "{substring}"') def step_branch_assert_logs_contains(context: Context, substring: str) -> None: logs = context.branch_result["storage"]["logs"] assert substring in logs, f"Expected '{substring}' in '{logs}'" @then('system cli branch check status should be "{expected}"') def step_branch_assert_check_status(context: Context, expected: str) -> None: status = context.branch_result["status"] assert status == expected, f"Expected status '{expected}', got '{status}'" @then('system cli branch check details should be "{expected}"') def step_branch_assert_check_details(context: Context, expected: str) -> None: details = context.branch_result["details"] assert details == expected, f"Expected details '{expected}', got '{details}'" @then('system cli branch check details should contain "{substring}"') def step_branch_assert_check_details_contains(context: Context, substring: str) -> None: details = context.branch_result["details"] assert substring in details, f"Expected '{substring}' in '{details}'" @then("system cli branch no exception should be raised") def step_branch_assert_no_exception(context: Context) -> None: assert context.branch_exception is None, ( f"Unexpected exception: {context.branch_exception}" )