Files
placeholder/features/steps/resource_cli_coverage_r3_steps.py
freemo 31472b5413 test(coverage): add Behave scenarios for 39 under-covered modules
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate.

ISSUES CLOSED: #1232
2026-03-31 21:47:12 +00:00

878 lines
31 KiB
Python

"""Step definitions for resource_cli_coverage_r3.feature.
Covers the remaining uncovered lines in
``src/cleveragents/cli/commands/resource.py``:
- Lines 159, 162: _resource_dict str() fallback for dates
- Lines 280-281: type_remove ResourceTypeParentRemovalError
- Lines 288-289: type_remove generic Exception
- Lines 310-311: type_list empty
- Lines 347-351: type_list generic Exception
- Lines 388-392: type_show generic Exception
- Lines 420-422: _print_type_panel chain resolution failure
- Lines 465-471: _format_properties mounts JSON
- Lines 622, 624-626, 629-630: resource_add image / mount validation
- Lines 729-733: resource_list generic Exception
- Lines 786-790: resource_show generic Exception
- Lines 847-851: resource_tree generic Exception
- Line 950: resource_inspect empty tree_data
- Lines 1004-1008: resource_inspect generic Exception
- Lines 1019, 1028: _read_resource_file path from props / traversal
- Lines 1091-1095: resource_link_child generic Exception
- Lines 1162-1163: resource_unlink_child generic Exception
- Lines 1314-1316: resource_stop confirmation prompt
- Lines 1333-1340: resource_stop CleverAgentsError / generic Exception
- Lines 1389-1390: resource_rebuild path from properties
- Lines 1397-1399: resource_rebuild confirmation prompt
- Lines 1412-1419: resource_rebuild CleverAgentsError / generic Exception
"""
from __future__ import annotations
import json
import os
import tempfile
from io import StringIO
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 rich.console import Console
from cleveragents.core.exceptions import CleverAgentsError
from cleveragents.domain.models.core.container_lifecycle import (
ContainerLifecycleState,
)
from cleveragents.resource.inheritance import ResourceTypeParentRemovalError
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_PATCH_SVC = "cleveragents.cli.commands.resource._get_registry_service"
_PATCH_CONSOLE = "cleveragents.cli.commands.resource.console"
_PATCH_STOP = "cleveragents.cli.commands.resource.stop_container"
_PATCH_REBUILD = "cleveragents.cli.commands.resource.rebuild_container"
_PATCH_TRACKER = "cleveragents.cli.commands.resource.get_lifecycle_tracker"
def _plain_console() -> Console:
"""A Console that never emits ANSI codes."""
return Console(no_color=True, highlight=False, force_terminal=False)
def _capture(func: Any, *args: Any, **kwargs: Any) -> tuple[str, bool]:
"""Run a CLI function, capture its rich console output and success flag."""
import contextlib
buf = StringIO()
console = Console(
file=buf, width=200, no_color=True, highlight=False, force_terminal=False
)
import cleveragents.cli.commands.resource as mod
orig = mod.console
mod.console = console
failed = False
try:
with contextlib.redirect_stdout(buf):
func(*args, **kwargs)
except SystemExit:
failed = True
except Exception:
failed = True
finally:
mod.console = orig
return buf.getvalue(), failed
def _mock_resource(
*,
type_name: str = "git-checkout",
location: str | None = "/tmp/mock",
properties: dict[str, Any] | None = None,
resource_id: str = "01HXYZ1234567890ABCDEFGHIJ",
name: str = "local/mock-res",
) -> MagicMock:
"""Build a mock Resource domain object."""
res = MagicMock()
res.resource_id = resource_id
res.name = name
res.resource_type_name = type_name
res.classification = "physical"
res.description = "A mock resource"
res.location = location
res.properties = properties if properties is not None else {"path": "/tmp/mock"}
res.created_at = "2025-01-01T00:00:00"
res.updated_at = "2025-01-01T00:00:00"
return res
def _mock_type_spec(
*, built_in: bool = False, inherits: str | None = None
) -> MagicMock:
"""Build a mock ResourceTypeSpec."""
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 = []
spec.inherits = inherits
spec.equivalence = None
return spec
# ===================================================================
# _resource_dict: str() fallback for dates (lines 159, 162)
# ===================================================================
@given("rccov3 a mock resource whose dates lack isoformat")
def step_rccov3_resource_no_isoformat(context: Context) -> None:
"""Create a mock resource whose created_at/updated_at are plain ints (no isoformat)."""
res = _mock_resource()
# Replace datetime-like attrs with plain ints that have no isoformat method
res.created_at = 1234567890
res.updated_at = 9876543210
context.rccov3_resource = res
@when("rccov3 I call _resource_dict on it")
def step_rccov3_call_resource_dict(context: Context) -> None:
from cleveragents.cli.commands.resource import _resource_dict
context.rccov3_result = _resource_dict(context.rccov3_resource)
@then("rccov3 the result created_at should be a plain string")
def step_rccov3_created_at_str(context: Context) -> None:
val = context.rccov3_result["created_at"]
assert val == "1234567890", f"Expected '1234567890', got {val!r}"
@then("rccov3 the result updated_at should be a plain string")
def step_rccov3_updated_at_str(context: Context) -> None:
val = context.rccov3_result["updated_at"]
assert val == "9876543210", f"Expected '9876543210', got {val!r}"
# ===================================================================
# type_remove: ResourceTypeParentRemovalError (lines 280-281)
# ===================================================================
@given("rccov3 a service whose remove_type raises ResourceTypeParentRemovalError")
def step_rccov3_svc_parent_removal(context: Context) -> None:
svc = MagicMock()
svc.show_type.return_value = _mock_type_spec(built_in=False)
svc.remove_type.side_effect = ResourceTypeParentRemovalError(
"Cannot remove: type has subtypes"
)
context.rccov3_svc = svc
@when("rccov3 I invoke type_remove with yes flag")
def step_rccov3_invoke_type_remove_yes(context: Context) -> None:
from cleveragents.cli.commands.resource import type_remove
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(type_remove, name="test/mock-type", yes=True)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# type_remove: generic Exception (lines 288-289)
# ===================================================================
@given("rccov3 a service whose remove_type raises a RuntimeError")
def step_rccov3_svc_remove_runtime(context: Context) -> None:
svc = MagicMock()
svc.show_type.return_value = _mock_type_spec(built_in=False)
svc.remove_type.side_effect = RuntimeError("boom")
context.rccov3_svc = svc
# (reuses: rccov3 I invoke type_remove with yes flag)
# ===================================================================
# type_list: empty list (lines 310-311)
# ===================================================================
@given("rccov3 a service whose list_types returns empty")
def step_rccov3_svc_list_empty(context: Context) -> None:
svc = MagicMock()
svc.list_types.return_value = []
context.rccov3_svc = svc
@when("rccov3 I invoke type_list")
def step_rccov3_invoke_type_list(context: Context) -> None:
from cleveragents.cli.commands.resource import type_list
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(type_list, fmt="rich")
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# type_list: generic Exception (lines 347-351)
# ===================================================================
@given("rccov3 a service whose list_types raises a RuntimeError")
def step_rccov3_svc_list_runtime(context: Context) -> None:
svc = MagicMock()
svc.list_types.side_effect = RuntimeError("boom")
context.rccov3_svc = svc
# (reuses: rccov3 I invoke type_list)
# ===================================================================
# type_show: generic Exception (lines 388-392)
# ===================================================================
@given("rccov3 a service whose show_type raises a RuntimeError")
def step_rccov3_svc_show_type_runtime(context: Context) -> None:
svc = MagicMock()
svc.show_type.side_effect = RuntimeError("boom")
context.rccov3_svc = svc
@when('rccov3 I invoke type_show for "{name}"')
def step_rccov3_invoke_type_show(context: Context, name: str) -> None:
from cleveragents.cli.commands.resource import type_show
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(type_show, name=name, fmt="rich")
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# _print_type_panel: chain resolution failure (lines 420-422)
# ===================================================================
@given("rccov3 a type spec that inherits from another type")
def step_rccov3_spec_inherits(context: Context) -> None:
context.rccov3_spec = _mock_type_spec(inherits="base-type")
@given("rccov3 a service whose resolve_type_inheritance_chain raises an exception")
def step_rccov3_svc_chain_failure(context: Context) -> None:
svc = MagicMock()
svc.resolve_type_inheritance_chain.side_effect = RuntimeError("chain broken")
context.rccov3_svc = svc
@when("rccov3 I call _print_type_panel on the spec")
def step_rccov3_call_print_type_panel(context: Context) -> None:
from cleveragents.cli.commands.resource import _print_type_panel
buf = StringIO()
console = Console(
file=buf, width=200, no_color=True, highlight=False, force_terminal=False
)
import cleveragents.cli.commands.resource as mod
orig = mod.console
mod.console = console
try:
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
_print_type_panel(context.rccov3_spec)
finally:
mod.console = orig
context.rccov3_output = buf.getvalue()
@then('rccov3 the panel output should contain "..."')
def step_rccov3_panel_contains_ellipsis(context: Context) -> None:
assert "..." in context.rccov3_output, (
f"Expected '...' in panel output, got:\n{context.rccov3_output}"
)
# ===================================================================
# _format_properties: mounts JSON rendering (lines 465-471)
# ===================================================================
@when("rccov3 I call _format_properties with valid mounts JSON")
def step_rccov3_format_props_mounts(context: Context) -> None:
from cleveragents.cli.commands.resource import _format_properties
mounts = [
{"source": "local/api-repo", "target": "/workspace", "mode": "rw"},
{"source": "/var/config", "target": "/config"},
]
props = {"mounts": json.dumps(mounts), "other_key": "other_val"}
context.rccov3_formatted = _format_properties(props)
@then('rccov3 the formatted output should contain "mount:"')
def step_rccov3_formatted_mount(context: Context) -> None:
assert "mount:" in context.rccov3_formatted, (
f"Expected 'mount:' in output:\n{context.rccov3_formatted}"
)
@then('rccov3 the formatted output should contain "/workspace"')
def step_rccov3_formatted_workspace(context: Context) -> None:
assert "/workspace" in context.rccov3_formatted, (
f"Expected '/workspace' in output:\n{context.rccov3_formatted}"
)
@when("rccov3 I call _format_properties with invalid mounts JSON")
def step_rccov3_format_props_bad_mounts(context: Context) -> None:
from cleveragents.cli.commands.resource import _format_properties
props = {"mounts": "not-valid-json{{{"}
context.rccov3_formatted = _format_properties(props)
@then('rccov3 the formatted output should contain "mounts:"')
def step_rccov3_formatted_mounts_key(context: Context) -> None:
assert "mounts:" in context.rccov3_formatted, (
f"Expected 'mounts:' in output:\n{context.rccov3_formatted}"
)
# ===================================================================
# resource_add: image property (line 622)
# ===================================================================
@given("rccov3 a service that successfully registers a resource")
def step_rccov3_svc_register_ok(context: Context) -> None:
svc = MagicMock()
svc.register_resource.return_value = _mock_resource(
type_name="container-instance",
name="local/img-test",
)
context.rccov3_svc = svc
@when('rccov3 I invoke resource_add with image "ubuntu:latest"')
def step_rccov3_invoke_add_image(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_add
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(
resource_add,
type_name="container-instance",
name="local/img-test",
path=None,
branch=None,
description=None,
image="ubuntu:latest",
mount=None,
clone_into=None,
read_only=False,
fmt="rich",
)
context.rccov3_output = output
context.rccov3_failed = failed
@then("rccov3 the service register_resource was called with image in properties")
def step_rccov3_verify_image_prop(context: Context) -> None:
call_kwargs = context.rccov3_svc.register_resource.call_args
props = call_kwargs.kwargs.get("properties") or call_kwargs[1].get("properties")
assert props is not None and "image" in props, (
f"Expected 'image' in properties, got: {props}"
)
assert props["image"] == "ubuntu:latest"
# ===================================================================
# resource_add: mount on non-container type (lines 624-626, 629-630)
# ===================================================================
@when("rccov3 I invoke resource_add with mount on git-checkout type")
def step_rccov3_invoke_add_mount_bad_type(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_add
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(
resource_add,
type_name="git-checkout",
name="local/bad-mount",
path="/tmp/x",
branch=None,
description=None,
image=None,
mount=["local/repo:/workspace"],
clone_into=None,
read_only=False,
fmt="rich",
)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_list: generic Exception (lines 729-733)
# ===================================================================
@given("rccov3 a service whose list_resources raises a RuntimeError")
def step_rccov3_svc_list_resources_runtime(context: Context) -> None:
svc = MagicMock()
svc.list_resources.side_effect = RuntimeError("boom")
context.rccov3_svc = svc
@when("rccov3 I invoke resource_list")
def step_rccov3_invoke_resource_list(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_list
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(resource_list, type_filter=None, fmt="rich")
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_show: generic Exception (lines 786-790)
# ===================================================================
@given("rccov3 a service whose show_resource raises a RuntimeError")
def step_rccov3_svc_show_resource_runtime(context: Context) -> None:
svc = MagicMock()
svc.show_resource.side_effect = RuntimeError("boom")
context.rccov3_svc = svc
@when('rccov3 I invoke resource_show for "{name}"')
def step_rccov3_invoke_resource_show(context: Context, name: str) -> None:
from cleveragents.cli.commands.resource import resource_show
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(resource_show, resource=name, fmt="rich")
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_tree: generic Exception (lines 847-851)
# ===================================================================
@given("rccov3 a service whose get_resource_tree raises a RuntimeError")
def step_rccov3_svc_tree_runtime(context: Context) -> None:
svc = MagicMock()
svc.get_resource_tree.side_effect = RuntimeError("boom")
context.rccov3_svc = svc
@when('rccov3 I invoke resource_tree for "{name}"')
def step_rccov3_invoke_resource_tree(context: Context, name: str) -> None:
from cleveragents.cli.commands.resource import resource_tree
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(
resource_tree, resource=name, depth=-1, type_filter=None, fmt="rich"
)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_inspect: empty tree_data (line 950)
# ===================================================================
@given(
"rccov3 a service whose show_resource returns a mock and get_resource_tree returns empty"
)
def step_rccov3_svc_inspect_empty_tree(context: Context) -> None:
svc = MagicMock()
svc.show_resource.return_value = _mock_resource()
svc.get_resource_tree.return_value = [] # empty tree
context.rccov3_svc = svc
@when("rccov3 I invoke resource_inspect with tree flag and json format")
def step_rccov3_invoke_inspect_tree_json(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_inspect
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(
resource_inspect,
resource="local/mock-res",
tree=True,
file=None,
fmt="json",
)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_inspect: generic Exception (lines 1004-1008)
# ===================================================================
@when('rccov3 I invoke resource_inspect for "{name}"')
def step_rccov3_invoke_resource_inspect(context: Context, name: str) -> None:
from cleveragents.cli.commands.resource import resource_inspect
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(
resource_inspect, resource=name, tree=False, file=None, fmt="rich"
)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# _read_resource_file: path from properties (line 1019)
# ===================================================================
@given("rccov3 a mock resource with no location but properties path")
def step_rccov3_resource_no_location(context: Context) -> None:
# Create a temp directory with a test file
tmpdir = tempfile.mkdtemp()
test_file = os.path.join(tmpdir, "hello.txt")
with open(test_file, "w") as f:
f.write("hello from test")
context.rccov3_resource = _mock_resource(
location=None,
properties={"path": tmpdir},
)
context.rccov3_tmpdir = tmpdir
@when("rccov3 I call _read_resource_file with a valid relative path")
def step_rccov3_call_read_resource_file(context: Context) -> None:
from cleveragents.cli.commands.resource import _read_resource_file
context.rccov3_file_result = _read_resource_file(
context.rccov3_resource, "hello.txt"
)
@then("rccov3 the file read result should contain expected content")
def step_rccov3_file_result_ok(context: Context) -> None:
assert "hello from test" in context.rccov3_file_result, (
f"Expected 'hello from test', got: {context.rccov3_file_result!r}"
)
# ===================================================================
# _read_resource_file: path traversal rejection (line 1028)
# ===================================================================
@given("rccov3 a mock resource with a known location")
def step_rccov3_resource_known_location(context: Context) -> None:
tmpdir = tempfile.mkdtemp()
context.rccov3_resource = _mock_resource(location=tmpdir)
@when("rccov3 I call _read_resource_file with a traversal path")
def step_rccov3_call_read_file_traversal(context: Context) -> None:
from cleveragents.cli.commands.resource import _read_resource_file
context.rccov3_file_result = _read_resource_file(
context.rccov3_resource, "../../etc/passwd"
)
@then('rccov3 the file read result should contain "path traversal rejected"')
def step_rccov3_file_traversal_rejected(context: Context) -> None:
assert "path traversal rejected" in context.rccov3_file_result, (
f"Expected 'path traversal rejected', got: {context.rccov3_file_result!r}"
)
# ===================================================================
# resource_link_child: generic Exception (lines 1091-1095)
# ===================================================================
@given("rccov3 a service whose link_child raises a RuntimeError")
def step_rccov3_svc_link_runtime(context: Context) -> None:
svc = MagicMock()
svc.link_child.side_effect = RuntimeError("boom")
context.rccov3_svc = svc
@when("rccov3 I invoke resource_link_child")
def step_rccov3_invoke_link_child(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_link_child
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(
resource_link_child, parent="local/a", child="local/b", fmt="rich"
)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_unlink_child: generic Exception (lines 1162-1163)
# ===================================================================
@given("rccov3 a service whose unlink_child raises a RuntimeError")
def step_rccov3_svc_unlink_runtime(context: Context) -> None:
svc = MagicMock()
svc.unlink_child.side_effect = RuntimeError("boom")
context.rccov3_svc = svc
@when("rccov3 I invoke resource_unlink_child with yes flag")
def step_rccov3_invoke_unlink_child(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_unlink_child
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(
resource_unlink_child,
parent="local/a",
child="local/b",
yes=True,
fmt="rich",
)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_stop: confirmation prompt (lines 1314-1316)
# ===================================================================
def _make_running_devcontainer_svc() -> tuple[MagicMock, MagicMock]:
"""Return (service, tracker) for a running devcontainer resource."""
svc = MagicMock()
res = _mock_resource(type_name="devcontainer-instance", name="local/dc")
svc.show_resource.return_value = res
tracker = MagicMock()
tracker.current_state = ContainerLifecycleState.RUNNING
return svc, tracker
@given("rccov3 a running devcontainer resource for stop")
def step_rccov3_running_dc_stop(context: Context) -> None:
svc, tracker = _make_running_devcontainer_svc()
context.rccov3_svc = svc
context.rccov3_tracker = tracker
@when("rccov3 I invoke resource_stop without yes flag and confirm")
def step_rccov3_invoke_stop_confirm(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_stop
with (
patch(_PATCH_SVC, return_value=context.rccov3_svc),
patch(_PATCH_TRACKER, return_value=context.rccov3_tracker),
patch(_PATCH_STOP),
patch("cleveragents.cli.commands.resource.typer.confirm"),
):
output, failed = _capture(resource_stop, name="local/dc", yes=False)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_stop: CleverAgentsError (lines 1333-1335)
# ===================================================================
@given("rccov3 a service whose show_resource raises CleverAgentsError")
def step_rccov3_svc_show_clever_error(context: Context) -> None:
svc = MagicMock()
svc.show_resource.side_effect = CleverAgentsError(message="service failure")
context.rccov3_svc = svc
@when('rccov3 I invoke resource_stop for "{name}"')
def step_rccov3_invoke_stop(context: Context, name: str) -> None:
from cleveragents.cli.commands.resource import resource_stop
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(resource_stop, name=name, yes=True)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_stop: generic Exception (lines 1336-1340)
# — reuses "rccov3 a service whose show_resource raises a RuntimeError"
# — reuses "rccov3 I invoke resource_stop for ..."
# ===================================================================
# ===================================================================
# resource_stop: generic Exception (lines 1336-1340)
# — needs a non-RuntimeError, non-ValueError, non-CleverAgentsError
# exception so it falls through to the bare except handler.
# ===================================================================
@given("rccov3 a service whose show_resource raises a TypeError for stop")
def step_rccov3_svc_show_typeerror_stop(context: Context) -> None:
svc = MagicMock()
svc.show_resource.side_effect = TypeError("unexpected type")
context.rccov3_svc = svc
# (reuses: rccov3 I invoke resource_stop for ...)
# ===================================================================
# resource_rebuild: generic Exception (lines 1415-1419)
# ===================================================================
@given("rccov3 a service whose show_resource raises a TypeError for rebuild")
def step_rccov3_svc_show_typeerror_rebuild(context: Context) -> None:
svc = MagicMock()
svc.show_resource.side_effect = TypeError("unexpected type")
context.rccov3_svc = svc
# (reuses: rccov3 I invoke resource_rebuild for ...)
# ===================================================================
# resource_rebuild: path from properties (lines 1389-1390)
# ===================================================================
def _make_stopped_devcontainer_svc(
*, location: str | None = "/project", properties: dict[str, Any] | None = None
) -> tuple[MagicMock, MagicMock]:
"""Return (service, tracker) for a stopped devcontainer resource."""
svc = MagicMock()
props = properties if properties is not None else {"path": "/project"}
res = _mock_resource(
type_name="devcontainer-instance",
name="local/dc",
location=location,
properties=props,
)
svc.show_resource.return_value = res
tracker = MagicMock()
tracker.current_state = ContainerLifecycleState.STOPPED
return svc, tracker
@given("rccov3 a stopped devcontainer resource with no location but properties path")
def step_rccov3_stopped_dc_no_location(context: Context) -> None:
svc, tracker = _make_stopped_devcontainer_svc(
location=None,
properties={"path": "/project/from/props"},
)
context.rccov3_svc = svc
context.rccov3_tracker = tracker
@when("rccov3 I invoke resource_rebuild with yes flag")
def step_rccov3_invoke_rebuild_yes(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_rebuild
with (
patch(_PATCH_SVC, return_value=context.rccov3_svc),
patch(_PATCH_TRACKER, return_value=context.rccov3_tracker),
patch(_PATCH_REBUILD),
):
output, failed = _capture(resource_rebuild, name="local/dc", yes=True)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_rebuild: confirmation prompt (lines 1397-1399)
# ===================================================================
@given("rccov3 a stopped devcontainer resource for rebuild")
def step_rccov3_stopped_dc_rebuild(context: Context) -> None:
svc, tracker = _make_stopped_devcontainer_svc()
context.rccov3_svc = svc
context.rccov3_tracker = tracker
@when("rccov3 I invoke resource_rebuild without yes flag and confirm")
def step_rccov3_invoke_rebuild_confirm(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_rebuild
with (
patch(_PATCH_SVC, return_value=context.rccov3_svc),
patch(_PATCH_TRACKER, return_value=context.rccov3_tracker),
patch(_PATCH_REBUILD),
patch("cleveragents.cli.commands.resource.typer.confirm"),
):
output, failed = _capture(resource_rebuild, name="local/dc", yes=False)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_rebuild: CleverAgentsError (lines 1412-1414)
# ===================================================================
# Reuses: "rccov3 a service whose show_resource raises CleverAgentsError"
@when('rccov3 I invoke resource_rebuild for "{name}"')
def step_rccov3_invoke_rebuild(context: Context, name: str) -> None:
from cleveragents.cli.commands.resource import resource_rebuild
with patch(_PATCH_SVC, return_value=context.rccov3_svc):
output, failed = _capture(resource_rebuild, name=name, yes=True)
context.rccov3_output = output
context.rccov3_failed = failed
# ===================================================================
# resource_rebuild: generic Exception (lines 1415-1419)
# — reuses "rccov3 a service whose show_resource raises a RuntimeError"
# — reuses "rccov3 I invoke resource_rebuild for ..."
# ===================================================================
# ===================================================================
# Shared Then steps
# ===================================================================
@then("rccov3 the command should have failed")
def step_rccov3_failed(context: Context) -> None:
assert context.rccov3_failed, (
f"Expected command to fail, output was:\n{context.rccov3_output}"
)
@then('rccov3 the output should contain "{text}"')
def step_rccov3_output_contains(context: Context, text: str) -> None:
output = context.rccov3_output
assert text.lower() in output.lower(), (
f"Expected '{text}' in output, got:\n{output}"
)