Files
cleveragents-core/features/steps/system_coverage_boost_steps.py

330 lines
10 KiB
Python

"""Step definitions for system.py coverage boost."""
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 _mock_settings(**overrides):
"""Create mock settings 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.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_providers = []
s.debug_enabled = False
return s
@given("a system module for coverage testing")
def step_system_module(context: Context) -> None:
context.result_data = None
context.exception = None
@when("I call git_sha with git not on path")
def step_git_sha_unavailable(context: Context) -> None:
from cleveragents.cli.commands.system import _git_sha
with patch(
"cleveragents.cli.commands.system.subprocess.run",
side_effect=FileNotFoundError("git not found"),
):
context.result_data = _git_sha()
@then('the git sha should be "{expected}"')
def step_assert_git_sha(context: Context, expected: str) -> None:
assert context.result_data == expected
@when("I call dep_version for a missing package")
def step_dep_version_missing(context: Context) -> None:
from cleveragents.cli.commands.system import _dep_version
context.result_data = _dep_version("nonexistent-package-xyz-99999")
@then('the dep version should be "{expected}"')
def step_assert_dep_version(context: Context, expected: str) -> None:
assert context.result_data == expected
@when("I call build_info_data with no database file")
def step_info_no_db(context: Context) -> None:
from cleveragents.cli.commands.system import build_info_data
tmpdir = tempfile.mkdtemp()
ms = _mock_settings(
database_url=f"sqlite:///{tmpdir}/nonexistent.db",
log_dir=Path(tmpdir) / "logs",
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
context.result_data = build_info_data()
@then('the info data should contain db_size as "{expected}"')
def step_assert_db_size(context: Context, expected: str) -> None:
assert context.result_data["storage"]["db_size"] == expected
@when("I call build_info_data with no log directory")
def step_info_no_log_dir(context: Context) -> None:
from cleveragents.cli.commands.system import build_info_data
tmpdir = tempfile.mkdtemp()
ms = _mock_settings(
database_url=f"sqlite:///{tmpdir}/nonexistent.db",
log_dir=Path(tmpdir) / "nonexistent_logs_dir",
)
with patch("cleveragents.config.settings.get_settings", return_value=ms):
context.result_data = build_info_data()
@then('the info data should contain logs as "{expected}"')
def step_assert_logs(context: Context, expected: str) -> None:
assert context.result_data["storage"]["logs"] == expected
@when("I call check_data_dir with non-existent directory")
def step_check_data_dir_nonexistent(context: Context) -> None:
from cleveragents.cli.commands.system import _check_data_dir
ms = _mock_settings(data_dir=Path("/nonexistent/data/dir/xyz99"))
with patch("cleveragents.config.settings.get_settings", return_value=ms):
context.result_data = _check_data_dir()
@then("the check result should have WARN status")
def step_assert_warn(context: Context) -> None:
from cleveragents.cli.commands.system import CheckStatus
assert context.result_data["status"] == CheckStatus.WARN
@when("I call check_database with postgres URL")
def step_check_db_postgres(context: Context) -> None:
from cleveragents.cli.commands.system import _check_database
ms = _mock_settings(database_url="postgresql://localhost/test")
with patch("cleveragents.config.settings.get_settings", return_value=ms):
context.result_data = _check_database()
@then('the check result should have OK status and details "{expected}"')
def step_assert_ok_details(context: Context, expected: str) -> None:
from cleveragents.cli.commands.system import CheckStatus
assert context.result_data["status"] == CheckStatus.OK
assert context.result_data["details"] == expected
@when("I call check_database with sqlite and non-writable parent")
def step_check_db_nonwritable(context: Context) -> None:
from cleveragents.cli.commands.system import _check_database
ms = _mock_settings(database_url="sqlite:////nonexistent/parent/dir/test.db")
with patch("cleveragents.config.settings.get_settings", return_value=ms):
context.result_data = _check_database()
@then("the check result should have ERROR status")
def step_assert_error(context: Context) -> None:
from cleveragents.cli.commands.system import CheckStatus
assert context.result_data["status"] == CheckStatus.ERROR
@when("I call check_file_permissions with non-existent directory")
def step_check_perms_nonexistent(context: Context) -> None:
from cleveragents.cli.commands.system import _check_file_permissions
ms = _mock_settings(data_dir=Path("/nonexistent/data/dir/xyz99"))
with patch("cleveragents.config.settings.get_settings", return_value=ms):
context.result_data = _check_file_permissions()
@when("I call check_file_permissions with read-only directory")
def step_check_perms_readonly(context: Context) -> None:
from cleveragents.cli.commands.system import _check_file_permissions
tmpdir = tempfile.mkdtemp()
ms = _mock_settings(data_dir=Path(tmpdir))
original_access = os.access
def mock_access(p, mode):
p_str = str(p)
if tmpdir in p_str:
return mode == os.R_OK
return original_access(p, mode)
with (
patch("cleveragents.config.settings.get_settings", return_value=ms),
patch("cleveragents.cli.commands.system.os.access", side_effect=mock_access),
):
context.result_data = _check_file_permissions()
@then("the file perm result status should not be OK")
def step_assert_not_ok(context: Context) -> None:
from cleveragents.cli.commands.system import CheckStatus
assert context.result_data["status"] != CheckStatus.OK
@when("I call check_git with git unavailable")
def step_check_git_unavailable(context: Context) -> None:
from cleveragents.cli.commands.system import _check_git
with patch(
"cleveragents.cli.commands.system.subprocess.run",
side_effect=FileNotFoundError("git"),
):
context.result_data = _check_git()
@when("I call render_version_rich with sample data")
def step_render_version_rich(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": "1.0",
"platform": "Linux",
"dependencies": {"typer": "0.9.0", "rich": "13.0.0"},
}
try:
render_version_rich(data)
context.exception = None
except Exception as e:
context.exception = e
@then("no system exception should be raised")
def step_no_system_exception(context: Context) -> None:
assert context.exception is None
@when("I call render_info_rich with sample data")
def step_render_info_rich(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",
"automation": "auto",
"providers_configured": 0,
"debug_mode": False,
"storage": {"db_size": "1.0 MB", "logs": "0.5 MB"},
}
try:
render_info_rich(data)
context.exception = None
except Exception as e:
context.exception = e
@when("I call render_diagnostics_rich with error data")
def step_render_diag_errors(context: Context) -> None:
from cleveragents.cli.commands.system import CheckStatus, render_diagnostics_rich
data = {
"checks": [
{"name": "Config file", "status": CheckStatus.OK, "details": "ok"},
{"name": "Database", "status": CheckStatus.ERROR, "details": "missing"},
],
"summary": {
"total": 2,
"ok": 1,
"warnings": 0,
"errors": 1,
"duration_s": 0.01,
},
"recommendations": ["Fix database"],
"has_errors": True,
"has_warnings": False,
}
try:
render_diagnostics_rich(data)
context.exception = None
except Exception as e:
context.exception = e
@when("I call render_diagnostics_rich with warning data")
def step_render_diag_warnings(context: Context) -> None:
from cleveragents.cli.commands.system import CheckStatus, render_diagnostics_rich
data = {
"checks": [
{"name": "Config", "status": CheckStatus.OK, "details": "ok"},
{"name": "Disk", "status": CheckStatus.WARN, "details": "low"},
],
"summary": {
"total": 2,
"ok": 1,
"warnings": 1,
"errors": 0,
"duration_s": 0.01,
},
"recommendations": [],
"has_errors": False,
"has_warnings": True,
}
try:
render_diagnostics_rich(data)
context.exception = None
except Exception as e:
context.exception = e
@when("I call render_diagnostics_rich with all OK data")
def step_render_diag_ok(context: Context) -> None:
from cleveragents.cli.commands.system import CheckStatus, render_diagnostics_rich
data = {
"checks": [
{"name": "Config", "status": CheckStatus.OK, "details": "ok"},
{"name": "DB", "status": CheckStatus.OK, "details": "ok"},
],
"summary": {
"total": 2,
"ok": 2,
"warnings": 0,
"errors": 0,
"duration_s": 0.01,
},
"recommendations": [],
"has_errors": False,
"has_warnings": False,
}
try:
render_diagnostics_rich(data)
context.exception = None
except Exception as e:
context.exception = e