866 lines
28 KiB
Python
866 lines
28 KiB
Python
"""Step definitions for Resource CLI feature tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
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 sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from cleveragents.application.services.resource_registry_service import (
|
|
ResourceRegistryService,
|
|
)
|
|
from cleveragents.core.exceptions import (
|
|
CleverAgentsError,
|
|
NotFoundError,
|
|
ValidationError,
|
|
)
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
|
|
def _make_service(context: Context) -> ResourceRegistryService:
|
|
"""Create or return a cached ResourceRegistryService."""
|
|
if not hasattr(context, "resource_cli_service"):
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
context.resource_cli_service = ResourceRegistryService(session_factory=factory)
|
|
return context.resource_cli_service
|
|
|
|
|
|
def _capture_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, bool]:
|
|
"""Run a CLI function capturing its console output and success status."""
|
|
import contextlib
|
|
|
|
buf = StringIO()
|
|
# force_terminal=False and highlight=False prevent Rich from emitting
|
|
# ANSI escape codes (bold, syntax highlighting) when FORCE_COLOR is set
|
|
# or a real colour terminal is detected.
|
|
console = Console(
|
|
file=buf, width=200, no_color=True, highlight=False, force_terminal=False
|
|
)
|
|
|
|
import cleveragents.cli.commands.resource as resource_mod
|
|
|
|
orig_console = resource_mod.console
|
|
resource_mod.console = console
|
|
|
|
failed = False
|
|
try:
|
|
with contextlib.redirect_stdout(buf):
|
|
func(*args, **kwargs)
|
|
except SystemExit:
|
|
failed = True
|
|
except Exception:
|
|
failed = True
|
|
finally:
|
|
resource_mod.console = orig_console
|
|
|
|
return buf.getvalue(), failed
|
|
|
|
|
|
@given("a fresh in-memory resource registry")
|
|
def step_fresh_registry(context: Context) -> None:
|
|
"""Set up a fresh in-memory database for resource tests."""
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
context.resource_cli_service = ResourceRegistryService(session_factory=factory)
|
|
context.resource_cli_output = ""
|
|
context.resource_cli_failed = False
|
|
|
|
|
|
@given("built-in types are bootstrapped")
|
|
def step_bootstrap_builtins(context: Context) -> None:
|
|
"""Bootstrap built-in resource types."""
|
|
service = _make_service(context)
|
|
service.bootstrap_builtin_types()
|
|
|
|
|
|
@given("a valid custom resource type YAML file")
|
|
def step_create_custom_yaml(context: Context) -> None:
|
|
"""Create a temporary YAML file for a custom resource type."""
|
|
yaml_content = """
|
|
name: test/custom-type
|
|
description: A test custom resource type
|
|
resource_kind: physical
|
|
sandbox_strategy: copy_on_write
|
|
user_addable: true
|
|
cli_args:
|
|
- name: path
|
|
type: path
|
|
required: true
|
|
description: Path to the resource
|
|
"""
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp:
|
|
tmp.write(yaml_content)
|
|
tmp.flush()
|
|
context.resource_cli_yaml_path = tmp.name
|
|
|
|
|
|
def _patch_service(context: Context) -> Any:
|
|
"""Monkey-patch the container to return our in-memory service."""
|
|
import cleveragents.cli.commands.resource as resource_mod
|
|
|
|
orig_fn = resource_mod._get_registry_service
|
|
|
|
def _mock_get() -> ResourceRegistryService:
|
|
return _make_service(context)
|
|
|
|
resource_mod._get_registry_service = _mock_get
|
|
return orig_fn
|
|
|
|
|
|
def _unpatch_service(orig_fn: Any) -> None:
|
|
"""Restore original service getter."""
|
|
import cleveragents.cli.commands.resource as resource_mod
|
|
|
|
resource_mod._get_registry_service = orig_fn
|
|
|
|
|
|
# ---- Resource Type List ----
|
|
|
|
|
|
@when("I run resource type list")
|
|
def step_run_type_list(context: Context) -> None:
|
|
"""Run resource type list command."""
|
|
from cleveragents.cli.commands.resource import type_list
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(type_list, fmt="rich")
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when('I run resource type list with format "{fmt}"')
|
|
def step_run_type_list_fmt(context: Context, fmt: str) -> None:
|
|
"""Run resource type list with specific format."""
|
|
from cleveragents.cli.commands.resource import type_list
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(type_list, fmt=fmt)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
# ---- Resource Type Show ----
|
|
|
|
|
|
@when('I run resource type show "{name}" using default format')
|
|
def step_run_type_show(context: Context, name: str) -> None:
|
|
"""Run resource type show command."""
|
|
from cleveragents.cli.commands.resource import type_show
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(type_show, name=name, fmt="rich")
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when('I run resource type show "{name}" using format "{fmt}"')
|
|
def step_run_type_show_fmt(context: Context, name: str, fmt: str) -> None:
|
|
"""Run resource type show with specific format."""
|
|
from cleveragents.cli.commands.resource import type_show
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(type_show, name=name, fmt=fmt)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
# ---- Resource Type Add ----
|
|
|
|
|
|
def _do_type_add(context: Context) -> None:
|
|
"""Run resource type add with the YAML config file (shared impl)."""
|
|
from cleveragents.cli.commands.resource import type_add
|
|
|
|
config_path = Path(context.resource_cli_yaml_path)
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
type_add, config=config_path, update=False, fmt="rich"
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when("I run resource type add with the config file")
|
|
def step_run_type_add(context: Context) -> None:
|
|
"""Run resource type add with the YAML config file."""
|
|
_do_type_add(context)
|
|
|
|
|
|
@given("I run resource type add with the config file")
|
|
def step_given_type_add(context: Context) -> None:
|
|
"""Run resource type add in Given context."""
|
|
_do_type_add(context)
|
|
|
|
|
|
@when("I run resource type add with a nonexistent config file")
|
|
def step_run_type_add_nonexistent(context: Context) -> None:
|
|
"""Run resource type add with a nonexistent file."""
|
|
from cleveragents.cli.commands.resource import type_add
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
type_add,
|
|
config=Path("/tmp/nonexistent_resource_type.yaml"),
|
|
update=False,
|
|
fmt="rich",
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
# ---- Resource Type Remove ----
|
|
|
|
|
|
@when('I run resource type remove "{name}" with yes flag')
|
|
def step_run_type_remove(context: Context, name: str) -> None:
|
|
"""Run resource type remove with --yes flag."""
|
|
from cleveragents.cli.commands.resource import type_remove
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(type_remove, name=name, yes=True)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
# ---- Resource Add ----
|
|
|
|
|
|
def _do_resource_add(
|
|
context: Context,
|
|
type_name: str,
|
|
name: str,
|
|
path: str | None,
|
|
*,
|
|
branch: str | None = None,
|
|
description: str | None = None,
|
|
image: str | None = None,
|
|
container_id: str | None = None,
|
|
mount: list[str] | None = None,
|
|
url: str | None = None,
|
|
clone_into: str | None = None,
|
|
read_only: bool = False,
|
|
update: bool = False,
|
|
fmt: str | None = None,
|
|
) -> None:
|
|
"""Run resource add command (shared impl)."""
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
resolved_fmt = fmt or getattr(context, "resource_cli_format", "rich")
|
|
output, failed = _capture_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=path,
|
|
branch=branch,
|
|
description=description,
|
|
image=image,
|
|
container_id=container_id,
|
|
mount=mount,
|
|
url=url,
|
|
clone_into=clone_into,
|
|
read_only=read_only,
|
|
update=update,
|
|
fmt=resolved_fmt,
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
if fmt is None and hasattr(context, "resource_cli_format"):
|
|
delattr(context, "resource_cli_format")
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when('I run resource add "{type_name}" "{name}" with path "{path}"')
|
|
def step_run_resource_add(
|
|
context: Context, type_name: str, name: str, path: str
|
|
) -> None:
|
|
"""Run resource add command."""
|
|
_do_resource_add(context, type_name, name, path)
|
|
|
|
|
|
@given('I run resource add "{type_name}" "{name}" with path "{path}"')
|
|
def step_given_resource_add(
|
|
context: Context, type_name: str, name: str, path: str
|
|
) -> None:
|
|
"""Run resource add in Given context."""
|
|
_do_resource_add(context, type_name, name, path)
|
|
|
|
|
|
@when('I run resource add "{type_name}" "{name}" with url "{url_value}"')
|
|
def step_run_resource_add_url(
|
|
context: Context, type_name: str, name: str, url_value: str
|
|
) -> None:
|
|
"""Run resource add for a resource using only --url."""
|
|
_do_resource_add(context, type_name, name, None, url=url_value)
|
|
|
|
|
|
@when(
|
|
'I run resource add type "{type_name}" name "{name}" path "{path}" url "{url_value}"'
|
|
)
|
|
def step_run_resource_add_path_url(
|
|
context: Context, type_name: str, name: str, path: str, url_value: str
|
|
) -> None:
|
|
"""Run resource add providing both path and url flags."""
|
|
_do_resource_add(context, type_name, name, path, url=url_value)
|
|
|
|
|
|
@given('the resource CLI output format is "{fmt}"')
|
|
def step_set_resource_cli_format(context: Context, fmt: str) -> None:
|
|
"""Set the desired output format for the next resource CLI command."""
|
|
context.resource_cli_format = fmt
|
|
|
|
|
|
@given("auto discovery will fail during resource registration")
|
|
def step_auto_discovery_failure(context: Context) -> None:
|
|
"""Force ResourceRepository.auto_discover_children to raise during the next call."""
|
|
|
|
patcher = patch(
|
|
"cleveragents.infrastructure.database.repositories.ResourceRepository.auto_discover_children",
|
|
side_effect=RuntimeError("Simulated auto-discovery failure"),
|
|
)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
|
|
# ---- Resource List ----
|
|
|
|
|
|
@when("I run resource list")
|
|
def step_run_resource_list(context: Context) -> None:
|
|
"""Run resource list command."""
|
|
from cleveragents.cli.commands.resource import resource_list
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(resource_list, type_filter=None, fmt="rich")
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when('I run resource list with type filter "{type_name}"')
|
|
def step_run_resource_list_type(context: Context, type_name: str) -> None:
|
|
"""Run resource list with type filter."""
|
|
from cleveragents.cli.commands.resource import resource_list
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
resource_list, type_filter=type_name, fmt="rich"
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when('I run resource list with format "{fmt}"')
|
|
def step_run_resource_list_fmt(context: Context, fmt: str) -> None:
|
|
"""Run resource list with specific format."""
|
|
from cleveragents.cli.commands.resource import resource_list
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(resource_list, type_filter=None, fmt=fmt)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
# ---- Resource Show ----
|
|
|
|
|
|
@when('I run resource show "{name}" using default format')
|
|
def step_run_resource_show(context: Context, name: str) -> None:
|
|
"""Run resource show command."""
|
|
from cleveragents.cli.commands.resource import resource_show
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(resource_show, resource=name, fmt="rich")
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when('I run resource show "{name}" using format "{fmt}"')
|
|
def step_run_resource_show_fmt(context: Context, name: str, fmt: str) -> None:
|
|
"""Run resource show with specific format."""
|
|
from cleveragents.cli.commands.resource import resource_show
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(resource_show, resource=name, fmt=fmt)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
# ---- Resource Remove ----
|
|
|
|
|
|
@when('I run resource remove "{name}" with yes flag')
|
|
def step_run_resource_remove(context: Context, name: str) -> None:
|
|
"""Run resource remove with --yes flag."""
|
|
from cleveragents.cli.commands.resource import resource_remove
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(resource_remove, resource=name, yes=True)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
# ---- Assertions ----
|
|
|
|
|
|
@then('the resource output should contain "{text}"')
|
|
def step_resource_output_contains(context: Context, text: str) -> None:
|
|
"""Check that CLI output contains the expected text."""
|
|
output = context.resource_cli_output
|
|
assert text.lower() in output.lower(), (
|
|
f"Expected '{text}' in output, got:\n{output}"
|
|
)
|
|
|
|
|
|
@then("the resource command should succeed")
|
|
def step_resource_command_succeeded(context: Context) -> None:
|
|
"""Ensure the resource CLI command completed successfully."""
|
|
failed = getattr(context, "resource_cli_failed", False)
|
|
assert not failed, "Expected command to succeed but it failed"
|
|
|
|
|
|
@then("the resource command should fail")
|
|
def step_resource_command_failed(context: Context) -> None:
|
|
"""Check that the command failed."""
|
|
assert context.resource_cli_failed, "Expected command to fail but it succeeded"
|
|
|
|
|
|
@then(
|
|
'the resource registry resource "{resource_name}" should have property '
|
|
'"{property_name}" equal to "{expected_value}"'
|
|
)
|
|
def step_resource_property_equals(
|
|
context: Context, resource_name: str, property_name: str, expected_value: str
|
|
) -> None:
|
|
"""Assert that the stored resource has the expected property value."""
|
|
service = _make_service(context)
|
|
resource = service.show_resource(resource_name)
|
|
properties = resource.properties or {}
|
|
actual_value = properties.get(property_name)
|
|
assert actual_value == expected_value, (
|
|
f"Expected property '{property_name}' to be '{expected_value}', got '{actual_value}'"
|
|
)
|
|
|
|
|
|
@then("the resource output should be valid JSON")
|
|
def step_resource_output_valid_json(context: Context) -> None:
|
|
"""Check that the output is valid JSON."""
|
|
output = context.resource_cli_output.strip()
|
|
try:
|
|
json.loads(output)
|
|
except json.JSONDecodeError as exc:
|
|
raise AssertionError(f"Output is not valid JSON: {exc}\n{output}") from exc
|
|
|
|
|
|
@then('the resource JSON output should contain key "{key}"')
|
|
def step_resource_json_has_key(context: Context, key: str) -> None:
|
|
"""Check that the JSON output contains a specific key."""
|
|
output = context.resource_cli_output.strip()
|
|
data = json.loads(output)
|
|
if isinstance(data, list):
|
|
assert len(data) > 0, "JSON output is an empty list"
|
|
assert key in data[0], f"Key '{key}' not found in first item of JSON list"
|
|
else:
|
|
if key in data:
|
|
return
|
|
payload = data.get("data") if isinstance(data, dict) else None
|
|
if isinstance(payload, dict) and key in payload:
|
|
return
|
|
raise AssertionError(f"Key '{key}' not found in JSON output")
|
|
|
|
|
|
@then('the resource registry should not contain resource "{name}"')
|
|
def step_registry_missing_resource(context: Context, name: str) -> None:
|
|
"""Ensure that a resource was not persisted in the registry."""
|
|
|
|
service = _make_service(context)
|
|
try:
|
|
service.show_resource(name)
|
|
except NotFoundError:
|
|
return
|
|
raise AssertionError(f"Resource '{name}' was persisted despite failure")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Additional steps for resource_cli_coverage.feature
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I run resource type add with the config file using update flag")
|
|
def step_run_type_add_update(context: Context) -> None:
|
|
"""Run resource type add with --update flag."""
|
|
from cleveragents.cli.commands.resource import type_add
|
|
|
|
config_path = Path(context.resource_cli_yaml_path)
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
type_add, config=config_path, update=True, fmt="rich"
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@given("a custom resource type YAML file with invalid schema")
|
|
def step_create_invalid_yaml(context: Context) -> None:
|
|
"""Create a temp YAML file with invalid schema (missing required fields)."""
|
|
yaml_content = """
|
|
name: 123
|
|
description: Missing resource_kind and other required fields
|
|
"""
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp:
|
|
tmp.write(yaml_content)
|
|
tmp.flush()
|
|
context.resource_cli_yaml_path = tmp.name
|
|
|
|
|
|
@when('I run resource type add with the config file and format "{fmt}"')
|
|
def step_run_type_add_fmt(context: Context, fmt: str) -> None:
|
|
"""Run resource type add with a specific output format."""
|
|
from cleveragents.cli.commands.resource import type_add
|
|
|
|
config_path = Path(context.resource_cli_yaml_path)
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
type_add, config=config_path, update=False, fmt=fmt
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when('I run resource type remove "{name}" without yes flag')
|
|
def step_run_type_remove_no_yes(context: Context, name: str) -> None:
|
|
"""Run resource type remove without --yes (confirmation will be denied)."""
|
|
from unittest.mock import patch
|
|
|
|
from cleveragents.cli.commands.resource import type_remove
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
# Mock typer.confirm to return False (user declines)
|
|
with patch(
|
|
"cleveragents.cli.commands.resource.typer.confirm", return_value=False
|
|
):
|
|
output, failed = _capture_output(type_remove, name=name, yes=False)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when(
|
|
'I run resource add formatted as "{fmt}" type "{type_name}"'
|
|
' name "{name}" path "{path}"'
|
|
)
|
|
def step_run_resource_add_fmt(
|
|
context: Context, fmt: str, type_name: str, name: str, path: str
|
|
) -> None:
|
|
"""Run resource add with a specific output format."""
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=path,
|
|
branch=None,
|
|
description=None,
|
|
read_only=False,
|
|
clone_into=None,
|
|
update=False,
|
|
fmt=fmt,
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when(
|
|
'I run resource add with extras type "{type_name}" name "{name}"'
|
|
' path "{path}" branch "{branch}" description "{desc}"'
|
|
)
|
|
def step_run_resource_add_branch_desc(
|
|
context: Context,
|
|
type_name: str,
|
|
name: str,
|
|
path: str,
|
|
branch: str,
|
|
desc: str,
|
|
) -> None:
|
|
"""Run resource add with branch and description flags."""
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=path,
|
|
branch=branch if branch else None,
|
|
description=desc,
|
|
read_only=False,
|
|
clone_into=None,
|
|
update=False,
|
|
fmt="rich",
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@given('I add a resource referencing custom type "{type_name}"')
|
|
def step_add_resource_referencing_type(context: Context, type_name: str) -> None:
|
|
"""Add a resource instance that references the given type."""
|
|
service = _make_service(context)
|
|
service.register_resource(
|
|
type_name=type_name,
|
|
name="local/ref-test",
|
|
location="/tmp/ref",
|
|
description="Test resource for type removal test",
|
|
read_only=False,
|
|
properties={"path": "/tmp/ref"},
|
|
)
|
|
|
|
|
|
@given("I add a resource with a long description")
|
|
def step_add_resource_long_desc(context: Context) -> None:
|
|
"""Add a resource with a description exceeding the 40-char truncation limit."""
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
long_desc = (
|
|
"This is a very long description that exceeds "
|
|
"the forty character limit for display"
|
|
)
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
resource_add,
|
|
type_name="git-checkout",
|
|
name="local/long-desc",
|
|
path="/tmp/ld",
|
|
branch=None,
|
|
description=long_desc,
|
|
read_only=False,
|
|
clone_into=None,
|
|
update=False,
|
|
fmt="rich",
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when('I run resource remove "{name}" without yes flag')
|
|
def step_run_resource_remove_no_yes(context: Context, name: str) -> None:
|
|
"""Run resource remove without --yes (confirmation will be denied)."""
|
|
from cleveragents.cli.commands.resource import resource_remove
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
with patch(
|
|
"cleveragents.cli.commands.resource.typer.confirm", return_value=False
|
|
):
|
|
output, failed = _capture_output(resource_remove, resource=name, yes=False)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Error-mocking steps for CleverAgentsError / ValidationError catch-all paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_error_service(
|
|
method_name: str, error_cls: type = CleverAgentsError
|
|
) -> MagicMock:
|
|
"""Create a mock service where one method raises an error."""
|
|
svc = MagicMock(spec=ResourceRegistryService)
|
|
err = error_cls(message="Simulated error")
|
|
getattr(svc, method_name).side_effect = err
|
|
return svc
|
|
|
|
|
|
def _patch_error_service(
|
|
context: Context, method_name: str, error_cls: type = CleverAgentsError
|
|
) -> None:
|
|
"""Store a mock service on the context that errors on a specific method."""
|
|
context.resource_cli_service = _make_error_service(method_name, error_cls)
|
|
|
|
|
|
@given("a service that raises CleverAgentsError on list_types")
|
|
def step_service_error_list_types(context: Context) -> None:
|
|
_patch_error_service(context, "list_types")
|
|
|
|
|
|
@given("a service that raises CleverAgentsError on show_type")
|
|
def step_service_error_show_type(context: Context) -> None:
|
|
_patch_error_service(context, "show_type")
|
|
|
|
|
|
@given("a service that raises CleverAgentsError on register_type")
|
|
def step_service_error_register_type(context: Context) -> None:
|
|
_patch_error_service(context, "register_type")
|
|
|
|
|
|
@given("a service that raises CleverAgentsError on register_resource")
|
|
def step_service_error_register_resource(context: Context) -> None:
|
|
_patch_error_service(context, "register_resource")
|
|
|
|
|
|
@given("a service that raises ValidationError on register_resource")
|
|
def step_service_validation_register_resource(context: Context) -> None:
|
|
_patch_error_service(context, "register_resource", ValidationError)
|
|
|
|
|
|
@given("a service that raises CleverAgentsError on list_resources")
|
|
def step_service_error_list_resources(context: Context) -> None:
|
|
_patch_error_service(context, "list_resources")
|
|
|
|
|
|
@given("a service that raises CleverAgentsError on show_resource")
|
|
def step_service_error_show_resource(context: Context) -> None:
|
|
_patch_error_service(context, "show_resource")
|
|
|
|
|
|
@given("a type without cli_args is registered")
|
|
def step_type_without_cli_args(context: Context) -> None:
|
|
"""Register a custom type that has no cli_args."""
|
|
service = _make_service(context)
|
|
yaml_content = """
|
|
name: test/no-args
|
|
description: A type with no CLI arguments
|
|
resource_kind: physical
|
|
sandbox_strategy: copy_on_write
|
|
user_addable: true
|
|
cli_args: []
|
|
"""
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp:
|
|
tmp.write(yaml_content)
|
|
tmp.flush()
|
|
service.register_type(Path(tmp.name))
|
|
|
|
|
|
@given("I add a resource with empty properties")
|
|
def step_add_resource_empty_props(context: Context) -> None:
|
|
"""Add a resource with no properties at all."""
|
|
service = _make_service(context)
|
|
service.register_resource(
|
|
type_name="git-checkout",
|
|
name="local/empty-props",
|
|
location=None,
|
|
description="No properties resource",
|
|
read_only=False,
|
|
properties=None,
|
|
)
|
|
|
|
|
|
@when('I run resource add "{type_name}" "{name}" with clone-into "{clone_into_val}"')
|
|
def step_run_resource_add_clone_into(
|
|
context: Context, type_name: str, name: str, clone_into_val: str
|
|
) -> None:
|
|
"""Run resource add with --clone-into flag."""
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=None,
|
|
branch=None,
|
|
description=None,
|
|
image=None,
|
|
clone_into=clone_into_val,
|
|
read_only=False,
|
|
fmt="rich",
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|
|
|
|
|
|
@when("I run resource type add with path to missing file")
|
|
def step_run_type_add_missing_file(context: Context) -> None:
|
|
"""Run type_add directly with a path that triggers FileNotFoundError."""
|
|
from cleveragents.cli.commands.resource import type_add
|
|
|
|
orig = _patch_service(context)
|
|
try:
|
|
output, failed = _capture_output(
|
|
type_add,
|
|
config=Path("/tmp/absolutely_does_not_exist_xyz.yaml"),
|
|
update=False,
|
|
fmt="rich",
|
|
)
|
|
context.resource_cli_output = output
|
|
context.resource_cli_failed = failed
|
|
finally:
|
|
_unpatch_service(orig)
|