forked from HAL9000/cleveragents-core
freemo
a808c395f9
Add 53 new .feature files and corresponding step definition files targeting uncovered lines identified in build/coverage.xml. Fix AmbiguousStep conflicts in 7 pre-existing step files by disambiguating step text. New tests cover: ACP clients/facade, actor CLI/config, application container, ACMS service/strategies, async worker, automation profile CLI, autonomy guardrail, bridge, change model, config CLI/service, context service, cross-plan correction, database models, decision service, decomposition clustering/service, discovery handler, langchain chat provider, langgraph nodes, materializers, multi-project service, plan apply/CLI/lifecycle/model/ preflight/resume/service, PostgreSQL analyzer, project CLI/context CLI, provider registry, reactive application/route, repositories, resolver handler, resource registry service, resume model, retry patterns, sandbox protocol, server CLI, skill CLI/service, skills registry, subplan execution/service, system CLI, UKO loader, UoW, and YAML template engine. Closes #645
226 lines
8.3 KiB
Python
226 lines
8.3 KiB
Python
"""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}'"
|