forked from HAL9000/cleveragents-core
31472b5413
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
878 lines
31 KiB
Python
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}"
|
|
)
|