forked from HAL9000/cleveragents-core
freemo
b9707ae165
Add tree, inspect, link-child, and unlink-child subcommands to the agents resource CLI group. Extend ResourceRegistryService with link_child, unlink_child, get_children, and get_resource_tree methods. - tree: display resource hierarchy with --depth, --type, and format options - inspect: show resource details with --tree and --file options - link-child/unlink-child: manage DAG edges with cycle detection - 24 Behave scenarios, 4 Robot integration tests, ASV benchmarks - CLI reference documentation in docs/reference/resource_cli.md
439 lines
14 KiB
Python
439 lines
14 KiB
Python
"""Step definitions for resource_cli_tree.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from rich.console import Console
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from cleveragents.application.services.resource_registry_service import (
|
|
ResourceRegistryService,
|
|
)
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
|
|
def _make_tree_service(context: Context) -> ResourceRegistryService:
|
|
"""Create or return a cached ResourceRegistryService for tree tests."""
|
|
if not hasattr(context, "tree_cli_service"):
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def _fk(conn: Any, _rec: Any) -> None:
|
|
conn.cursor().execute("PRAGMA foreign_keys=ON")
|
|
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
context.tree_cli_service = ResourceRegistryService(session_factory=factory)
|
|
return context.tree_cli_service
|
|
|
|
|
|
def _capture_tree_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, bool]:
|
|
"""Run a CLI function capturing its console output and success status."""
|
|
buf = StringIO()
|
|
console = Console(file=buf, width=200, no_color=True)
|
|
|
|
import cleveragents.cli.commands.resource as resource_mod
|
|
|
|
orig_console = resource_mod.console
|
|
resource_mod.console = console
|
|
|
|
failed = False
|
|
try:
|
|
func(*args, **kwargs)
|
|
except SystemExit:
|
|
failed = True
|
|
except Exception:
|
|
failed = True
|
|
finally:
|
|
resource_mod.console = orig_console
|
|
|
|
return buf.getvalue(), failed
|
|
|
|
|
|
def _patch_tree_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_tree_service(context)
|
|
|
|
resource_mod._get_registry_service = _mock_get
|
|
return orig_fn
|
|
|
|
|
|
def _unpatch_tree_service(orig_fn: Any) -> None:
|
|
"""Restore original service getter."""
|
|
import cleveragents.cli.commands.resource as resource_mod
|
|
|
|
resource_mod._get_registry_service = orig_fn
|
|
|
|
|
|
# ---- Background ----
|
|
|
|
|
|
@given("a fresh resource tree test registry")
|
|
def step_fresh_tree_registry(context: Context) -> None:
|
|
"""Set up a fresh in-memory database for tree tests."""
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def _fk(conn: Any, _rec: Any) -> None:
|
|
conn.cursor().execute("PRAGMA foreign_keys=ON")
|
|
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
context.tree_cli_service = ResourceRegistryService(session_factory=factory)
|
|
context.tree_cli_output = ""
|
|
context.tree_cli_failed = False
|
|
context.tree_temp_dir = None
|
|
|
|
|
|
@given("resource tree built-in types are bootstrapped")
|
|
def step_tree_bootstrap(context: Context) -> None:
|
|
"""Bootstrap built-in resource types for tree tests."""
|
|
service = _make_tree_service(context)
|
|
service.bootstrap_builtin_types()
|
|
|
|
|
|
@given('a resource tree resource "{type_name}" named "{name}" at "{path}"')
|
|
def step_tree_add_resource(
|
|
context: Context, type_name: str, name: str, path: str
|
|
) -> None:
|
|
"""Add a resource for tree tests."""
|
|
service = _make_tree_service(context)
|
|
service.register_resource(
|
|
type_name=type_name,
|
|
name=name,
|
|
location=path,
|
|
description=f"Test resource {name}",
|
|
read_only=False,
|
|
properties={"path": path},
|
|
)
|
|
|
|
|
|
@given('a resource tree resource "{type_name}" named "{name}" at a temp directory')
|
|
def step_tree_add_resource_temp(context: Context, type_name: str, name: str) -> None:
|
|
"""Add a resource at a temp directory."""
|
|
tmp_dir = tempfile.mkdtemp()
|
|
context.tree_temp_dir = tmp_dir
|
|
service = _make_tree_service(context)
|
|
service.register_resource(
|
|
type_name=type_name,
|
|
name=name,
|
|
location=tmp_dir,
|
|
description=f"Test resource {name}",
|
|
read_only=False,
|
|
properties={"path": tmp_dir},
|
|
)
|
|
|
|
|
|
@given('a temp file "{filename}" with content "{content}"')
|
|
def step_tree_create_temp_file(context: Context, filename: str, content: str) -> None:
|
|
"""Create a temp file in the temp directory."""
|
|
tmp_dir = context.tree_temp_dir
|
|
filepath = Path(tmp_dir) / filename
|
|
filepath.write_text(content, encoding="utf-8")
|
|
|
|
|
|
@given('resource tree child "{child}" is linked to parent "{parent}"')
|
|
def step_tree_link_child(context: Context, child: str, parent: str) -> None:
|
|
"""Link a child to a parent."""
|
|
service = _make_tree_service(context)
|
|
service.link_child(parent, child)
|
|
|
|
|
|
# ---- Tree commands ----
|
|
|
|
|
|
@when('I display resource tree for "{name}"')
|
|
def step_run_tree(context: Context, name: str) -> None:
|
|
"""Run resource tree command."""
|
|
from cleveragents.cli.commands.resource import resource_tree
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_tree, resource=name, depth=-1, type_filter=None, fmt="rich"
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I display resource tree depth-limited to {depth:d} for "{name}"')
|
|
def step_run_tree_depth(context: Context, depth: int, name: str) -> None:
|
|
"""Run resource tree with depth limit."""
|
|
from cleveragents.cli.commands.resource import resource_tree
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_tree, resource=name, depth=depth, type_filter=None, fmt="rich"
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I display resource tree type-filtered to "{type_filter}" for "{name}"')
|
|
def step_run_tree_type(context: Context, type_filter: str, name: str) -> None:
|
|
"""Run resource tree with type filter."""
|
|
from cleveragents.cli.commands.resource import resource_tree
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_tree, resource=name, depth=-1, type_filter=type_filter, fmt="rich"
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I display resource tree as "{fmt}" for "{name}"')
|
|
def step_run_tree_fmt(context: Context, fmt: str, name: str) -> None:
|
|
"""Run resource tree with format."""
|
|
from cleveragents.cli.commands.resource import resource_tree
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_tree, resource=name, depth=-1, type_filter=None, fmt=fmt
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
# ---- Inspect commands ----
|
|
|
|
|
|
@when('I run basic inspect on "{name}"')
|
|
def step_run_inspect(context: Context, name: str) -> None:
|
|
"""Run resource inspect command."""
|
|
from cleveragents.cli.commands.resource import resource_inspect
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_inspect, resource=name, tree=False, file=None, fmt="rich"
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I run inspect with subtree on "{name}"')
|
|
def step_run_inspect_tree(context: Context, name: str) -> None:
|
|
"""Run resource inspect with --tree."""
|
|
from cleveragents.cli.commands.resource import resource_inspect
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_inspect, resource=name, tree=True, file=None, fmt="rich"
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I run inspect with file "{filepath}" on "{name}"')
|
|
def step_run_inspect_file(context: Context, filepath: str, name: str) -> None:
|
|
"""Run resource inspect with --file."""
|
|
from cleveragents.cli.commands.resource import resource_inspect
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_inspect, resource=name, tree=False, file=filepath, fmt="rich"
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I run inspect formatted as "{fmt}" on "{name}"')
|
|
def step_run_inspect_fmt(context: Context, fmt: str, name: str) -> None:
|
|
"""Run resource inspect with format."""
|
|
from cleveragents.cli.commands.resource import resource_inspect
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_inspect, resource=name, tree=False, file=None, fmt=fmt
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
# ---- Link-child commands ----
|
|
|
|
|
|
@when('I link DAG child "{child}" to parent "{parent}"')
|
|
def step_run_link_child(context: Context, child: str, parent: str) -> None:
|
|
"""Run resource link-child command."""
|
|
from cleveragents.cli.commands.resource import resource_link_child
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_link_child, parent=parent, child=child, fmt="rich"
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I link as "{fmt}" DAG child "{child}" to parent "{parent}"')
|
|
def step_run_link_child_fmt(
|
|
context: Context, fmt: str, child: str, parent: str
|
|
) -> None:
|
|
"""Run resource link-child with format."""
|
|
from cleveragents.cli.commands.resource import resource_link_child
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_link_child, parent=parent, child=child, fmt=fmt
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
# ---- Unlink-child commands ----
|
|
|
|
|
|
@when('I unlink DAG child "{child}" from parent "{parent}" confirmed')
|
|
def step_run_unlink_child_yes(context: Context, child: str, parent: str) -> None:
|
|
"""Run resource unlink-child with --yes."""
|
|
from cleveragents.cli.commands.resource import resource_unlink_child
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_unlink_child, parent=parent, child=child, yes=True, fmt="rich"
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I unlink as "{fmt}" DAG child "{child}" from parent "{parent}" confirmed')
|
|
def step_run_unlink_child_yes_fmt(
|
|
context: Context, fmt: str, child: str, parent: str
|
|
) -> None:
|
|
"""Run resource unlink-child with --yes and format."""
|
|
from cleveragents.cli.commands.resource import resource_unlink_child
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
output, failed = _capture_tree_output(
|
|
resource_unlink_child, parent=parent, child=child, yes=True, fmt=fmt
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
@when('I unlink DAG child "{child}" from parent "{parent}" declined')
|
|
def step_run_unlink_child_no_yes(context: Context, child: str, parent: str) -> None:
|
|
"""Run resource unlink-child without --yes (confirmation denied)."""
|
|
from cleveragents.cli.commands.resource import resource_unlink_child
|
|
|
|
orig = _patch_tree_service(context)
|
|
try:
|
|
with patch(
|
|
"cleveragents.cli.commands.resource.typer.confirm",
|
|
return_value=False,
|
|
):
|
|
output, failed = _capture_tree_output(
|
|
resource_unlink_child,
|
|
parent=parent,
|
|
child=child,
|
|
yes=False,
|
|
fmt="rich",
|
|
)
|
|
context.tree_cli_output = output
|
|
context.tree_cli_failed = failed
|
|
finally:
|
|
_unpatch_tree_service(orig)
|
|
|
|
|
|
# ---- Assertions ----
|
|
|
|
|
|
@then('the tree output should contain "{text}"')
|
|
def step_tree_output_contains(context: Context, text: str) -> None:
|
|
"""Check that CLI output contains the expected text."""
|
|
output = context.tree_cli_output
|
|
assert text.lower() in output.lower(), (
|
|
f"Expected '{text}' in output, got:\n{output}"
|
|
)
|
|
|
|
|
|
@then('the tree output should not contain "{text}"')
|
|
def step_tree_output_not_contains(context: Context, text: str) -> None:
|
|
"""Check that CLI output does not contain the text."""
|
|
output = context.tree_cli_output
|
|
assert text.lower() not in output.lower(), (
|
|
f"Did not expect '{text}' in output, got:\n{output}"
|
|
)
|
|
|
|
|
|
@then("the tree command should fail")
|
|
def step_tree_command_failed(context: Context) -> None:
|
|
"""Check that the command failed."""
|
|
assert context.tree_cli_failed, "Expected command to fail but it succeeded"
|
|
|
|
|
|
@then("the tree output should be valid JSON")
|
|
def step_tree_output_valid_json(context: Context) -> None:
|
|
"""Check that the output is valid JSON."""
|
|
output = context.tree_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 tree output should show "{first}" before "{second}"')
|
|
def step_tree_output_order(context: Context, first: str, second: str) -> None:
|
|
"""Check that first appears before second in output."""
|
|
output = context.tree_cli_output.lower()
|
|
first_pos = output.find(first.lower())
|
|
second_pos = output.find(second.lower())
|
|
assert first_pos >= 0, f"'{first}' not found in output:\n{context.tree_cli_output}"
|
|
assert second_pos >= 0, (
|
|
f"'{second}' not found in output:\n{context.tree_cli_output}"
|
|
)
|
|
assert first_pos < second_pos, (
|
|
f"'{first}' (pos {first_pos}) should appear before "
|
|
f"'{second}' (pos {second_pos}) in output:\n{context.tree_cli_output}"
|
|
)
|