"""Step definitions for devcontainer_health_coverage.feature. These steps target specific uncovered lines in devcontainer_health.py: - Line 157: break in unhealthy-probe path when state changed concurrently - Lines 171, 173-176: ValueError caught during transition after unhealthy probe - Line 185: break in exception path when state changed concurrently - Lines 199, 201-204: ValueError caught during transition after probe exception """ import logging import threading from types import SimpleNamespace from unittest.mock import MagicMock, patch from behave import given, then, when from cleveragents.domain.models.core.container_lifecycle import ( ContainerLifecycleState, ContainerLifecycleTracker, ) from cleveragents.resource.handlers.devcontainer_health import _health_check_loop # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("the devcontainer health module is imported") def step_health_module_imported(context): """Ensure the health module is importable.""" assert _health_check_loop is not None # --------------------------------------------------------------------------- # Shared Given steps # --------------------------------------------------------------------------- @given('a tracker in RUNNING state for resource "{resource_id}"') def step_tracker_running(context, resource_id): """Create a tracker in RUNNING state for use in scenarios.""" context.health_resource_id = resource_id context.health_tracker_running = ContainerLifecycleTracker( resource_id=resource_id, current_state=ContainerLifecycleState.RUNNING, host_workspace_path="/workspace/test", ) context.health_tracker_stopped = ContainerLifecycleTracker( resource_id=resource_id, current_state=ContainerLifecycleState.STOPPED, ) # Defaults context.health_get_tracker_side_effect = None context.health_transition_side_effect = None context.health_run_command = None context.health_debug_messages = [] @given("a run_command that returns a non-zero exit code") def step_run_command_nonzero(context): """Create a run_command mock that returns returncode=1 (unhealthy).""" result = SimpleNamespace(returncode=1, stdout="", stderr="probe failed") context.health_run_command = MagicMock(return_value=result) @given("a run_command that raises an OSError") def step_run_command_raises(context): """Create a run_command mock that raises an OSError.""" context.health_run_command = MagicMock(side_effect=OSError("connection refused")) @given("get_lifecycle_tracker returns STOPPED state on re-read under lock") def step_get_tracker_returns_stopped_on_reread(context): """Configure get_lifecycle_tracker to return RUNNING first, then STOPPED. First call: returns RUNNING tracker (the initial check at line 147). Second call: returns STOPPED tracker (re-read under lock at line 155 or 183). """ call_count = {"n": 0} running = context.health_tracker_running stopped = context.health_tracker_stopped def side_effect(rid): call_count["n"] += 1 if call_count["n"] == 1: return running return stopped context.health_get_tracker_side_effect = side_effect @given("get_lifecycle_tracker always returns RUNNING state") def step_get_tracker_always_running(context): """Configure get_lifecycle_tracker to always return RUNNING.""" running = context.health_tracker_running context.health_get_tracker_side_effect = lambda rid: running @given("transition_state raises ValueError on call") def step_transition_raises_valueerror(context): """Configure transition_state to raise ValueError.""" context.health_transition_side_effect = ValueError( "Invalid lifecycle transition: running -> stopping" ) # --------------------------------------------------------------------------- # When: run the health check loop # --------------------------------------------------------------------------- @when("I run the health check loop for one iteration") def step_run_health_check_loop(context): """Run _health_check_loop with mocked dependencies for a single iteration. Uses a stop_event that is initially unset with a tiny interval so the loop executes one probe, then exits via break (all target paths end with break). """ stop_event = threading.Event() resource_id = context.health_resource_id # Capture debug log messages for assertion captured = context.health_debug_messages class _CapturingHandler(logging.Handler): def emit(self, record): captured.append(record.getMessage()) cap_handler = _CapturingHandler() cap_handler.setLevel(logging.DEBUG) health_logger = logging.getLogger( "cleveragents.resource.handlers.devcontainer_health" ) # Re-enable in case Alembic's fileConfig() disabled it health_logger.disabled = False health_logger.addHandler(cap_handler) original_level = health_logger.level health_logger.setLevel(logging.DEBUG) # Build transition mock if context.health_transition_side_effect is not None: transition_mock = MagicMock(side_effect=context.health_transition_side_effect) else: transition_mock = MagicMock(return_value=context.health_tracker_running) loop_error_holder = [None] loop_done = threading.Event() def run_loop(): try: with ( patch( "cleveragents.resource.handlers.devcontainer_health.get_lifecycle_tracker", side_effect=context.health_get_tracker_side_effect, ), patch( "cleveragents.resource.handlers.devcontainer_health.set_lifecycle_tracker", ) as set_mock_inner, patch( "cleveragents.resource.handlers.devcontainer_health.transition_state", transition_mock, ), ): _health_check_loop( resource_id, interval=0.001, # tiny interval so wait returns immediately stop_event=stop_event, run_command=context.health_run_command, ) context.health_set_tracker_mock = set_mock_inner except Exception as exc: loop_error_holder[0] = exc finally: loop_done.set() t = threading.Thread(target=run_loop, daemon=True) t.start() # Wait up to 5 seconds for the loop to finish finished = loop_done.wait(timeout=5.0) if not finished: # Force stop the loop if it's stuck stop_event.set() t.join(timeout=2.0) context.health_transition_mock = transition_mock context.health_loop_completed = finished and loop_error_holder[0] is None if loop_error_holder[0] is not None: context.health_loop_error = loop_error_holder[0] health_logger.removeHandler(cap_handler) health_logger.setLevel(original_level) # --------------------------------------------------------------------------- # Then: assertions # --------------------------------------------------------------------------- @then("the loop exits via the TOCTOU guard break for unhealthy probe") def step_verify_toctou_break_unhealthy(context): """Verify the loop completed and transition_state was NOT called. Line 157: the break fires because the re-read tracker is no longer RUNNING, so transition_state is never invoked. """ assert context.health_loop_completed, ( f"Loop did not complete: {getattr(context, 'health_loop_error', 'unknown')}" ) context.health_transition_mock.assert_not_called() @then('the ValueError is caught and a debug message is logged for "{resource_id}"') def step_verify_valueerror_caught(context, resource_id): """Verify the loop completed and a debug message about skipped transition was logged. Lines 171-176 and 199-204: ValueError from transition_state is caught and logged. """ assert context.health_loop_completed, ( f"Loop did not complete: {getattr(context, 'health_loop_error', 'unknown')}" ) expected_fragment = "Health check transition skipped for" matched = [m for m in context.health_debug_messages if expected_fragment in m] assert len(matched) > 0, ( f"Expected debug message containing '{expected_fragment}', " f"got: {context.health_debug_messages}" ) @then("the loop exits via the TOCTOU guard break for probe exception") def step_verify_toctou_break_exception(context): """Verify the loop completed and transition_state was NOT called. Line 185: the break fires because the re-read tracker is no longer RUNNING after the probe raised an exception. """ assert context.health_loop_completed, ( f"Loop did not complete: {getattr(context, 'health_loop_error', 'unknown')}" ) context.health_transition_mock.assert_not_called()