forked from HAL9000/cleveragents-core
freemo
9a3c4265a0
Adds missing CLI flags to `resource add`, `resource list`, `resource tree`, `resource type list`, and `lsp list` per specification. Changes: - Added --update, --clone-into to resource_add; expanded --mount for devcontainer-instance - Added --all to resource_list; changed --depth default to 3 - Added [REGEX] positional to type_list and lsp list - Added include_auto_discovered parameter to resource registry ops - 12 new BDD scenarios with step definitions - Fixed critical bug: clone-into URL parsing used split() instead of rsplit(), causing HTTPS URLs to be stored with incorrect data Closes #904 Co-authored-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me> Co-committed-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
551 lines
17 KiB
Python
551 lines
17 KiB
Python
"""Step definitions for ``resource_cli_flags_904.feature`` (issue #904).
|
|
|
|
Tests the new CLI flags:
|
|
- resource add --update, --mount (container types), --clone-into
|
|
- resource list --all
|
|
- resource tree --depth default changed to 3
|
|
- resource type list [REGEX]
|
|
- lsp list [REGEX]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from io import StringIO
|
|
from typing import Any
|
|
|
|
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.cli.commands.lsp import _reset_registry
|
|
from cleveragents.infrastructure.database.models import Base, ResourceModel
|
|
from cleveragents.lsp.models import LspCapability, LspServerConfig
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_flags_service(context: Context) -> ResourceRegistryService:
|
|
"""Create or return a cached ResourceRegistryService for flags tests."""
|
|
if not hasattr(context, "flags_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.flags_cli_service = ResourceRegistryService(session_factory=factory)
|
|
return context.flags_cli_service
|
|
|
|
|
|
def _capture_flags_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()
|
|
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
|
|
|
|
|
|
def _patch_flags_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_flags_service(context)
|
|
|
|
resource_mod._get_registry_service = _mock_get
|
|
return orig_fn
|
|
|
|
|
|
def _unpatch_flags_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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh resource flags test registry")
|
|
def step_fresh_flags_registry(context: Context) -> None:
|
|
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.flags_cli_service = ResourceRegistryService(session_factory=factory)
|
|
context.flags_cli_output = ""
|
|
context.flags_cli_failed = False
|
|
context.flags_tree_depth_used = None
|
|
|
|
|
|
@given("resource flags built-in types are bootstrapped")
|
|
def step_flags_bootstrap(context: Context) -> None:
|
|
service = _make_flags_service(context)
|
|
service.bootstrap_builtin_types()
|
|
|
|
|
|
@given(
|
|
'I add a resource flags resource "{type_name}" named "{name}" with path "{path}"'
|
|
)
|
|
def step_flags_add_resource_given(
|
|
context: Context, type_name: str, name: str, path: str
|
|
) -> None:
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=path,
|
|
branch=None,
|
|
description=None,
|
|
read_only=False,
|
|
clone_into=None,
|
|
update=False,
|
|
fmt="rich",
|
|
)
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
@given('a resource flags resource "{type_name}" named "{name}" with path "{path}"')
|
|
def step_flags_add_resource_given_alt(
|
|
context: Context, type_name: str, name: str, path: str
|
|
) -> None:
|
|
service = _make_flags_service(context)
|
|
service.register_resource(
|
|
type_name=type_name,
|
|
name=name,
|
|
location=path,
|
|
description=None,
|
|
read_only=False,
|
|
properties={"path": path},
|
|
)
|
|
|
|
|
|
@given(
|
|
'a user-added resource flags resource "{type_name}" named "{name}" with path "{path}"'
|
|
)
|
|
def step_flags_add_user_resource(
|
|
context: Context, type_name: str, name: str, path: str
|
|
) -> None:
|
|
service = _make_flags_service(context)
|
|
service.register_resource(
|
|
type_name=type_name,
|
|
name=name,
|
|
location=path,
|
|
description=None,
|
|
read_only=False,
|
|
properties={"path": path},
|
|
)
|
|
|
|
|
|
@given('an auto-discovered resource flags resource named "{name}"')
|
|
def step_flags_add_auto_discovered(context: Context, name: str) -> None:
|
|
"""Directly insert an auto-discovered resource into the DB."""
|
|
from ulid import ULID
|
|
|
|
service = _make_flags_service(context)
|
|
session = service._session()
|
|
try:
|
|
now_iso = datetime.now(tz=UTC).isoformat()
|
|
db_resource = ResourceModel(
|
|
resource_id=str(ULID()),
|
|
namespaced_name=name,
|
|
namespace=name.split("/", 1)[0] if "/" in name else None,
|
|
type_name="git-checkout",
|
|
resource_kind="physical",
|
|
location="/tmp/auto",
|
|
description="Auto-discovered",
|
|
read_only=False,
|
|
auto_discovered=True,
|
|
sandbox_strategy=None,
|
|
content_hash=None,
|
|
properties_json=None,
|
|
metadata_json=None,
|
|
created_at=now_iso,
|
|
updated_at=now_iso,
|
|
)
|
|
session.add(db_resource)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LSP Given
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a clean LSP flags test environment")
|
|
def step_clean_lsp_flags(context: Context) -> None:
|
|
_reset_registry()
|
|
context.lsp_flags_output = ""
|
|
context.lsp_flags_failed = False
|
|
|
|
|
|
@given('a registered LSP flags server "{name}" for language "{lang}"')
|
|
def step_register_lsp_flags_server(context: Context, name: str, lang: str) -> None:
|
|
from cleveragents.cli.commands.lsp import _get_registry
|
|
|
|
registry = _get_registry()
|
|
config = LspServerConfig(
|
|
name=name,
|
|
command="test-server",
|
|
args=["--stdio"],
|
|
languages=[lang],
|
|
capabilities=[LspCapability.DIAGNOSTICS],
|
|
)
|
|
registry.register(config)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — resource add
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I run resource flags add "{type_name}" "{name}" with path "{path}" and update')
|
|
def step_flags_add_update(
|
|
context: Context, type_name: str, name: str, path: str
|
|
) -> None:
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=path,
|
|
branch=None,
|
|
description=None,
|
|
read_only=False,
|
|
clone_into=None,
|
|
update=True,
|
|
fmt="rich",
|
|
)
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
@when(
|
|
'I run resource flags add "{type_name}" "{name}" with path "{path}" and mount "{mount_spec}"'
|
|
)
|
|
def step_flags_add_mount(
|
|
context: Context, type_name: str, name: str, path: str, mount_spec: str
|
|
) -> None:
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=path,
|
|
branch=None,
|
|
description=None,
|
|
read_only=False,
|
|
mount=[mount_spec],
|
|
clone_into=None,
|
|
update=False,
|
|
fmt="rich",
|
|
)
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
@when(
|
|
'I run resource flags add "{type_name}" "{name}" with path "{path}" and clone-into "{clone_spec}"'
|
|
)
|
|
def step_flags_add_clone_into(
|
|
context: Context, type_name: str, name: str, path: str, clone_spec: str
|
|
) -> None:
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=path,
|
|
branch=None,
|
|
description=None,
|
|
read_only=False,
|
|
clone_into=clone_spec,
|
|
update=False,
|
|
fmt="rich",
|
|
)
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
@when('I run resource flags add "{type_name}" "{name}" with clone-into "{clone_spec}"')
|
|
def step_flags_add_clone_into_no_path(
|
|
context: Context, type_name: str, name: str, clone_spec: str
|
|
) -> None:
|
|
from cleveragents.cli.commands.resource import resource_add
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(
|
|
resource_add,
|
|
type_name=type_name,
|
|
name=name,
|
|
path=None,
|
|
branch=None,
|
|
description=None,
|
|
read_only=False,
|
|
clone_into=clone_spec,
|
|
update=False,
|
|
fmt="rich",
|
|
)
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — resource list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I run resource flags list")
|
|
def step_flags_list(context: Context) -> None:
|
|
from cleveragents.cli.commands.resource import resource_list
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(
|
|
resource_list, type_filter=None, show_all=False, fmt="rich"
|
|
)
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
@when("I run resource flags list with --all")
|
|
def step_flags_list_all(context: Context) -> None:
|
|
from cleveragents.cli.commands.resource import resource_list
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(
|
|
resource_list, type_filter=None, show_all=True, fmt="rich"
|
|
)
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — resource tree
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I run resource flags tree for "{name}" with default depth')
|
|
def step_flags_tree_default(context: Context, name: str) -> None:
|
|
"""Run resource tree and capture the depth that was actually used."""
|
|
from cleveragents.cli.commands.resource import resource_tree
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
# We call resource_tree *without* passing depth so the default is used.
|
|
# We capture what depth the function uses by inspecting its default.
|
|
import inspect
|
|
|
|
sig = inspect.signature(resource_tree)
|
|
default_depth = sig.parameters["depth"].default
|
|
context.flags_tree_depth_used = default_depth
|
|
|
|
output, failed = _capture_flags_output(
|
|
resource_tree, resource=name, type_filter=None, fmt="rich"
|
|
)
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — resource type list [REGEX]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I run resource flags type list with regex "{regex}"')
|
|
def step_flags_type_list_regex(context: Context, regex: str) -> None:
|
|
from cleveragents.cli.commands.resource import type_list
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(type_list, regex=regex, fmt="rich")
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
@when("I run resource flags type list without regex")
|
|
def step_flags_type_list_no_regex(context: Context) -> None:
|
|
from cleveragents.cli.commands.resource import type_list
|
|
|
|
orig = _patch_flags_service(context)
|
|
try:
|
|
output, failed = _capture_flags_output(type_list, regex=None, fmt="rich")
|
|
context.flags_cli_output = output
|
|
context.flags_cli_failed = failed
|
|
finally:
|
|
_unpatch_flags_service(orig)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — lsp list [REGEX]
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _capture_lsp_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, bool]:
|
|
"""Run an LSP CLI function capturing its console output."""
|
|
import contextlib
|
|
|
|
buf = StringIO()
|
|
console = Console(
|
|
file=buf, width=200, no_color=True, highlight=False, force_terminal=False
|
|
)
|
|
|
|
import cleveragents.cli.commands.lsp as lsp_mod
|
|
|
|
orig_console = lsp_mod.console
|
|
lsp_mod.console = console
|
|
|
|
failed = False
|
|
try:
|
|
with contextlib.redirect_stdout(buf):
|
|
func(*args, **kwargs)
|
|
except SystemExit:
|
|
failed = True
|
|
except Exception:
|
|
failed = True
|
|
finally:
|
|
lsp_mod.console = orig_console
|
|
|
|
return buf.getvalue(), failed
|
|
|
|
|
|
@when('I run lsp flags list with regex "{regex}"')
|
|
def step_lsp_flags_list_regex(context: Context, regex: str) -> None:
|
|
from cleveragents.cli.commands.lsp import list_servers
|
|
|
|
output, failed = _capture_lsp_output(list_servers, regex=regex, fmt="rich")
|
|
context.lsp_flags_output = output
|
|
context.lsp_flags_failed = failed
|
|
|
|
|
|
@when("I run lsp flags list without regex")
|
|
def step_lsp_flags_list_no_regex(context: Context) -> None:
|
|
from cleveragents.cli.commands.lsp import list_servers
|
|
|
|
output, failed = _capture_lsp_output(list_servers, regex=None, fmt="rich")
|
|
context.lsp_flags_output = output
|
|
context.lsp_flags_failed = failed
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the resource flags output should contain "{text}"')
|
|
def step_flags_output_contains(context: Context, text: str) -> None:
|
|
output = context.flags_cli_output
|
|
assert text.lower() in output.lower(), (
|
|
f"Expected '{text}' in output, got:\n{output}"
|
|
)
|
|
|
|
|
|
@then('the resource flags output should not contain "{text}"')
|
|
def step_flags_output_not_contains(context: Context, text: str) -> None:
|
|
output = context.flags_cli_output
|
|
assert text.lower() not in output.lower(), (
|
|
f"Did not expect '{text}' in output, got:\n{output}"
|
|
)
|
|
|
|
|
|
@then("the resource flags command should fail")
|
|
def step_flags_command_failed(context: Context) -> None:
|
|
assert context.flags_cli_failed, "Expected command to fail but it succeeded"
|
|
|
|
|
|
@then("the resource flags tree depth used should be {expected_depth:d}")
|
|
def step_flags_tree_depth(context: Context, expected_depth: int) -> None:
|
|
assert context.flags_tree_depth_used == expected_depth, (
|
|
f"Expected default depth {expected_depth}, got {context.flags_tree_depth_used}"
|
|
)
|
|
|
|
|
|
@then('the lsp flags output should contain "{text}"')
|
|
def step_lsp_flags_output_contains(context: Context, text: str) -> None:
|
|
output = context.lsp_flags_output
|
|
assert text.lower() in output.lower(), (
|
|
f"Expected '{text}' in output, got:\n{output}"
|
|
)
|
|
|
|
|
|
@then('the lsp flags output should not contain "{text}"')
|
|
def step_lsp_flags_output_not_contains(context: Context, text: str) -> None:
|
|
output = context.lsp_flags_output
|
|
assert text.lower() not in output.lower(), (
|
|
f"Did not expect '{text}' in output, got:\n{output}"
|
|
)
|