forked from HAL9000/cleveragents-core
freemo
051ee7c290
Added 52 new .feature files and corresponding _steps.py files targeting previously uncovered code paths in the following areas: - TUI layer: app, commands, persona (state/schema/registry), widgets, input (shell_exec, reference_parser) - Application services: plan lifecycle/service/executor, session, project, repo indexing, correction, checkpoint, actor, llm_actors, strategy coordinator, resource file watcher, service retry wiring - CLI commands: session, resource, repl, plan, db, automation_profile - Domain models: retry_policy, resource_type, cost_budget, docker_compose_analyzer, detail_level, _sql_string_aware, _postgresql_helpers - Core: circuit_breaker, retry_service_patterns - Infrastructure: repositories, transaction_sandbox, strategy_registry, plugins/loader, container - Config: settings - Agents: plan_generation, context_analysis, auto_debug - A2A: facade All new tests follow the Behave/Gherkin BDD standard. Resolved step definition collisions with unique prefixes. Fixed Alembic fileConfig logger disabling issue (disable_existing_loggers=False). ISSUES CLOSED: #1068
250 lines
10 KiB
Python
250 lines
10 KiB
Python
"""Step definitions for checkpoint_service_coverage_boost.feature.
|
|
|
|
These steps target specific uncovered lines in checkpoint_service.py:
|
|
- Lines 484-487: subprocess.CalledProcessError handler in _run_git
|
|
- Lines 488-491: subprocess.TimeoutExpired handler in _run_git
|
|
- Line 417: ResourceNotFoundError when lifecycle service returns None plan
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from ulid import ULID
|
|
|
|
from cleveragents.application.services.checkpoint_service import CheckpointService
|
|
from cleveragents.core.exceptions import BusinessRuleViolation, ResourceNotFoundError
|
|
|
|
# Generate a stable ULID for the rollback test plan
|
|
_ROLLBACK_PLAN_ID = str(ULID())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
@given("the checkpoint service module is imported for coverage boost")
|
|
def step_checkpoint_module_imported(context):
|
|
"""Ensure the checkpoint service module is importable."""
|
|
assert CheckpointService is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared: create a checkpoint service with in-memory storage
|
|
# ---------------------------------------------------------------------------
|
|
@given("a checkpoint service with in-memory storage for coverage boost")
|
|
def step_create_in_memory_service(context):
|
|
"""Create a CheckpointService with no repository (in-memory mode)."""
|
|
context.cp_service = CheckpointService()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: Git CalledProcessError → BusinessRuleViolation (lines 484-487)
|
|
# ---------------------------------------------------------------------------
|
|
@given("subprocess.run is mocked to raise CalledProcessError")
|
|
def step_mock_subprocess_called_process_error(context):
|
|
"""Prepare a mock that raises CalledProcessError."""
|
|
context.subprocess_side_effect = subprocess.CalledProcessError(
|
|
returncode=128,
|
|
cmd=["git", "status"],
|
|
stderr="fatal: not a git repository",
|
|
)
|
|
|
|
|
|
@when('I invoke _run_git with args "{git_args}"')
|
|
def step_invoke_run_git(context, git_args):
|
|
"""Call _run_git with a mocked subprocess.run."""
|
|
args = git_args.split()
|
|
context.run_git_error = None
|
|
with patch("subprocess.run", side_effect=context.subprocess_side_effect):
|
|
try:
|
|
context.cp_service._run_git(args, cwd="/tmp/fake-sandbox")
|
|
except (BusinessRuleViolation, Exception) as exc:
|
|
context.run_git_error = exc
|
|
|
|
|
|
@then('a BusinessRuleViolation should be raised mentioning "{fragment}"')
|
|
def step_verify_business_rule_violation(context, fragment):
|
|
"""Verify a BusinessRuleViolation was raised containing the fragment."""
|
|
error = getattr(context, "run_git_error", None) or getattr(
|
|
context, "rollback_error", None
|
|
)
|
|
assert error is not None, "Expected an exception but none was raised"
|
|
assert isinstance(error, BusinessRuleViolation), (
|
|
f"Expected BusinessRuleViolation, got {type(error).__name__}: {error}"
|
|
)
|
|
assert fragment in str(error), (
|
|
f"Expected '{fragment}' in error message, got: {error}"
|
|
)
|
|
|
|
|
|
@then('the error message should contain the git command "{cmd}"')
|
|
def step_verify_error_contains_git_command(context, cmd):
|
|
"""Verify the error message includes the git subcommand."""
|
|
assert cmd in str(context.run_git_error), (
|
|
f"Expected '{cmd}' in error message, got: {context.run_git_error}"
|
|
)
|
|
|
|
|
|
@then("the error message should include the stderr output")
|
|
def step_verify_error_contains_stderr(context):
|
|
"""Verify the CalledProcessError handler includes stderr in the message."""
|
|
msg = str(context.run_git_error)
|
|
assert "fatal:" in msg or "not a git repository" in msg, (
|
|
f"Expected stderr content in error message, got: {msg}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: Git TimeoutExpired → BusinessRuleViolation (lines 488-491)
|
|
# ---------------------------------------------------------------------------
|
|
@given("subprocess.run is mocked to raise TimeoutExpired")
|
|
def step_mock_subprocess_timeout(context):
|
|
"""Prepare a mock that raises TimeoutExpired."""
|
|
context.subprocess_side_effect = subprocess.TimeoutExpired(
|
|
cmd=["git", "log"],
|
|
timeout=60,
|
|
)
|
|
|
|
|
|
@then('the timeout error message should contain the git command "{cmd}"')
|
|
def step_verify_timeout_error_contains_git_command(context, cmd):
|
|
"""Verify the timeout error message includes the git subcommand."""
|
|
assert context.run_git_error is not None, (
|
|
"Expected an exception but none was raised"
|
|
)
|
|
assert isinstance(context.run_git_error, BusinessRuleViolation), (
|
|
f"Expected BusinessRuleViolation, got {type(context.run_git_error).__name__}"
|
|
)
|
|
assert "timed out" in str(context.run_git_error), (
|
|
f"Expected 'timed out' in error message, got: {context.run_git_error}"
|
|
)
|
|
assert cmd in str(context.run_git_error), (
|
|
f"Expected '{cmd}' in error message, got: {context.run_git_error}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: CalledProcessError during rollback (lines 484-487 via rollback)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a sandbox directory exists at a temporary path")
|
|
def step_create_temp_sandbox(context):
|
|
"""Create a temporary directory with a .git subdirectory to act as sandbox."""
|
|
context.sandbox_tmpdir = tempfile.mkdtemp(prefix="sandbox_test_")
|
|
os.makedirs(os.path.join(context.sandbox_tmpdir, ".git"), exist_ok=True)
|
|
context.rollback_plan_id = _ROLLBACK_PLAN_ID
|
|
context.cp_service.register_sandbox(
|
|
context.rollback_plan_id, context.sandbox_tmpdir
|
|
)
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(context.sandbox_tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@given("a checkpoint was created for the rollback test plan")
|
|
def step_create_checkpoint_for_rollback(context):
|
|
"""Create a checkpoint for the rollback test plan."""
|
|
context.rollback_checkpoint = context.cp_service.create_checkpoint(
|
|
plan_id=context.rollback_plan_id,
|
|
sandbox_ref="abc123deadbeef",
|
|
reason="test checkpoint for rollback",
|
|
)
|
|
|
|
|
|
@given("subprocess.run is mocked to raise CalledProcessError for git diff")
|
|
def step_mock_called_process_error_for_diff(context):
|
|
"""Mock subprocess.run to raise CalledProcessError."""
|
|
context.subprocess_patch = patch(
|
|
"subprocess.run",
|
|
side_effect=subprocess.CalledProcessError(
|
|
returncode=1,
|
|
cmd=["git", "diff", "--name-only"],
|
|
stderr="fatal: bad revision",
|
|
),
|
|
)
|
|
context.subprocess_mock = context.subprocess_patch.start()
|
|
context.add_cleanup(context.subprocess_patch.stop)
|
|
|
|
|
|
@when("I attempt to rollback to the checkpoint")
|
|
def step_attempt_rollback(context):
|
|
"""Attempt rollback and capture any error."""
|
|
context.rollback_error = None
|
|
context.run_git_error = None
|
|
try:
|
|
context.cp_service.rollback_to_checkpoint(
|
|
plan_id=context.rollback_plan_id,
|
|
checkpoint_id=context.rollback_checkpoint.checkpoint_id,
|
|
)
|
|
except Exception as exc:
|
|
context.rollback_error = exc
|
|
context.run_git_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: TimeoutExpired during rollback (lines 488-491 via rollback)
|
|
# ---------------------------------------------------------------------------
|
|
@given("subprocess.run is mocked to raise TimeoutExpired for git reset")
|
|
def step_mock_timeout_for_reset(context):
|
|
"""Mock subprocess.run to raise TimeoutExpired."""
|
|
context.subprocess_patch = patch(
|
|
"subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(
|
|
cmd=["git", "reset", "--hard"],
|
|
timeout=60,
|
|
),
|
|
)
|
|
context.subprocess_mock = context.subprocess_patch.start()
|
|
context.add_cleanup(context.subprocess_patch.stop)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: _resolve_sandbox_path with lifecycle service returning None
|
|
# (line 417)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a checkpoint service with a lifecycle service that returns None for any plan")
|
|
def step_create_service_with_null_lifecycle(context):
|
|
"""Create a CheckpointService with a mocked lifecycle service that returns None."""
|
|
mock_lifecycle = MagicMock()
|
|
mock_lifecycle.get_plan.return_value = None
|
|
context.cp_service = CheckpointService(plan_lifecycle_service=mock_lifecycle)
|
|
|
|
|
|
@when('I attempt to resolve the sandbox path for plan "{plan_id}"')
|
|
def step_attempt_resolve_sandbox(context, plan_id):
|
|
"""Attempt to call _resolve_sandbox_path and capture the error."""
|
|
context.resolve_error = None
|
|
try:
|
|
context.cp_service._resolve_sandbox_path(plan_id)
|
|
except Exception as exc:
|
|
context.resolve_error = exc
|
|
|
|
|
|
@then('a ResourceNotFoundError should be raised for resource type "{resource_type}"')
|
|
def step_verify_resource_not_found(context, resource_type):
|
|
"""Verify a ResourceNotFoundError was raised with the expected resource type."""
|
|
assert context.resolve_error is not None, (
|
|
"Expected an exception but none was raised"
|
|
)
|
|
assert isinstance(context.resolve_error, ResourceNotFoundError), (
|
|
f"Expected ResourceNotFoundError, got {type(context.resolve_error).__name__}: "
|
|
f"{context.resolve_error}"
|
|
)
|
|
assert context.resolve_error.resource_type == resource_type, (
|
|
f"Expected resource_type='{resource_type}', "
|
|
f"got '{context.resolve_error.resource_type}'"
|
|
)
|
|
|
|
|
|
@then('the ResourceNotFoundError should reference plan id "{plan_id}"')
|
|
def step_verify_resource_not_found_plan_id(context, plan_id):
|
|
"""Verify the error references the correct plan ID."""
|
|
assert context.resolve_error.resource_id == plan_id, (
|
|
f"Expected resource_id='{plan_id}', got '{context.resolve_error.resource_id}'"
|
|
)
|