mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-07-30 19:02:16 +00:00
feat: add CLI commands: config, registry, tool (#98)
* feat: add CLI commands: config, registry, tool * fix: address review feedback on CLI config/registry/tool commands - Replace non-existent `create_config_template` with `generate_default_toml` to fix ImportError when config file is missing - Deduplicate registry maps into shared `_load_registry_map()` helper - Remove dead `_get_registry_class()` function and its tests - Route JSON output to stdout for pipeability (`jarvis config show json | jq`) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Jon Saad-Falcon <41205309+jonsaadfalcon@users.noreply.github.com> Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
geodab
co-authored by
Claude Opus 4.6
Jon Saad-Falcon
Claude Opus 4.6
Jon Saad-Falcon
parent
8468f08c41
commit
04c28ef7ed
@@ -12,6 +12,7 @@ from openjarvis.cli.bench_cmd import bench
|
||||
from openjarvis.cli.channel_cmd import channel
|
||||
from openjarvis.cli.chat_cmd import chat
|
||||
from openjarvis.cli.compose_cmd import compose
|
||||
from openjarvis.cli.config_cmd import config
|
||||
from openjarvis.cli.daemon_cmd import restart, start, status, stop
|
||||
from openjarvis.cli.doctor_cmd import doctor
|
||||
from openjarvis.cli.eval_cmd import eval_group
|
||||
@@ -24,10 +25,12 @@ from openjarvis.cli.model import model
|
||||
from openjarvis.cli.operators_cmd import operators
|
||||
from openjarvis.cli.optimize_cmd import optimize_group
|
||||
from openjarvis.cli.quickstart_cmd import quickstart
|
||||
from openjarvis.cli.registry_cmd import registry
|
||||
from openjarvis.cli.scheduler_cmd import scheduler
|
||||
from openjarvis.cli.serve import serve
|
||||
from openjarvis.cli.skill_cmd import skill
|
||||
from openjarvis.cli.telemetry_cmd import telemetry
|
||||
from openjarvis.cli.tool_cmd import tool
|
||||
from openjarvis.cli.vault_cmd import vault
|
||||
from openjarvis.cli.workflow_cmd import workflow
|
||||
|
||||
@@ -81,6 +84,9 @@ cli.add_command(optimize_group, "optimize")
|
||||
cli.add_command(feedback_group, "feedback")
|
||||
cli.add_command(compose, "compose")
|
||||
cli.add_command(gateway, "gateway")
|
||||
cli.add_command(tool, "tool")
|
||||
cli.add_command(registry, "registry")
|
||||
cli.add_command(config, "config")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
|
||||
@@ -0,0 +1,277 @@
|
||||
"""``jarvis config`` — configuration inspection commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.syntax import Syntax
|
||||
from rich.table import Table
|
||||
|
||||
|
||||
@click.group()
|
||||
def config() -> None:
|
||||
"""Inspect configuration — show loaded settings, hardware, and config files."""
|
||||
|
||||
|
||||
def _get_config_path(path: str | None) -> Path:
|
||||
"""Determine the config path from argument or environment."""
|
||||
from openjarvis.core.config import DEFAULT_CONFIG_PATH
|
||||
|
||||
if path:
|
||||
return Path(path)
|
||||
return Path(os.environ.get("OPENJARVIS_CONFIG", DEFAULT_CONFIG_PATH))
|
||||
|
||||
|
||||
def _show_hardware_info(console: Console, show_recommendations: bool = True) -> None:
|
||||
"""Display detected hardware information."""
|
||||
from openjarvis.core.config import detect_hardware, recommend_engine
|
||||
|
||||
hardware = detect_hardware()
|
||||
|
||||
console.print("\n[bold]Detected Hardware[/bold]")
|
||||
|
||||
hardware_table = Table(show_header=True, header_style="cyan")
|
||||
hardware_table.add_column("Component", style="green")
|
||||
hardware_table.add_column("Value", style="white")
|
||||
|
||||
# Platform
|
||||
hardware_table.add_row("Platform", str(hardware.platform))
|
||||
|
||||
# CPU
|
||||
if hardware.cpu_brand:
|
||||
hardware_table.add_row("CPU", hardware.cpu_brand)
|
||||
hardware_table.add_row("CPU Count", str(hardware.cpu_count))
|
||||
hardware_table.add_row("RAM", f"{hardware.ram_gb:.1f} GB")
|
||||
|
||||
# GPU
|
||||
if hardware.gpu:
|
||||
hardware_table.add_row("GPU Vendor", hardware.gpu.vendor)
|
||||
hardware_table.add_row("GPU Model", hardware.gpu.name)
|
||||
hardware_table.add_row("GPU VRAM", f"{hardware.gpu.vram_gb:.1f} GB")
|
||||
hardware_table.add_row("GPU Count", str(hardware.gpu.count))
|
||||
|
||||
console.print(hardware_table)
|
||||
|
||||
# Show recommended engine
|
||||
if show_recommendations:
|
||||
recommended = recommend_engine(hardware)
|
||||
console.print(f"\n[bold]Recommended Engine:[/bold] [cyan]{recommended}[/cyan]")
|
||||
|
||||
|
||||
def _show_config_template(console: Console, config_path: Path) -> None:
|
||||
"""Show default config template when config file doesn't exist."""
|
||||
from openjarvis.core.config import (
|
||||
DEFAULT_CONFIG_DIR,
|
||||
detect_hardware,
|
||||
generate_default_toml,
|
||||
)
|
||||
|
||||
console.print(f"[yellow]Config file not found: {config_path}[/yellow]")
|
||||
config_location = str(DEFAULT_CONFIG_DIR / "config.toml")
|
||||
msg = f"Create one at {config_location} or use --path to specify a location."
|
||||
console.print(f"[dim]{msg}[/dim]")
|
||||
|
||||
console.print("\n[bold]Default Configuration Template:[/bold]")
|
||||
hw = detect_hardware()
|
||||
template = generate_default_toml(hw)
|
||||
syntax = Syntax(template, "toml", theme="monokai", line_numbers=True)
|
||||
console.print(Panel(syntax, border_style="dim"))
|
||||
|
||||
|
||||
def _show_loaded_config(console: Console, config_path: Path, as_json: bool) -> None:
|
||||
"""Show the loaded effective configuration from config.toml."""
|
||||
from openjarvis.core.config import load_config
|
||||
|
||||
console.print(f"[dim]Loading config from: {config_path}[/dim]")
|
||||
|
||||
if config_path.exists():
|
||||
config = load_config(config_path)
|
||||
|
||||
if as_json:
|
||||
# Convert the dataclass to a dict and output as JSON
|
||||
from dataclasses import fields, is_dataclass
|
||||
|
||||
def convert(obj):
|
||||
if obj is None:
|
||||
return None
|
||||
if isinstance(obj, (str, int, float, bool)):
|
||||
return obj
|
||||
if isinstance(obj, (list, tuple)):
|
||||
return [convert(item) for item in obj]
|
||||
if isinstance(obj, dict):
|
||||
return {k: convert(v) for k, v in obj.items()}
|
||||
if is_dataclass(obj):
|
||||
result = {}
|
||||
for field in fields(obj):
|
||||
field_value = getattr(obj, field.name)
|
||||
if field_value is not None:
|
||||
result[field.name] = convert(field_value)
|
||||
return result
|
||||
# Fallback for other types
|
||||
return str(obj)
|
||||
|
||||
config_dict = convert(config)
|
||||
# Write JSON to stdout so it is pipeable
|
||||
stdout_console = Console()
|
||||
stdout_console.print_json(json.dumps(config_dict, indent=2, default=str))
|
||||
else:
|
||||
# Show as formatted table
|
||||
console.print("[bold]Engine Configuration[/bold]")
|
||||
console.print(f" Default Engine: [cyan]{config.engine.default}[/cyan]")
|
||||
|
||||
console.print("\n[bold]Intelligence Configuration[/bold]")
|
||||
console.print(
|
||||
f" Default Model: [cyan]{config.intelligence.default_model}[/cyan]"
|
||||
)
|
||||
fallback = config.intelligence.fallback_model or "N/A"
|
||||
console.print(f" Fallback Model: [cyan]{fallback}[/cyan]")
|
||||
console.print(
|
||||
f" Temperature: [cyan]{config.intelligence.temperature}[/cyan]"
|
||||
)
|
||||
console.print(
|
||||
f" Max Tokens: [cyan]{config.intelligence.max_tokens}[/cyan]"
|
||||
)
|
||||
|
||||
console.print("\n[bold]Agent Configuration[/bold]")
|
||||
console.print(f" Default Agent: [cyan]{config.agent.default_agent}[/cyan]")
|
||||
console.print(f" Max Turns: [cyan]{config.agent.max_turns}[/cyan]")
|
||||
console.print(f" Tools: [cyan]{config.agent.tools or 'none'}[/cyan]")
|
||||
ctx_mem = config.agent.context_from_memory
|
||||
console.print(f" Context from Memory: [cyan]{ctx_mem}[/cyan]")
|
||||
|
||||
# Show hardware info
|
||||
_show_hardware_info(console)
|
||||
|
||||
else:
|
||||
_show_config_template(console, config_path)
|
||||
|
||||
|
||||
def _show_toml_config(console: Console, config_path: Path) -> None:
|
||||
"""Show the raw TOML configuration file content with syntax highlighting."""
|
||||
console.print(f"[dim]Loading config from: {config_path}[/dim]")
|
||||
|
||||
if config_path.exists():
|
||||
config_content = config_path.read_text()
|
||||
syntax = Syntax(config_content, "toml", theme="monokai", line_numbers=True)
|
||||
console.print(Panel(syntax, title="Config File", border_style="cyan"))
|
||||
else:
|
||||
_show_config_template(console, config_path)
|
||||
|
||||
|
||||
def _show_json_config(console: Console, config_path: Path) -> None:
|
||||
"""Show the parsed TOML configuration as JSON."""
|
||||
console.print(f"[dim]Loading config from: {config_path}[/dim]")
|
||||
|
||||
if config_path.exists():
|
||||
config_content = config_path.read_text()
|
||||
|
||||
try:
|
||||
import tomllib # Python 3.11+
|
||||
except ModuleNotFoundError:
|
||||
import tomli as tomllib # type: ignore[no-redef]
|
||||
|
||||
config_dict = tomllib.loads(config_content)
|
||||
# Write JSON to stdout so it is pipeable
|
||||
stdout_console = Console()
|
||||
stdout_console.print_json(json.dumps(config_dict, indent=2))
|
||||
else:
|
||||
_show_config_template(console, config_path)
|
||||
|
||||
|
||||
def _show_hardware(console: Console) -> None:
|
||||
"""Show detected hardware information with recommended engine and model."""
|
||||
from openjarvis.core.config import (
|
||||
detect_hardware,
|
||||
recommend_engine,
|
||||
recommend_model,
|
||||
)
|
||||
|
||||
hardware = detect_hardware()
|
||||
_show_hardware_info(console, show_recommendations=False)
|
||||
|
||||
# Show recommended engine
|
||||
recommended_engine = recommend_engine(hardware)
|
||||
console.print(
|
||||
f"\n[bold]Recommended Engine:[/bold] [cyan]{recommended_engine}[/cyan]"
|
||||
)
|
||||
|
||||
# Show recommended model
|
||||
console.print("\n[bold]Model Recommendations[/bold]")
|
||||
recommended_model = recommend_model(hardware, recommended_engine)
|
||||
console.print(f" Recommended Model: [cyan]{recommended_model}[/cyan]")
|
||||
|
||||
|
||||
# Nested group for show sub-commands
|
||||
@click.group(invoke_without_command=True)
|
||||
@click.option("--path", "-p", default=None, help="Explicit config file path")
|
||||
@click.option("--json", "as_json", is_flag=True, default=False, help="Output as JSON")
|
||||
@click.pass_context
|
||||
def show_group(ctx: click.Context, path: str | None, as_json: bool) -> None:
|
||||
"""Show configuration details."""
|
||||
# Default to 'loaded' if no subcommand is invoked
|
||||
if ctx.invoked_subcommand is None:
|
||||
console = Console(stderr=True)
|
||||
try:
|
||||
config_path = _get_config_path(path)
|
||||
_show_loaded_config(console, config_path, as_json)
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error: {exc}[/red]")
|
||||
|
||||
|
||||
@show_group.command()
|
||||
@click.option("--path", "-p", default=None, help="Explicit config file path")
|
||||
@click.option("--json", "as_json", is_flag=True, default=False, help="Output as JSON")
|
||||
def loaded(path: str | None, as_json: bool) -> None:
|
||||
"""Show the loaded effective configuration from config.toml."""
|
||||
console = Console(stderr=True)
|
||||
try:
|
||||
config_path = _get_config_path(path)
|
||||
_show_loaded_config(console, config_path, as_json)
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error: {exc}[/red]")
|
||||
|
||||
|
||||
@show_group.command()
|
||||
@click.option("--path", "-p", default=None, help="Explicit config file path")
|
||||
def toml(path: str | None) -> None:
|
||||
"""Show the raw TOML configuration file content with syntax highlighting."""
|
||||
console = Console(stderr=True)
|
||||
try:
|
||||
config_path = _get_config_path(path)
|
||||
_show_toml_config(console, config_path)
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error: {exc}[/red]")
|
||||
|
||||
|
||||
@show_group.command("json")
|
||||
@click.option("--path", "-p", default=None, help="Explicit config file path")
|
||||
def as_json(path: str | None) -> None:
|
||||
"""Show the parsed TOML configuration as JSON."""
|
||||
console = Console(stderr=True)
|
||||
try:
|
||||
config_path = _get_config_path(path)
|
||||
_show_json_config(console, config_path)
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error: {exc}[/red]")
|
||||
|
||||
|
||||
@show_group.command()
|
||||
def hardware() -> None:
|
||||
"""Show detected hardware information with recommended engine and model."""
|
||||
console = Console(stderr=True)
|
||||
try:
|
||||
_show_hardware(console)
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error: {exc}[/red]")
|
||||
|
||||
|
||||
# Register the show group under config
|
||||
config.add_command(show_group, "show")
|
||||
|
||||
|
||||
__all__ = ["config"]
|
||||
@@ -0,0 +1,146 @@
|
||||
"""``jarvis registry`` — registry inspection commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import click
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
|
||||
def _load_registry_map() -> tuple[dict[str, object], dict[str, object]]:
|
||||
"""Import all registries and return (by_name, aliases) lookup dicts."""
|
||||
from openjarvis.core.registry import (
|
||||
AgentRegistry,
|
||||
BenchmarkRegistry,
|
||||
ChannelRegistry,
|
||||
CompressionRegistry,
|
||||
EngineRegistry,
|
||||
LearningRegistry,
|
||||
MemoryRegistry,
|
||||
ModelRegistry,
|
||||
RouterPolicyRegistry,
|
||||
SkillRegistry,
|
||||
SpeechRegistry,
|
||||
ToolRegistry,
|
||||
)
|
||||
|
||||
by_name = {
|
||||
"ToolRegistry": ToolRegistry,
|
||||
"AgentRegistry": AgentRegistry,
|
||||
"EngineRegistry": EngineRegistry,
|
||||
"MemoryRegistry": MemoryRegistry,
|
||||
"ModelRegistry": ModelRegistry,
|
||||
"ChannelRegistry": ChannelRegistry,
|
||||
"LearningRegistry": LearningRegistry,
|
||||
"SkillRegistry": SkillRegistry,
|
||||
"BenchmarkRegistry": BenchmarkRegistry,
|
||||
"RouterPolicyRegistry": RouterPolicyRegistry,
|
||||
"SpeechRegistry": SpeechRegistry,
|
||||
"CompressionRegistry": CompressionRegistry,
|
||||
}
|
||||
|
||||
aliases: dict[str, object] = {}
|
||||
_alias_map = {
|
||||
"ToolRegistry": ("tool", "tools"),
|
||||
"AgentRegistry": ("agent", "agents"),
|
||||
"EngineRegistry": ("engine", "engines"),
|
||||
"MemoryRegistry": ("memory", "memories"),
|
||||
"ModelRegistry": ("model", "models"),
|
||||
"ChannelRegistry": ("channel", "channels"),
|
||||
"LearningRegistry": ("learning", "learnings"),
|
||||
"SkillRegistry": ("skill", "skills"),
|
||||
"BenchmarkRegistry": ("benchmark", "benchmarks"),
|
||||
"RouterPolicyRegistry": ("router", "routers"),
|
||||
"SpeechRegistry": ("speech", "speeches"),
|
||||
"CompressionRegistry": ("compression", "compressions"),
|
||||
}
|
||||
for class_name, cls in by_name.items():
|
||||
aliases[class_name] = cls
|
||||
for alias in _alias_map[class_name]:
|
||||
aliases[alias] = cls
|
||||
|
||||
return by_name, aliases
|
||||
|
||||
|
||||
@click.group()
|
||||
def registry() -> None:
|
||||
"""Inspect registered components — list registries, show entries."""
|
||||
|
||||
|
||||
@registry.command("list")
|
||||
def list_registries() -> None:
|
||||
"""List all available registries."""
|
||||
console = Console(stderr=True)
|
||||
|
||||
table = Table(title="Available Registries")
|
||||
table.add_column("Registry", style="cyan")
|
||||
table.add_column("Module", style="green")
|
||||
table.add_column("Entry Count", style="yellow")
|
||||
|
||||
try:
|
||||
by_name, _ = _load_registry_map()
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error loading registries: {exc}[/red]")
|
||||
return
|
||||
|
||||
module_path = "openjarvis.core.registry"
|
||||
for reg_name, registry_cls in by_name.items():
|
||||
try:
|
||||
count = len(registry_cls.keys())
|
||||
table.add_row(reg_name, module_path, str(count))
|
||||
except Exception as exc:
|
||||
table.add_row(reg_name, module_path, f"[red]Error: {exc}[/red]")
|
||||
|
||||
console.print(table)
|
||||
|
||||
|
||||
@registry.command()
|
||||
@click.argument("registry_name")
|
||||
@click.option(
|
||||
"--verbose", "-v", is_flag=True, default=False, help="Show full entry details"
|
||||
)
|
||||
def show(registry_name: str, verbose: bool) -> None:
|
||||
"""Show entries in a specific registry."""
|
||||
console = Console(stderr=True)
|
||||
|
||||
try:
|
||||
_, aliases = _load_registry_map()
|
||||
|
||||
registry_cls = aliases.get(registry_name)
|
||||
if registry_cls is None:
|
||||
console.print(f"[red]Unknown registry: {registry_name}[/red]")
|
||||
console.print(
|
||||
"[dim]Run 'jarvis registry list' to see available registries.[/dim]"
|
||||
)
|
||||
return
|
||||
|
||||
keys = registry_cls.keys()
|
||||
if not keys:
|
||||
console.print(f"[dim]{registry_name} is empty.[/dim]")
|
||||
return
|
||||
|
||||
console.print(f"[bold]{registry_name}[/bold] — {len(keys)} entry/entries")
|
||||
|
||||
if verbose:
|
||||
for key in keys:
|
||||
entry = registry_cls.get(key)
|
||||
console.print(f"\n [cyan]{key}[/cyan]")
|
||||
console.print(f" Type: {type(entry).__name__}")
|
||||
console.print(f" Value: {entry}")
|
||||
else:
|
||||
table = Table()
|
||||
table.add_column("Key", style="cyan")
|
||||
table.add_column("Type", style="green")
|
||||
table.add_column("Value", style="white", max_width=80)
|
||||
for key in keys:
|
||||
entry = registry_cls.get(key)
|
||||
entry_type = type(entry).__name__
|
||||
entry_value = str(entry)
|
||||
table.add_row(key, entry_type, entry_value)
|
||||
console.print(table)
|
||||
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error: {exc}[/red]")
|
||||
|
||||
|
||||
__all__ = ["registry"]
|
||||
@@ -0,0 +1,107 @@
|
||||
"""``jarvis tool`` — tool management commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import click
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
|
||||
@click.group()
|
||||
def tool() -> None:
|
||||
"""Manage tools — list, inspect."""
|
||||
|
||||
|
||||
@tool.command("list")
|
||||
def list_tools() -> None:
|
||||
"""List all registered tools with their descriptions."""
|
||||
console = Console(stderr=True)
|
||||
try:
|
||||
# Trigger tool registration by importing the tools module
|
||||
import openjarvis.tools # noqa: F401
|
||||
from openjarvis.core.registry import ToolRegistry
|
||||
|
||||
keys = sorted(ToolRegistry.keys())
|
||||
if not keys:
|
||||
console.print("[dim]No tools registered.[/dim]")
|
||||
return
|
||||
|
||||
table = Table(title="Registered Tools")
|
||||
table.add_column("Name", style="cyan")
|
||||
table.add_column("Description", style="green", max_width=60)
|
||||
table.add_column("Category", style="yellow")
|
||||
|
||||
for key in keys:
|
||||
tool_cls = ToolRegistry.get(key)
|
||||
description = ""
|
||||
category = ""
|
||||
|
||||
# Try to get spec from the class or an instance
|
||||
try:
|
||||
# Some tools may require initialization, so try with default init
|
||||
tool_instance = tool_cls() if callable(tool_cls) else tool_cls
|
||||
if hasattr(tool_instance, "spec"):
|
||||
spec = tool_instance.spec
|
||||
description = getattr(spec, "description", "")[:60]
|
||||
category = getattr(spec, "category", "")
|
||||
except Exception:
|
||||
# If instantiation fails, just show the key
|
||||
description = "N/A (instantiation error)"
|
||||
|
||||
table.add_row(key, description, category)
|
||||
|
||||
console.print(table)
|
||||
console.print(f"\n[dim]Total: {len(keys)} tool(s)[/dim]")
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error: {exc}[/red]")
|
||||
|
||||
|
||||
@tool.command()
|
||||
@click.argument("tool_name")
|
||||
def inspect(tool_name: str) -> None:
|
||||
"""Show detailed information about a specific tool."""
|
||||
console = Console(stderr=True)
|
||||
try:
|
||||
# Trigger tool registration by importing the tools module
|
||||
import openjarvis.tools # noqa: F401
|
||||
from openjarvis.core.registry import ToolRegistry
|
||||
|
||||
if not ToolRegistry.contains(tool_name):
|
||||
console.print(f"[red]Tool not found: {tool_name}[/red]")
|
||||
console.print("[dim]Run 'jarvis tool list' to see available tools.[/dim]")
|
||||
return
|
||||
|
||||
tool_cls = ToolRegistry.get(tool_name)
|
||||
console.print(f"[bold]{tool_name}[/bold]")
|
||||
|
||||
try:
|
||||
tool_instance = tool_cls() if callable(tool_cls) else tool_cls
|
||||
if hasattr(tool_instance, "spec"):
|
||||
spec = tool_instance.spec
|
||||
console.print(f" [cyan]Name:[/cyan] {getattr(spec, 'name', 'N/A')}")
|
||||
console.print(
|
||||
f" [cyan]Description:[/cyan] {getattr(spec, 'description', 'N/A')}"
|
||||
)
|
||||
category = getattr(spec, "category", "N/A") or "none"
|
||||
console.print(f" [cyan]Category:[/cyan] {category}")
|
||||
|
||||
params = getattr(spec, "parameters", {})
|
||||
if params:
|
||||
console.print(" [cyan]Parameters:[/cyan]")
|
||||
if isinstance(params, dict) and "properties" in params:
|
||||
for param_name, param_info in params["properties"].items():
|
||||
param_type = param_info.get("type", "any")
|
||||
param_desc = param_info.get("description", "")
|
||||
console.print(
|
||||
f" • {param_name}: {param_type} — {param_desc}"
|
||||
)
|
||||
else:
|
||||
console.print(f" {params}")
|
||||
except Exception as e:
|
||||
console.print(f"[yellow]Note: Could not instantiate tool: {e}[/yellow]")
|
||||
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Error: {exc}[/red]")
|
||||
|
||||
|
||||
__all__ = ["tool"]
|
||||
@@ -0,0 +1,210 @@
|
||||
"""Tests for the ``jarvis config`` CLI commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from openjarvis.cli import cli
|
||||
|
||||
|
||||
class TestConfigCmd:
|
||||
"""Test cases for the jarvis config CLI group."""
|
||||
|
||||
def test_config_group_help(self) -> None:
|
||||
"""Test that the config group help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["config", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "config" in result.output.lower()
|
||||
|
||||
def test_config_show_help(self) -> None:
|
||||
"""Test that the config show help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["config", "show", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "show" in result.output.lower()
|
||||
|
||||
def test_config_show_loaded_help(self) -> None:
|
||||
"""Test that the config show loaded help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["config", "show", "loaded", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "loaded" in result.output.lower()
|
||||
|
||||
def test_config_show_toml_help(self) -> None:
|
||||
"""Test that the config show toml help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["config", "show", "toml", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "toml" in result.output.lower()
|
||||
|
||||
def test_config_show_json_help(self) -> None:
|
||||
"""Test that the config show json help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["config", "show", "json", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "json" in result.output.lower()
|
||||
|
||||
def test_config_show_hardware_help(self) -> None:
|
||||
"""Test that the config show hardware help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["config", "show", "hardware", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "hardware" in result.output.lower()
|
||||
|
||||
def test_config_show_loaded_displays_config(self, tmp_path: Path) -> None:
|
||||
"""Test that config show loaded displays the configuration."""
|
||||
# Create a temporary config file
|
||||
config_file = tmp_path / "test_config.toml"
|
||||
config_file.write_text(
|
||||
"""
|
||||
[engine]
|
||||
default = "ollama"
|
||||
|
||||
[intelligence]
|
||||
default_model = "test-model"
|
||||
temperature = 0.7
|
||||
max_tokens = 1024
|
||||
|
||||
[agent]
|
||||
default_agent = "simple"
|
||||
max_turns = 5
|
||||
"""
|
||||
)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["config", "show", "loaded", "--path", str(config_file)]
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "ollama" in result.output
|
||||
assert "test-model" in result.output
|
||||
|
||||
def test_config_show_loaded_json_output(self, tmp_path: Path) -> None:
|
||||
"""Test that config show loaded --json outputs valid JSON."""
|
||||
# Create a temporary config file
|
||||
config_file = tmp_path / "test_config.toml"
|
||||
config_file.write_text(
|
||||
"""
|
||||
[engine]
|
||||
default = "ollama"
|
||||
|
||||
[intelligence]
|
||||
default_model = "test-model"
|
||||
temperature = 0.7
|
||||
"""
|
||||
)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["config", "show", "loaded", "--path", str(config_file), "--json"]
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
# Try to parse the output as JSON (may have prefix lines)
|
||||
# Find the JSON part by looking for the opening brace
|
||||
json_start = result.output.find("{")
|
||||
assert json_start >= 0, f"Could not find JSON in output: {result.output}"
|
||||
try:
|
||||
json_data = json.loads(result.output[json_start:])
|
||||
assert "engine" in json_data
|
||||
assert json_data["engine"]["default"] == "ollama"
|
||||
except json.JSONDecodeError:
|
||||
pytest.fail(f"Output is not valid JSON: {result.output}")
|
||||
|
||||
def test_config_show_toml_displays_raw_content(self, tmp_path: Path) -> None:
|
||||
"""Test that config show toml displays the raw TOML content."""
|
||||
# Create a temporary config file
|
||||
config_file = tmp_path / "test_config.toml"
|
||||
config_file.write_text('[engine]\ndefault = "ollama"\n')
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["config", "show", "toml", "--path", str(config_file)]
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "[engine]" in result.output
|
||||
assert "ollama" in result.output
|
||||
|
||||
def test_config_show_json_displays_parsed_content(self, tmp_path: Path) -> None:
|
||||
"""Test that config show json displays parsed TOML as JSON."""
|
||||
# Create a temporary config file
|
||||
config_file = tmp_path / "test_config.toml"
|
||||
config_file.write_text('[engine]\ndefault = "ollama"\n')
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["config", "show", "json", "--path", str(config_file)]
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
# The output should be valid JSON (may have prefix lines)
|
||||
json_start = result.output.find("{")
|
||||
assert json_start >= 0, f"Could not find JSON in output: {result.output}"
|
||||
try:
|
||||
json_data = json.loads(result.output[json_start:])
|
||||
assert json_data["engine"]["default"] == "ollama"
|
||||
except json.JSONDecodeError:
|
||||
pytest.fail(f"Output is not valid JSON: {result.output}")
|
||||
|
||||
def test_config_show_hardware_displays_info(self) -> None:
|
||||
"""Test that config show hardware displays hardware information."""
|
||||
result = CliRunner().invoke(cli, ["config", "show", "hardware"])
|
||||
assert result.exit_code == 0
|
||||
# Should display hardware-related content
|
||||
output = result.output.lower()
|
||||
assert "hardware" in output or "cpu" in output or "ram" in output
|
||||
|
||||
def test_config_show_default_to_loaded(self, tmp_path: Path) -> None:
|
||||
"""Test that config show (no subcommand) defaults to loaded."""
|
||||
# Create a temporary config file
|
||||
config_file = tmp_path / "test_config.toml"
|
||||
config_file.write_text('[engine]\ndefault = "ollama"\n')
|
||||
|
||||
result = CliRunner().invoke(cli, ["config", "show", "--path", str(config_file)])
|
||||
|
||||
assert result.exit_code == 0
|
||||
# Default behavior should show loaded config
|
||||
assert "ollama" in result.output
|
||||
|
||||
def test_config_show_no_config_file(self, tmp_path: Path) -> None:
|
||||
"""Test that config show handles missing config file gracefully."""
|
||||
# Create a path for a non-existent config file
|
||||
config_file = tmp_path / "nonexistent_config.toml"
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["config", "show", "toml", "--path", str(config_file)]
|
||||
)
|
||||
|
||||
# Should exit 0 and show a message about missing config
|
||||
assert result.exit_code == 0
|
||||
assert "not found" in result.output.lower() or "config" in result.output.lower()
|
||||
|
||||
def test_config_show_path_option(self, tmp_path: Path) -> None:
|
||||
"""Test that the --path option works for all subcommands."""
|
||||
# Create a custom config file
|
||||
custom_config = tmp_path / "custom.toml"
|
||||
custom_config.write_text('[engine]\ndefault = "custom-engine"\n')
|
||||
|
||||
# Test various subcommands with custom path
|
||||
result = CliRunner().invoke(
|
||||
cli, ["config", "show", "toml", "--path", str(custom_config)]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "custom-engine" in result.output
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["config", "show", "json", "--path", str(custom_config)]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "custom-engine" in result.output
|
||||
|
||||
def test_config_show_invalid_subcommand(self) -> None:
|
||||
"""Test that an invalid subcommand shows an error."""
|
||||
result = CliRunner().invoke(cli, ["config", "show", "invalid_subcommand_xyz"])
|
||||
# Should show error about unknown command
|
||||
assert result.exit_code != 0
|
||||
assert "no such" in result.output.lower() or "unknown" in result.output.lower()
|
||||
|
||||
def test_config_subcommands_not_available_standalone(self) -> None:
|
||||
"""Test that config subcommands are not available directly under config."""
|
||||
# These should not be available as direct children of config
|
||||
result = CliRunner().invoke(cli, ["config", "loaded"])
|
||||
assert result.exit_code != 0
|
||||
assert "no such" in result.output.lower() or "unknown" in result.output.lower()
|
||||
@@ -0,0 +1,154 @@
|
||||
"""Tests for the ``jarvis registry`` CLI commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from openjarvis.cli import cli
|
||||
from openjarvis.core.registry import (
|
||||
ToolRegistry,
|
||||
)
|
||||
|
||||
|
||||
class TestRegistryCmd:
|
||||
"""Test cases for the jarvis registry CLI group."""
|
||||
|
||||
def test_registry_group_help(self) -> None:
|
||||
"""Test that the registry group help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["registry", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "registry" in result.output.lower()
|
||||
|
||||
def test_registry_list_help(self) -> None:
|
||||
"""Test that the registry list help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["registry", "list", "--help"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_registry_show_help(self) -> None:
|
||||
"""Test that the registry show help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "--help"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_registry_list_shows_all_registries(self) -> None:
|
||||
"""Test that registry list displays all available registries."""
|
||||
result = CliRunner().invoke(cli, ["registry", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should show registry-related content
|
||||
output = result.output.lower()
|
||||
assert "registry" in output
|
||||
|
||||
def test_registry_show_unknown_registry(self) -> None:
|
||||
"""Test that showing an unknown registry shows an error."""
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "unknown_registry_xyz"])
|
||||
assert result.exit_code == 0 # CLI still exits 0, just shows error message
|
||||
assert (
|
||||
"unknown" in result.output.lower() or "not found" in result.output.lower()
|
||||
)
|
||||
|
||||
def test_registry_show_tool_registry(self) -> None:
|
||||
"""Test that showing the tool registry displays entries."""
|
||||
# Trigger tool registration
|
||||
import openjarvis.tools # noqa: F401
|
||||
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "tool"])
|
||||
assert result.exit_code == 0
|
||||
# Should show tool-related content
|
||||
assert "tool" in result.output.lower() or "key" in result.output.lower()
|
||||
|
||||
def test_registry_show_tool_registry_verbose(self) -> None:
|
||||
"""Test that showing the tool registry with verbose flag shows details."""
|
||||
# Trigger tool registration
|
||||
import openjarvis.tools # noqa: F401
|
||||
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "tool", "-v"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_registry_show_empty_registry(self) -> None:
|
||||
"""Test registry show behavior with an empty registry."""
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "agent"])
|
||||
assert result.exit_code == 0
|
||||
# Should handle empty registries gracefully
|
||||
assert "agent" in result.output.lower() or "empty" in result.output.lower()
|
||||
|
||||
def test_registry_show_accepts_aliases(self) -> None:
|
||||
"""Test that registry show accepts various aliases."""
|
||||
# Trigger tool registration
|
||||
import openjarvis.tools # noqa: F401
|
||||
|
||||
# Test with 'tools' alias
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "tools"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_registry_keys_match_actual_registries(self) -> None:
|
||||
"""Test that the list command shows all expected registry names."""
|
||||
result = CliRunner().invoke(cli, ["registry", "list"])
|
||||
assert result.exit_code == 0
|
||||
output = result.output
|
||||
|
||||
# Check that key registries are mentioned
|
||||
assert "ToolRegistry" in output or "tool" in output.lower()
|
||||
assert "EngineRegistry" in output or "engine" in output.lower()
|
||||
assert "MemoryRegistry" in output or "memory" in output.lower()
|
||||
assert "ChannelRegistry" in output or "channel" in output.lower()
|
||||
|
||||
def test_registry_show_nonexistent_key(self) -> None:
|
||||
"""Test that showing a nonexistent key in a registry is handled."""
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "nonexistent"])
|
||||
assert result.exit_code == 0
|
||||
assert (
|
||||
"unknown" in result.output.lower() or "not found" in result.output.lower()
|
||||
)
|
||||
|
||||
def test_registry_list_handles_import_error(self) -> None:
|
||||
"""Test that registry list handles import errors gracefully."""
|
||||
# This tests the exception handling path in list_registries
|
||||
result = CliRunner().invoke(cli, ["registry", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should still complete even if some registries have issues
|
||||
|
||||
def test_registry_show_handles_exception(self) -> None:
|
||||
"""Test that registry show handles exceptions gracefully."""
|
||||
# Patch the registry to raise an exception during keys() call
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch(
|
||||
"openjarvis.core.registry.ToolRegistry.keys",
|
||||
side_effect=Exception("Test error"),
|
||||
):
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "tool"])
|
||||
assert result.exit_code == 0
|
||||
assert "error" in result.output.lower()
|
||||
|
||||
def test_registry_list_handles_registry_error(self) -> None:
|
||||
"""Test that registry list handles errors for specific registries."""
|
||||
from unittest.mock import patch
|
||||
|
||||
# Make one of the registry classes raise an error during keys()
|
||||
with patch.object(ToolRegistry, "keys", side_effect=Exception("Import error")):
|
||||
result = CliRunner().invoke(cli, ["registry", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should still complete and show other registries
|
||||
|
||||
def test_registry_show_handles_error_during_import(self) -> None:
|
||||
"""Test that registry show handles exceptions during import."""
|
||||
from unittest.mock import patch
|
||||
|
||||
# Patch the entire import to raise an exception
|
||||
with patch.dict("sys.modules", {"openjarvis.core.registry": None}):
|
||||
# This tests the outer exception handler
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "tool"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_registry_show_handles_error_during_iteration(self) -> None:
|
||||
"""Test that registry show handles exceptions during entry iteration."""
|
||||
from unittest.mock import patch
|
||||
|
||||
# Patch keys to return a fake entry so iteration happens,
|
||||
# then patch get to raise
|
||||
with patch.object(ToolRegistry, "keys", return_value=["fake_tool"]):
|
||||
with patch.object(
|
||||
ToolRegistry, "get", side_effect=Exception("Iteration error")
|
||||
):
|
||||
result = CliRunner().invoke(cli, ["registry", "show", "tool"])
|
||||
assert result.exit_code == 0
|
||||
assert "error" in result.output.lower()
|
||||
@@ -0,0 +1,238 @@
|
||||
"""Tests for the ``jarvis tool`` CLI commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from openjarvis.cli import cli
|
||||
from openjarvis.core.registry import ToolRegistry
|
||||
|
||||
|
||||
class TestToolCmd:
|
||||
"""Test cases for the jarvis tool CLI group."""
|
||||
|
||||
def test_tool_group_help(self) -> None:
|
||||
"""Test that the tool group help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["tool", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "tool" in result.output.lower() or "list" in result.output
|
||||
|
||||
def test_tool_list_help(self) -> None:
|
||||
"""Test that the tool list help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["tool", "list", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "list" in result.output.lower() or "registered" in result.output.lower()
|
||||
|
||||
def test_tool_inspect_help(self) -> None:
|
||||
"""Test that the tool inspect help displays correctly."""
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "tool" in result.output.lower() or "inspect" in result.output.lower()
|
||||
|
||||
def test_tool_list_shows_registered_tools(self) -> None:
|
||||
"""Test that tool list displays registered tools."""
|
||||
result = CliRunner().invoke(cli, ["tool", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should show at least some tools if they're registered
|
||||
output = result.output.lower()
|
||||
# The output should contain a table header or tool-related content
|
||||
assert "tool" in output or "name" in output or "description" in output
|
||||
|
||||
def test_tool_inspect_unknown_tool(self) -> None:
|
||||
"""Test that inspecting an unknown tool shows an error."""
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect", "nonexistent_tool_xyz"])
|
||||
assert result.exit_code == 0 # CLI still exits 0, just shows error message
|
||||
assert "not found" in result.output.lower() or "error" in result.output.lower()
|
||||
|
||||
def test_tool_inspect_known_tool(self) -> None:
|
||||
"""Test that inspecting a known tool shows details."""
|
||||
# First, trigger tool registration
|
||||
import openjarvis.tools # noqa: F401
|
||||
|
||||
# Get a known tool name
|
||||
registered_tools = ToolRegistry.keys()
|
||||
if registered_tools:
|
||||
tool_name = registered_tools[0]
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect", tool_name])
|
||||
assert result.exit_code == 0
|
||||
assert tool_name in result.output
|
||||
|
||||
def test_tool_list_empty_registry(self) -> None:
|
||||
"""Test tool list behavior with empty registry."""
|
||||
# This test verifies the command runs even if no tools are registered
|
||||
result = CliRunner().invoke(cli, ["tool", "list"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_tool_inspect_requires_tool_name(self) -> None:
|
||||
"""Test that inspect command requires a tool name argument."""
|
||||
# Running inspect without a tool name should show help or error
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect"])
|
||||
# Either shows help (exit 0) or error (exit non-zero)
|
||||
assert result.exit_code in (0, 2) # 0 for help, 2 for missing argument
|
||||
|
||||
def test_tool_list_with_registered_tools(self) -> None:
|
||||
"""Test that tool list shows details for registered tools."""
|
||||
# Trigger tool registration
|
||||
import openjarvis.tools # noqa: F401
|
||||
|
||||
result = CliRunner().invoke(cli, ["tool", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should either show tools or indicate no tools are registered
|
||||
output = result.output
|
||||
assert (
|
||||
"Registered Tools" in output
|
||||
or "No tools registered" in output
|
||||
or "Total:" in output
|
||||
)
|
||||
|
||||
def test_tool_list_handles_instantiation_error(self) -> None:
|
||||
"""Test that tool list handles tool instantiation errors gracefully."""
|
||||
from unittest.mock import patch
|
||||
|
||||
# Mock a tool class that raises an exception during instantiation
|
||||
with patch.object(ToolRegistry, "keys", return_value=["mock_tool"]):
|
||||
with patch.object(ToolRegistry, "get", return_value=object):
|
||||
result = CliRunner().invoke(cli, ["tool", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should still complete even if tools have instantiation issues
|
||||
assert "mock_tool" in result.output or "Total:" in result.output
|
||||
|
||||
def test_tool_inspect_with_spec_details(self) -> None:
|
||||
"""Test that inspect shows full spec details for tools with specs."""
|
||||
import openjarvis.tools # noqa: F401
|
||||
|
||||
registered_tools = ToolRegistry.keys()
|
||||
if registered_tools:
|
||||
tool_name = registered_tools[0]
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect", tool_name])
|
||||
assert result.exit_code == 0
|
||||
# Should show tool details including name, description, category
|
||||
output = result.output
|
||||
assert tool_name in output
|
||||
|
||||
def test_tool_inspect_handles_instantiation_error(self) -> None:
|
||||
"""Test that inspect handles tool instantiation errors gracefully."""
|
||||
from unittest.mock import patch
|
||||
|
||||
# Mock a tool that exists but fails to instantiate
|
||||
with patch.object(ToolRegistry, "contains", return_value=True):
|
||||
with patch.object(ToolRegistry, "get", return_value=object):
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect", "mock_tool"])
|
||||
assert result.exit_code == 0
|
||||
# Should show error note but still complete
|
||||
assert "mock_tool" in result.output
|
||||
|
||||
def test_tool_list_catches_registry_exception(self) -> None:
|
||||
"""Test that tool list handles exceptions during registry access."""
|
||||
from unittest.mock import patch
|
||||
|
||||
# Mock ToolRegistry.keys to raise an exception
|
||||
with patch.object(
|
||||
ToolRegistry, "keys", side_effect=Exception("Registry error")
|
||||
):
|
||||
result = CliRunner().invoke(cli, ["tool", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should catch exception and display error message
|
||||
assert "error" in result.output.lower()
|
||||
|
||||
def test_tool_inspect_catches_registry_exception(self) -> None:
|
||||
"""Test that tool inspect handles exceptions during registry access."""
|
||||
from unittest.mock import patch
|
||||
|
||||
# Mock ToolRegistry.contains to raise an exception
|
||||
with patch.object(
|
||||
ToolRegistry, "contains", side_effect=Exception("Registry error")
|
||||
):
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect", "mock_tool"])
|
||||
assert result.exit_code == 0
|
||||
# Should catch exception and display error message
|
||||
assert "error" in result.output.lower()
|
||||
|
||||
def test_tool_list_shows_tool_spec(self) -> None:
|
||||
"""Test that tool list displays spec details when available."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# Create a mock tool class with a spec
|
||||
mock_spec = MagicMock()
|
||||
mock_spec.description = "Test tool description"
|
||||
mock_spec.category = "test-category"
|
||||
|
||||
mock_tool_cls = MagicMock()
|
||||
mock_tool_instance = MagicMock()
|
||||
mock_tool_instance.spec = mock_spec
|
||||
mock_tool_cls.return_value = mock_tool_instance
|
||||
|
||||
with patch.object(ToolRegistry, "keys", return_value=["mock_tool"]):
|
||||
with patch.object(ToolRegistry, "get", return_value=mock_tool_cls):
|
||||
result = CliRunner().invoke(cli, ["tool", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should show spec details
|
||||
output = result.output
|
||||
assert "mock_tool" in output
|
||||
assert "Test tool description" in output
|
||||
|
||||
def test_tool_inspect_shows_full_spec_details(self) -> None:
|
||||
"""Test that inspect displays full spec details including parameters."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# Create a mock tool with full spec including parameters
|
||||
mock_spec = MagicMock()
|
||||
mock_spec.name = "MockTool"
|
||||
mock_spec.description = "A mock tool for testing"
|
||||
mock_spec.category = "testing"
|
||||
mock_spec.parameters = {
|
||||
"properties": {
|
||||
"input": {"type": "string", "description": "Input parameter"},
|
||||
"count": {"type": "integer", "description": "Count value"},
|
||||
}
|
||||
}
|
||||
|
||||
mock_tool_cls = MagicMock()
|
||||
mock_tool_instance = MagicMock()
|
||||
mock_tool_instance.spec = mock_spec
|
||||
mock_tool_cls.return_value = mock_tool_instance
|
||||
|
||||
with patch.object(ToolRegistry, "contains", return_value=True):
|
||||
with patch.object(ToolRegistry, "get", return_value=mock_tool_cls):
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect", "mock_tool"])
|
||||
assert result.exit_code == 0
|
||||
output = result.output
|
||||
assert "mock_tool" in output
|
||||
assert "MockTool" in output
|
||||
assert "A mock tool for testing" in output
|
||||
assert "testing" in output
|
||||
assert "Parameters" in output
|
||||
assert "input" in output
|
||||
assert "count" in output
|
||||
|
||||
def test_tool_inspect_handles_tool_without_spec(self) -> None:
|
||||
"""Test that inspect handles tools without spec attribute."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# Mock a tool class without spec attribute
|
||||
mock_tool_cls = MagicMock()
|
||||
del mock_tool_cls.spec # Remove spec attribute
|
||||
|
||||
with patch.object(ToolRegistry, "contains", return_value=True):
|
||||
with patch.object(ToolRegistry, "get", return_value=mock_tool_cls):
|
||||
result = CliRunner().invoke(cli, ["tool", "inspect", "mock_tool"])
|
||||
assert result.exit_code == 0
|
||||
# Should handle gracefully
|
||||
assert "mock_tool" in result.output
|
||||
|
||||
def test_tool_list_handles_tool_without_spec(self) -> None:
|
||||
"""Test that tool list handles tools without spec attribute."""
|
||||
from unittest.mock import patch
|
||||
|
||||
# Create a simple class without spec attribute
|
||||
class ToolWithoutSpec:
|
||||
pass
|
||||
|
||||
mock_tool_cls = ToolWithoutSpec
|
||||
|
||||
with patch.object(ToolRegistry, "keys", return_value=["mock_tool"]):
|
||||
with patch.object(ToolRegistry, "get", return_value=mock_tool_cls):
|
||||
result = CliRunner().invoke(cli, ["tool", "list"])
|
||||
assert result.exit_code == 0
|
||||
# Should handle gracefully and show the tool name
|
||||
assert "mock_tool" in result.output or "Total" in result.output
|
||||
Reference in New Issue
Block a user