"""Step definitions for resource_cli_coverage_boost.feature. Covers the remaining uncovered lines and partial branches in ``src/cleveragents/cli/commands/resource.py``: - Lines 79-81: ``_get_registry_service()`` - Lines 194-195: ``type_add`` FileNotFoundError handler - Lines 234-236: ``type_remove`` user declines confirmation - Lines 263-265: ``type_remove`` NotFoundError from service - Lines 269-271: ``type_remove`` ValidationError from service - Lines 653-658: ``resource_remove`` edge_count > 0 - Lines 668-672: ``resource_remove`` generic Exception → rollback """ from __future__ import annotations import tempfile from typing import Any from unittest.mock import MagicMock, patch from behave import given, then, when # type: ignore[import-untyped] from behave.runner import Context # type: ignore[import-untyped] from typer.testing import CliRunner from cleveragents.core.exceptions import ValidationError # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _PATCH_TARGET = "cleveragents.cli.commands.resource._get_registry_service" _PATCH_CONTAINER = "cleveragents.cli.commands.resource.get_container" _PATCH_CONSOLE = "cleveragents.cli.commands.resource.console" def _no_color_console(): """Return a Rich Console that never emits ANSI escape codes.""" from rich.console import Console as _Console return _Console(no_color=True, highlight=False, force_terminal=False) def _make_mock_type_spec(*, built_in: bool = False) -> MagicMock: """Return a mock ResourceTypeSpec-like object.""" spec = MagicMock() spec.name = "test/mock-type" spec.built_in = built_in spec.description = "mock type" spec.resource_kind = "physical" spec.sandbox_strategy = "copy_on_write" spec.user_addable = True spec.cli_args = [] spec.parent_types = [] spec.child_types = [] spec.handler = None spec.capabilities = [] return spec def _make_mock_resource() -> MagicMock: """Return a mock Resource domain object.""" res = MagicMock() res.resource_id = "01HXYZ1234567890ABCDEFGHIJ" res.name = "local/mock-res" res.resource_type_name = "git-checkout" res.classification = "physical" res.description = "A mock resource" res.location = "/tmp/mock" res.properties = {"path": "/tmp/mock"} res.created_at = "2025-01-01T00:00:00" res.updated_at = "2025-01-01T00:00:00" return res def _smart_query_side_effect(model_map: dict[str, MagicMock]) -> Any: """Return a side_effect for session.query() that dispatches by model class name.""" def _side_effect(model_cls: Any) -> MagicMock: name = getattr(model_cls, "__name__", None) or str(model_cls) if name in model_map: return model_map[name] # Fallback: return a generic mock chain m = MagicMock() return m return _side_effect # --------------------------------------------------------------------------- # Scenario: _get_registry_service delegates to the DI container # --------------------------------------------------------------------------- @given("the DI container is mocked for resource registry") def step_mock_di_container(context: Context) -> None: """Prepare a mock container whose resource_registry_service returns a mock.""" context.rcb_mock_service = MagicMock() context.rcb_mock_container = MagicMock() context.rcb_mock_container.resource_registry_service.return_value = ( context.rcb_mock_service ) @when("_get_registry_service is called directly") def step_call_get_registry_service(context: Context) -> None: """Call the real _get_registry_service with a patched get_container.""" from cleveragents.cli.commands.resource import _get_registry_service with patch(_PATCH_CONTAINER, return_value=context.rcb_mock_container): context.rcb_returned_service = _get_registry_service() @then("the returned service should be the mock registry service") def step_verify_returned_service(context: Context) -> None: assert context.rcb_returned_service is context.rcb_mock_service, ( "Expected _get_registry_service to return the container's service" ) # --------------------------------------------------------------------------- # Scenario: type_add --update re-raises non-"already exists" ValidationError # --------------------------------------------------------------------------- @given( "a mock resource service that raises a non-already-exists ValidationError on register_type" ) def step_mock_service_reraise_validation(context: Context) -> None: svc = MagicMock() svc.register_type.side_effect = ValidationError( message="schema mismatch: bad field" ) context.rcb_mock_service = svc @when("I invoke type add with update flag via CliRunner") def step_invoke_type_add_update(context: Context) -> None: from cleveragents.cli.commands.resource import app runner = CliRunner() # Create a minimal temp YAML so the --config path exists on disk with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: tmp.write("name: dummy\n") tmp.flush() config_path = tmp.name with ( patch(_PATCH_TARGET, return_value=context.rcb_mock_service), patch(_PATCH_CONSOLE, _no_color_console()), ): context.rcb_result = runner.invoke( app, ["type", "add", "--config", config_path, "--update"], ) # --------------------------------------------------------------------------- # Scenario: type_add catches FileNotFoundError from service # --------------------------------------------------------------------------- @given("a mock resource service that raises FileNotFoundError on register_type") def step_mock_service_fnf(context: Context) -> None: svc = MagicMock() svc.register_type.side_effect = FileNotFoundError("no such file: /fake.yaml") context.rcb_mock_service = svc @when("I invoke type add with a dummy config via CliRunner") def step_invoke_type_add_fnf(context: Context) -> None: from cleveragents.cli.commands.resource import app runner = CliRunner() with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: tmp.write("name: dummy\n") tmp.flush() config_path = tmp.name with ( patch(_PATCH_TARGET, return_value=context.rcb_mock_service), patch(_PATCH_CONSOLE, _no_color_console()), ): context.rcb_result = runner.invoke( app, ["type", "add", "--config", config_path], ) # --------------------------------------------------------------------------- # Scenario: type_remove aborts when user declines confirmation prompt # --------------------------------------------------------------------------- @given("a mock resource service with a removable custom type") def step_mock_service_removable_type(context: Context) -> None: svc = MagicMock() svc.show_type.return_value = _make_mock_type_spec(built_in=False) context.rcb_mock_service = svc @when("I invoke type remove without --yes and answer no via CliRunner") def step_invoke_type_remove_no(context: Context) -> None: from cleveragents.cli.commands.resource import app runner = CliRunner() with ( patch(_PATCH_TARGET, return_value=context.rcb_mock_service), patch(_PATCH_CONSOLE, _no_color_console()), ): context.rcb_result = runner.invoke( app, ["type", "remove", "test/mock-type"], input="n\n", ) # --------------------------------------------------------------------------- # Scenario: type_remove aborts when service.remove_type raises NotFoundError # --------------------------------------------------------------------------- @given("a mock resource service whose remove_type raises NotFoundError") def step_mock_service_remove_type_not_found(context: Context) -> None: from cleveragents.core.exceptions import NotFoundError svc = MagicMock() svc.show_type.return_value = _make_mock_type_spec(built_in=False) svc.remove_type.side_effect = NotFoundError( resource_type="resource_type", resource_id="test/mock-type", ) context.rcb_mock_service = svc @when("I invoke type remove with --yes via CliRunner for the phantom type") def step_invoke_type_remove_not_found(context: Context) -> None: from cleveragents.cli.commands.resource import app runner = CliRunner() with ( patch(_PATCH_TARGET, return_value=context.rcb_mock_service), patch(_PATCH_CONSOLE, _no_color_console()), ): context.rcb_result = runner.invoke( app, ["type", "remove", "test/mock-type", "--yes"], ) # --------------------------------------------------------------------------- # Scenario: type_remove aborts when service.remove_type raises ValidationError # --------------------------------------------------------------------------- @given("a mock resource service whose remove_type raises ValidationError") def step_mock_service_remove_type_validation_error(context: Context) -> None: svc = MagicMock() svc.show_type.return_value = _make_mock_type_spec(built_in=False) svc.remove_type.side_effect = ValidationError( message="Cannot remove type: 5 resource(s) still reference it.", details={"name": "test/mock-type"}, ) context.rcb_mock_service = svc @when("I invoke type remove with --yes via CliRunner for the failing type") def step_invoke_type_remove_validation_error(context: Context) -> None: from cleveragents.cli.commands.resource import app runner = CliRunner() with ( patch(_PATCH_TARGET, return_value=context.rcb_mock_service), patch(_PATCH_CONSOLE, _no_color_console()), ): context.rcb_result = runner.invoke( app, ["type", "remove", "test/mock-type", "--yes"], ) # --------------------------------------------------------------------------- # Scenario: resource_remove aborts when resource has edges # --------------------------------------------------------------------------- @given("a mock resource service whose session reports edges on the resource") def step_mock_service_resource_edges(context: Context) -> None: svc = MagicMock() mock_res = _make_mock_resource() svc.show_resource.return_value = mock_res mock_session = MagicMock() # ResourceEdgeModel query → count returns 3 (edges exist) edge_query = MagicMock() edge_query.filter.return_value.count.return_value = 3 mock_session.query.side_effect = _smart_query_side_effect( { "ResourceEdgeModel": edge_query, } ) svc._session.return_value = mock_session context.rcb_mock_service = svc @when("I invoke resource remove with --yes via CliRunner for the edged resource") def step_invoke_resource_remove_edges(context: Context) -> None: from cleveragents.cli.commands.resource import app runner = CliRunner() with ( patch(_PATCH_TARGET, return_value=context.rcb_mock_service), patch(_PATCH_CONSOLE, _no_color_console()), ): context.rcb_result = runner.invoke( app, ["remove", "local/mock-res", "--yes"], ) # --------------------------------------------------------------------------- # Scenario: resource_remove rolls back session on unexpected exception # --------------------------------------------------------------------------- @given( "a mock resource service whose session delete raises a generic exception for resource" ) def step_mock_service_resource_delete_exception(context: Context) -> None: svc = MagicMock() mock_res = _make_mock_resource() svc.show_resource.return_value = mock_res mock_session = MagicMock() # ResourceEdgeModel query → count returns 0 (no edges) edge_query = MagicMock() edge_query.filter.return_value.count.return_value = 0 # ResourceModel query → first returns a mock row mock_row = MagicMock() resource_query = MagicMock() resource_query.filter_by.return_value.first.return_value = mock_row mock_session.query.side_effect = _smart_query_side_effect( { "ResourceEdgeModel": edge_query, "ResourceModel": resource_query, } ) # session.delete raises a generic exception mock_session.delete.side_effect = RuntimeError("unexpected DB write failure") svc._session.return_value = mock_session context.rcb_mock_service = svc context.rcb_mock_session = mock_session @when("I invoke resource remove with --yes via CliRunner for the failing resource") def step_invoke_resource_remove_exception(context: Context) -> None: from cleveragents.cli.commands.resource import app runner = CliRunner() with ( patch(_PATCH_TARGET, return_value=context.rcb_mock_service), patch(_PATCH_CONSOLE, _no_color_console()), ): context.rcb_result = runner.invoke( app, ["remove", "local/mock-res", "--yes"], ) @then("the mock session rollback should have been called for resource remove") def step_verify_resource_rollback(context: Context) -> None: context.rcb_mock_session.rollback.assert_called() # --------------------------------------------------------------------------- # Shared assertion steps # --------------------------------------------------------------------------- @then("the CliRunner exit code should be non-zero") def step_exit_code_nonzero(context: Context) -> None: assert context.rcb_result.exit_code != 0, ( f"Expected non-zero exit code, got {context.rcb_result.exit_code}.\n" f"Output: {context.rcb_result.output}" ) @then('the CliRunner output should contain "{text}"') def step_cli_output_contains(context: Context, text: str) -> None: output = context.rcb_result.output assert text.lower() in output.lower(), ( f"Expected '{text}' in CliRunner output, got:\n{output}" )