forked from HAL9000/cleveragents-core
484 lines
17 KiB
Python
484 lines
17 KiB
Python
"""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}"
|
|
)
|