diff --git a/python/packages/jumpstarter-mcp/jumpstarter_mcp/introspect.py b/python/packages/jumpstarter-mcp/jumpstarter_mcp/introspect.py index b100040f0..7816a7551 100644 --- a/python/packages/jumpstarter-mcp/jumpstarter_mcp/introspect.py +++ b/python/packages/jumpstarter-mcp/jumpstarter_mcp/introspect.py @@ -1,205 +1,25 @@ -"""Introspection utilities for Click CLI trees and driver object trees.""" - -from __future__ import annotations - -import inspect -import logging -from typing import Any - -import click -import click.core - -logger = logging.getLogger(__name__) - - -def walk_click_tree(cmd: click.core.BaseCommand, path: list[str] | None = None) -> dict[str, Any]: # ty: ignore[unresolved-attribute] - """Recursively walk a Click command tree and return structured JSON. - - Returns command names, help text, parameters (with types and defaults), - and nested subcommands. - """ - path = path or [] - result: dict[str, Any] = { - "name": cmd.name, - "help": cmd.help, - "params": [ - { - "name": p.name, - "type": str(p.type), - "help": getattr(p, "help", None), - "required": getattr(p, "required", False), - "default": p.default if p.default is not None else None, - } - for p in cmd.params - if not getattr(p, "hidden", False) and p.name != "help" - ], - } - if isinstance(cmd, click.Group): - result["subcommands"] = { - name: walk_click_tree(sub, path + [name]) - for name, sub in cmd.commands.items() - } - return result - - -# Methods on the base DriverClient that should be excluded from introspection -_BASE_METHODS = { - "call", - "streamingcall", - "stream", - "stream_async", - "log_stream", - "log_stream_async", - "open_stream", - "close", - "reset", - "cli", - "call_async", - "streamingcall_async", - "report", - "report_async", - "get_status_async", - "end_session_async", - "status_monitor_async", -} - -# Methods inherited from base classes that add noise to driver listings -_INTERNAL_METHODS = { - "check_exporter_status", - "resource_async", - "wait_for_hook_complete_monitored", - "wait_for_hook_status", - "wait_for_lease_ready", - "wait_for_lease_ready_monitored", -} - - -def _get_public_method_names(obj: Any) -> list[str]: - """Return public method names for a driver instance, excluding base internals. - - Uses getmembers_static to avoid triggering property descriptors, which can - make gRPC calls that fail when invoked from the event loop thread. - """ - names = [] - try: - members = inspect.getmembers_static(obj) - except Exception: - logger.debug("inspect.getmembers_static failed for %s", type(obj).__name__, exc_info=True) - return names - for name, value in members: - if name.startswith("_") or name in _BASE_METHODS or name in _INTERNAL_METHODS: - continue - if isinstance(value, property): - continue - if not callable(value): - continue - try: - inspect.signature(value) - except (ValueError, TypeError, AttributeError): - continue - names.append(name) - return names - - -def list_drivers(client: Any, _keys: list[str] | None = None) -> list[dict[str, Any]]: - """Flatten the driver client tree into a list with dot-separated paths. - - Returns path (Python access path like ``client.power``), driver_path - (children-key list like ``["power"]`` for use with get_driver_methods), - class name, description, and method names for each driver. - """ - keys = _keys or [] - cls = type(client) - results = [ - { - "path": f"client.{'.'.join(keys)}" if keys else "client", - "driver_path": keys, - "class": f"{cls.__module__}.{cls.__qualname__}", - "description": getattr(client, "description", None), - "methods": _get_public_method_names(client), - } - ] - children = getattr(client, "children", {}) - for name, child in children.items(): - results.extend(list_drivers(child, keys + [name])) - return results - - -def _inspect_method(name: str, method: Any, driver_path: list[str]) -> dict[str, Any] | None: - """Build a method descriptor dict, or return None if the method can't be inspected.""" - try: - sig = inspect.signature(method) - except (ValueError, TypeError, AttributeError): - return None - - is_streaming = False - try: - source = inspect.getsource(method) - is_streaming = "streamingcall" in source - except (OSError, TypeError): - pass - - params = [ - { - "name": pname, - "annotation": str(p.annotation) if p.annotation != inspect.Parameter.empty else None, - "default": str(p.default) if p.default != inspect.Parameter.empty else None, - } - for pname, p in sig.parameters.items() - if pname != "self" - ] - - attr_path = ".".join(driver_path) - call_args = ", ".join(f"{p['name']}=..." for p in params) - method_call = f"client.{attr_path}.{name}({call_args})" - - return { - "name": name, - "signature": str(sig), - "docstring": inspect.getdoc(method), - "parameters": params, - "return_type": str(sig.return_annotation) if sig.return_annotation != inspect.Signature.empty else None, - "is_streaming": is_streaming, - "call_example": ( - "from jumpstarter.utils.env import env\n\n" - "with env() as client:\n" - f" {method_call}" - ), - } - - -def get_driver_methods(client: Any, driver_path: list[str]) -> dict[str, Any]: - """Inspect a specific driver client at the given path in the children tree. - - Uses Python inspect to return detailed method information for all public - methods defined on the concrete driver class. - """ - target = client - for key in driver_path: - children = getattr(target, "children", {}) - if key not in children: - raise KeyError(f"Driver path component '{key}' not found. Available: {list(children.keys())}") - target = children[key] - - cls = type(target) - try: - members = inspect.getmembers_static(target) - except Exception: - logger.debug("inspect.getmembers_static failed for %s", cls.__name__, exc_info=True) - members = [] - - methods = [] - for name, value in members: - if name.startswith("_") or name in _BASE_METHODS or name in _INTERNAL_METHODS: - continue - if isinstance(value, property) or not callable(value): - continue - info = _inspect_method(name, value, driver_path) - if info is not None: - methods.append(info) - - return { - "class": f"{cls.__module__}.{cls.__qualname__}", - "driver_path": driver_path, - "methods": methods, - } +"""Backward-compatible re-exports of the driver introspection utilities. + +The implementation lives in jumpstarter.client.introspect so it can be shared +between the CLI and the MCP server. +""" + +from jumpstarter.client.introspect import ( + _BASE_METHODS, + _INTERNAL_METHODS, + _get_public_method_names, + _inspect_method, + get_driver_methods, + list_drivers, + walk_click_tree, +) + +__all__ = [ + "_BASE_METHODS", + "_INTERNAL_METHODS", + "_get_public_method_names", + "_inspect_method", + "get_driver_methods", + "list_drivers", + "walk_click_tree", +] diff --git a/python/packages/jumpstarter-mcp/jumpstarter_mcp/server_test.py b/python/packages/jumpstarter-mcp/jumpstarter_mcp/server_test.py index afb646232..cc2fce59e 100644 --- a/python/packages/jumpstarter-mcp/jumpstarter_mcp/server_test.py +++ b/python/packages/jumpstarter-mcp/jumpstarter_mcp/server_test.py @@ -10,16 +10,9 @@ from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch -import click import pytest from jumpstarter_mcp.connections import Connection, ConnectionManager -from jumpstarter_mcp.introspect import ( - _get_public_method_names, - get_driver_methods, - list_drivers, - walk_click_tree, -) from jumpstarter_mcp.server import ( TOKEN_REFRESH_THRESHOLD_SECONDS, _ensure_fresh_token, @@ -112,135 +105,20 @@ class FakeLease: # --------------------------------------------------------------------------- -# walk_click_tree +# Introspection re-exports # --------------------------------------------------------------------------- -class TestWalkClickTree: - def test_simple_command(self): - @click.command("hello") - @click.option("--name", help="Your name") - def hello(name): - """Say hello.""" - - result = walk_click_tree(hello) - assert result["name"] == "hello" - assert result["help"] == "Say hello." - assert len(result["params"]) == 1 - assert result["params"][0]["name"] == "name" - assert result["params"][0]["help"] == "Your name" - assert "subcommands" not in result - - def test_group_with_subcommands(self): - @click.group("root") - def root(): - """Root group.""" - - @root.command("sub1") - def sub1(): - """First sub.""" - - @root.command("sub2") - @click.option("--count", type=int, default=5) - def sub2(count): - """Second sub.""" - - result = walk_click_tree(root) - assert result["name"] == "root" - assert "subcommands" in result - assert "sub1" in result["subcommands"] - assert "sub2" in result["subcommands"] - assert result["subcommands"]["sub2"]["params"][0]["name"] == "count" - assert result["subcommands"]["sub2"]["params"][0]["default"] == 5 - - def test_hidden_params_excluded(self): - @click.command("cmd") - @click.option("--visible", help="shown") - @click.option("--secret", hidden=True) - def cmd(visible, secret): - pass - - result = walk_click_tree(cmd) - names = [p["name"] for p in result["params"]] - assert "visible" in names - assert "secret" not in names - - -# Introspection tests -# --------------------------------------------------------------------------- +class TestIntrospectReexports: + def test_reexports_core_implementation(self): + import jumpstarter_mcp.introspect as shim + import jumpstarter.client.introspect as core -class TestGetPublicMethodNames: - def test_filters_private_and_base_methods(self): - names = _get_public_method_names(FakePowerClient()) - assert "on" in names - assert "off" in names - assert "cycle" in names - assert "_private" not in names - assert "call" not in names - assert "check_exporter_status" not in names - - -class TestListDrivers: - def test_flat_tree(self): - client = FakeCompositeClient() - result = list_drivers(client) - paths = [d["path"] for d in result] - assert "client" in paths - assert "client.power" in paths - assert "client.serial" in paths - - def test_driver_path_field(self): - client = FakeCompositeClient() - result = list_drivers(client) - root = next(d for d in result if d["path"] == "client") - assert root["driver_path"] == [] - power = next(d for d in result if d["path"] == "client.power") - assert power["driver_path"] == ["power"] - - def test_methods_populated(self): - client = FakeCompositeClient() - result = list_drivers(client) - power = next(d for d in result if d["path"] == "client.power") - assert "on" in power["methods"] - assert "off" in power["methods"] - - -class TestGetDriverMethods: - def test_returns_method_details(self): - client = FakeCompositeClient() - result = get_driver_methods(client, ["power"]) - assert result["driver_path"] == ["power"] - method_names = [m["name"] for m in result["methods"]] - assert "on" in method_names - assert "off" in method_names - assert "cycle" in method_names - - def test_call_example_uses_dot_notation(self): - client = FakeCompositeClient() - result = get_driver_methods(client, ["power"]) - cycle = next(m for m in result["methods"] if m["name"] == "cycle") - assert "client.power.cycle(" in cycle["call_example"] - assert 'children["power"]' not in cycle["call_example"] - - def test_invalid_path_raises(self): - client = FakeCompositeClient() - with pytest.raises(KeyError, match="nonexistent"): - get_driver_methods(client, ["nonexistent"]) - - def test_docstrings_captured(self): - client = FakeCompositeClient() - result = get_driver_methods(client, ["power"]) - on_method = next(m for m in result["methods"] if m["name"] == "on") - assert on_method["docstring"] == "Power on the device." - - def test_parameters_captured(self): - client = FakeCompositeClient() - result = get_driver_methods(client, ["power"]) - cycle = next(m for m in result["methods"] if m["name"] == "cycle") - assert len(cycle["parameters"]) == 1 - assert cycle["parameters"][0]["name"] == "wait" - assert cycle["parameters"][0]["default"] == "2" + assert shim.walk_click_tree is core.walk_click_tree + assert shim.list_drivers is core.list_drivers + assert shim.get_driver_methods is core.get_driver_methods + assert shim._get_public_method_names is core._get_public_method_names # --------------------------------------------------------------------------- diff --git a/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/commands.py b/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/commands.py index 8a8259da7..49bd8a273 100644 --- a/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/commands.py +++ b/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/commands.py @@ -12,7 +12,8 @@ import click from jumpstarter_mcp.connections import ConnectionManager -from jumpstarter_mcp.introspect import get_driver_methods, list_drivers, walk_click_tree + +from jumpstarter.client.introspect import get_driver_methods, list_drivers, walk_click_tree logger = logging.getLogger(__name__) @@ -119,7 +120,10 @@ async def drivers( ) -> list[dict]: """List all drivers in the client tree.""" conn = manager.get_connection(connection_id) - return list_drivers(conn.client) + # Off the loop thread, as explore() does. These walk a sync client facade + # over a blocking portal; they read attributes statically today and so do + # not dispatch, but a driver property that ever did would deadlock the loop. + return await anyio.to_thread.run_sync(list_drivers, conn.client) async def driver_methods( @@ -129,4 +133,4 @@ async def driver_methods( ) -> dict: """Inspect methods on a specific driver.""" conn = manager.get_connection(connection_id) - return get_driver_methods(conn.client, driver_path) + return await anyio.to_thread.run_sync(get_driver_methods, conn.client, driver_path) diff --git a/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/commands_test.py b/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/commands_test.py new file mode 100644 index 000000000..361c6f57c --- /dev/null +++ b/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/commands_test.py @@ -0,0 +1,61 @@ +"""Introspection must drive synchronous driver facades outside the event loop.""" + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import click +import pytest +from anyio import fail_after +from anyio.from_thread import BlockingPortal + +from jumpstarter_mcp.tools.commands import driver_methods, drivers, explore + + +class LeafClient: + children: dict = {} + + def on(self): + """Turn power on.""" + + +class PortalClient: + """Model driver attributes/CLI builders that dispatch through a live portal. + + Calling these on the event-loop thread raises immediately, just like a + synchronous driver facade. A static fake would miss that regression. + """ + + def __init__(self, portal): + self.portal = portal + + @property + def description(self): + return self.portal.call(lambda: "A remotely described driver") + + @property + def children(self): + return self.portal.call(lambda: {"power": LeafClient()}) + + def cli(self): + self.portal.call(lambda: None) + return click.Group("j", commands={"on": click.Command("on")}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tool", ["drivers", "driver_methods", "explore"]) +async def test_introspection_dispatches_off_the_loop(tool): + with fail_after(5): + async with BlockingPortal() as portal: + manager = MagicMock() + manager.get_connection.return_value = SimpleNamespace(client=PortalClient(portal)) + if tool == "drivers": + result = await drivers(manager, "connection-1") + assert result[0]["description"] == "A remotely described driver" + assert result[1]["driver_path"] == ["power"] + elif tool == "driver_methods": + result = await driver_methods(manager, "connection-1", ["power"]) + assert [method["name"] for method in result["methods"]] == ["on"] + else: + result = await explore(manager, "connection-1") + assert "on" in result["subcommands"] + manager.get_connection.assert_called_once_with("connection-1") diff --git a/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/connections.py b/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/connections.py index 6487ed8d0..85d889b7e 100644 --- a/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/connections.py +++ b/python/packages/jumpstarter-mcp/jumpstarter_mcp/tools/connections.py @@ -8,8 +8,8 @@ import anyio.to_thread # noqa: F401 from jumpstarter_mcp.connections import ConnectionManager -from jumpstarter_mcp.introspect import list_drivers, walk_click_tree +from jumpstarter.client.introspect import list_drivers, walk_click_tree from jumpstarter.config.client import ClientConfigV1Alpha1 logger = logging.getLogger(__name__) diff --git a/python/packages/jumpstarter/jumpstarter/client/__init__.py b/python/packages/jumpstarter/jumpstarter/client/__init__.py index 62dbf689c..b8df2a654 100644 --- a/python/packages/jumpstarter/jumpstarter/client/__init__.py +++ b/python/packages/jumpstarter/jumpstarter/client/__init__.py @@ -1,6 +1,17 @@ from .base import DriverClient from .client import client_from_path from .flasher import FlasherClient, FlasherClientInterface +from .introspect import describe_client, describe_devices, describe_devices_async from .lease import DirectLease, Lease -__all__ = ["DriverClient", "DirectLease", "FlasherClient", "FlasherClientInterface", "client_from_path", "Lease"] +__all__ = [ + "DriverClient", + "DirectLease", + "FlasherClient", + "FlasherClientInterface", + "client_from_path", + "Lease", + "describe_client", + "describe_devices", + "describe_devices_async", +] diff --git a/python/packages/jumpstarter/jumpstarter/client/introspect.py b/python/packages/jumpstarter/jumpstarter/client/introspect.py new file mode 100644 index 000000000..68beec77a --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/client/introspect.py @@ -0,0 +1,339 @@ +"""Introspection utilities for Click CLI trees and driver object trees.""" + +from __future__ import annotations + +import inspect +import logging +from collections.abc import Mapping +from contextlib import ExitStack, asynccontextmanager +from datetime import timedelta +from typing import TYPE_CHECKING, Any + +import click +import click.core +from anyio import to_thread +from anyio.from_thread import BlockingPortal, start_blocking_portal + +from .base import StubDriverClient +from .client import client_from_path + +if TYPE_CHECKING: + from jumpstarter.config.client import ClientConfigV1Alpha1 + +logger = logging.getLogger(__name__) + + +def _json_safe(value: Any) -> Any: + """Coerce a value to a JSON-serializable equivalent, stringifying as a last resort.""" + if value is None or isinstance(value, (bool, int, float, str)): + return value + if isinstance(value, (list, tuple)): + return [_json_safe(v) for v in value] + if isinstance(value, Mapping): + result: dict[str, Any] = {} + for key, item in value.items(): + json_key = str(key) + if json_key in result: + raise ValueError(f"Mapping keys collide after stringification: {json_key!r}") + result[json_key] = _json_safe(item) + return result + return str(value) + + +def _describe_param(param: click.Parameter) -> dict[str, Any]: + """Describe one Click parameter well enough to prompt for it. + + Beyond name and help, a caller building a command line needs to know + whether the parameter is positional or an option (and which flag spells + it), whether it is a boolean flag that takes no value, whether it repeats, + and the values it accepts when it is a choice. + """ + described: dict[str, Any] = { + "name": param.name, + "kind": param.param_type_name, + # Click's own type name ("integer", "path", "choice"); str() on some + # types renders an object repr, which is no use in a prompt. + "type": getattr(param.type, "name", None) or str(param.type), + "help": getattr(param, "help", None), + "required": getattr(param, "required", False), + "default": _json_safe(param.default) if param.default is not None else None, + "opts": list(param.opts), + "is_flag": bool(getattr(param, "is_flag", False)), + "multiple": bool(getattr(param, "multiple", False)), + "nargs": param.nargs, + } + if isinstance(param.type, click.Choice): + described["choices"] = [str(choice) for choice in param.type.choices] + return described + + +def walk_click_tree(cmd: click.core.BaseCommand, path: list[str] | None = None) -> dict[str, Any]: # ty: ignore[unresolved-attribute] + """Recursively walk a Click command tree and return structured JSON. + + Returns command names, help text, parameters (with types and defaults), + and nested subcommands. + """ + path = path or [] + result: dict[str, Any] = { + "name": cmd.name, + "help": cmd.help, + "params": [ + _describe_param(p) for p in cmd.params if not getattr(p, "hidden", False) + ], + } + if isinstance(cmd, click.Group): + result["subcommands"] = { + name: walk_click_tree(sub, path + [name]) + for name, sub in cmd.commands.items() + } + return result + + +# Methods on the base DriverClient that should be excluded from introspection +_BASE_METHODS = { + "call", + "streamingcall", + "stream", + "stream_async", + "log_stream", + "log_stream_async", + "open_stream", + "close", + "reset", + "cli", + "call_async", + "streamingcall_async", + "report", + "report_async", + "get_status_async", + "end_session_async", + "status_monitor_async", +} + +# Methods inherited from base classes that add noise to driver listings +_INTERNAL_METHODS = { + "check_exporter_status", + "resource_async", + "wait_for_hook_complete_monitored", + "wait_for_hook_status", + "wait_for_lease_ready", + "wait_for_lease_ready_monitored", +} + + +def _get_public_method_names(obj: Any) -> list[str]: + """Return public method names for a driver instance, excluding base internals. + + Uses getmembers_static to avoid triggering property descriptors, which can + make gRPC calls that fail when invoked from the event loop thread. + """ + names = [] + try: + members = inspect.getmembers_static(obj) + except Exception: + logger.debug("inspect.getmembers_static failed for %s", type(obj).__name__, exc_info=True) + return names + for name, value in members: + if name.startswith("_") or name in _BASE_METHODS or name in _INTERNAL_METHODS: + continue + if isinstance(value, property): + continue + if not callable(value): + continue + try: + inspect.signature(value) + except (ValueError, TypeError, AttributeError): + continue + names.append(name) + return names + + +def list_drivers(client: Any, _keys: list[str] | None = None) -> list[dict[str, Any]]: + """Flatten the driver client tree into a list with dot-separated paths. + + Returns path (Python access path like ``client.power``), driver_path + (children-key list like ``["power"]`` for use with get_driver_methods), + class name, description, and method names for each driver. + """ + keys = _keys or [] + cls = type(client) + results = [ + { + "path": f"client.{'.'.join(keys)}" if keys else "client", + "driver_path": keys, + "class": f"{cls.__module__}.{cls.__qualname__}", + "description": getattr(client, "description", None), + "methods": _get_public_method_names(client), + } + ] + children = getattr(client, "children", {}) + for name, child in children.items(): + results.extend(list_drivers(child, keys + [name])) + return results + + +def _inspect_method(name: str, method: Any, driver_path: list[str]) -> dict[str, Any] | None: + """Build a method descriptor dict, or return None if the method can't be inspected.""" + try: + sig = inspect.signature(method) + except (ValueError, TypeError, AttributeError): + return None + + is_streaming = False + try: + source = inspect.getsource(method) + is_streaming = "streamingcall" in source + except (OSError, TypeError): + pass + + params = [ + { + "name": pname, + "annotation": str(p.annotation) if p.annotation != inspect.Parameter.empty else None, + "default": str(p.default) if p.default != inspect.Parameter.empty else None, + } + for pname, p in sig.parameters.items() + if pname != "self" + ] + + attr_path = ".".join(driver_path) + call_args = ", ".join(f"{p['name']}=..." for p in params) + receiver = f"client.{attr_path}" if attr_path else "client" + method_call = f"{receiver}.{name}({call_args})" + + return { + "name": name, + "signature": str(sig), + "docstring": inspect.getdoc(method), + "parameters": params, + "return_type": str(sig.return_annotation) if sig.return_annotation != inspect.Signature.empty else None, + "is_streaming": is_streaming, + "call_example": ( + "from jumpstarter.utils.env import env\n\n" + "with env() as client:\n" + f" {method_call}" + ), + } + + +def get_driver_methods(client: Any, driver_path: list[str]) -> dict[str, Any]: + """Inspect a specific driver client at the given path in the children tree. + + Uses Python inspect to return detailed method information for all public + methods defined on the concrete driver class. + """ + target = client + for key in driver_path: + children = getattr(target, "children", {}) + if key not in children: + raise KeyError(f"Driver path component '{key}' not found. Available: {list(children.keys())}") + target = children[key] + + cls = type(target) + try: + members = inspect.getmembers_static(target) + except Exception: + logger.debug("inspect.getmembers_static failed for %s", cls.__name__, exc_info=True) + members = [] + + methods = [] + for name, value in members: + if name.startswith("_") or name in _BASE_METHODS or name in _INTERNAL_METHODS: + continue + if isinstance(value, property) or not callable(value): + continue + info = _inspect_method(name, value, driver_path) + if info is not None: + methods.append(info) + + return { + "class": f"{cls.__module__}.{cls.__qualname__}", + "driver_path": driver_path, + "methods": methods, + } + + +def describe_client(client: Any) -> dict[str, Any]: + """Build a plain-serializable description of a connected driver client tree. + + Returns the flattened driver listing and the Click CLI tree. Drivers whose + client packages are not installed appear as StubDriverClient entries in the + listing, and cli_tree is None when the root client does not expose a CLI + (including when the root client itself is a stub). + """ + cli_tree = None + # Inherited counts: QemuFlasherClient, for one, defines no cli of its own and + # gets a real one from FlasherClientInterface, so looking only at + # type(client).__dict__ would drop the CLI of every such driver. + if not isinstance(client, StubDriverClient) and getattr(type(client), "cli", None) is not None: + try: + cli_tree = walk_click_tree(client.cli()) + except Exception: + # A driver whose cli() is broken should cost us its CLI tree, not the + # whole description — the driver listing below is still worth having. + logger.warning("could not build the CLI tree for %s", type(client).__name__, exc_info=True) + return { + "drivers": list_drivers(client), + "cli_tree": cli_tree, + } + + +@asynccontextmanager +async def _connect_lease(config: ClientConfigV1Alpha1, lease_name: str, portal: BlockingPortal): + """Attach to an existing lease and yield its root driver client. + + Passing lease_name into lease_async attaches to that lease rather than + creating one, and leaves it unreleased on exit. + """ + async with config.lease_async( + selector=None, + exporter_name=None, + lease_name=lease_name, + # Attaching by name never reaches Lease._create, and with selector None + # the "selector changed, make a new one" branch cannot fire either, so + # no duration is ever sent to the controller. Naming 30 minutes here + # only suggested this call could extend a lease that it cannot. + duration=timedelta(0), + portal=portal, + ) as lease: + async with lease.serve_unix_async() as path: + with ExitStack() as stack: + async with client_from_path( + path, portal, stack, allow=lease.allow, unsafe=lease.unsafe + ) as client: + yield client + + +def describe_devices(config: ClientConfigV1Alpha1, lease_name: str) -> dict[str, Any]: + """Attach to an existing lease and describe its driver clients and CLI tree. + + Returns a plain-serializable dict with drivers and cli_tree keys. + + Driver clients are synchronous facades over a blocking portal, so the + portal has to run its event loop in its own thread and be driven from + this one — introspection therefore happens outside the loop. + + :raises ValueError: if lease_name is empty + :raises jumpstarter.client.exceptions.LeaseError: if the lease has ended + or cannot be acquired + :raises jumpstarter.common.exceptions.ConnectionError: if the lease does + not exist or the controller is unreachable + """ + if not lease_name: + raise ValueError("lease_name must be a non-empty existing lease name") + + with start_blocking_portal() as portal: + with portal.wrap_async_context_manager(_connect_lease(config, lease_name, portal)) as client: + return describe_client(client) + + +async def describe_devices_async(config: ClientConfigV1Alpha1, lease_name: str) -> dict[str, Any]: + """Async wrapper around describe_devices, run in a worker thread. + + The driver clients cannot be driven from an event loop thread, so the + whole attach-and-introspect flow runs off-loop. + """ + if not lease_name: + raise ValueError("lease_name must be a non-empty existing lease name") + + return await to_thread.run_sync(describe_devices, config, lease_name) diff --git a/python/packages/jumpstarter/jumpstarter/client/introspect_test.py b/python/packages/jumpstarter/jumpstarter/client/introspect_test.py new file mode 100644 index 000000000..16a6c7577 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/client/introspect_test.py @@ -0,0 +1,464 @@ +import ast +import json +from contextlib import asynccontextmanager +from datetime import timedelta +from pathlib import Path +from types import MappingProxyType +from unittest.mock import MagicMock, patch + +import click +import pytest + +from jumpstarter.client.introspect import ( + _get_public_method_names, + describe_client, + describe_devices, + describe_devices_async, + get_driver_methods, + list_drivers, + walk_click_tree, +) + + +def test_description_helpers_are_public_package_exports(): + import jumpstarter.client as client + + for name, helper in ( + ("describe_client", describe_client), + ("describe_devices", describe_devices), + ("describe_devices_async", describe_devices_async), + ): + assert name in client.__all__ + assert getattr(client, name) is helper + + +class FakePowerClient: + children: dict = {} + + def on(self) -> None: + """Power on the device.""" + + def off(self) -> None: + """Power off the device.""" + + def cycle(self, wait: int = 2) -> None: + """Power cycle the device.""" + + def _private(self): + pass + + def call(self): + pass + + def check_exporter_status(self): + pass + + +class FakeSerialClient: + children: dict = {} + + def open(self): + """Open serial port.""" + + def pexpect(self): + """Create pexpect adapter.""" + + +class FakeCompositeClient: + description = "Test composite device" + + def __init__(self): + self.children = { + "power": FakePowerClient(), + "serial": FakeSerialClient(), + } + + def __getattr__(self, name): + try: + return self.children[name] + except KeyError: + raise AttributeError(name) from None + + def cli(self): + @click.group("j") + def base(): + """Fake composite device""" + + @base.command("on") + def on(): + """Power on.""" + + return base + + +class TestWalkClickTree: + def test_simple_command(self): + @click.command("hello") + @click.option("--name", help="Your name") + def hello(name): + """Say hello.""" + + result = walk_click_tree(hello) + assert result["name"] == "hello" + assert result["help"] == "Say hello." + assert len(result["params"]) == 1 + assert result["params"][0]["name"] == "name" + assert result["params"][0]["help"] == "Your name" + assert "subcommands" not in result + + def test_group_with_subcommands(self): + @click.group("root") + def root(): + """Root group.""" + + @root.command("sub1") + def sub1(): + """First sub.""" + + @root.command("sub2") + @click.option("--count", type=int, default=5) + def sub2(count): + """Second sub.""" + + result = walk_click_tree(root) + assert result["name"] == "root" + assert "subcommands" in result + assert "sub1" in result["subcommands"] + assert "sub2" in result["subcommands"] + assert result["subcommands"]["sub2"]["params"][0]["name"] == "count" + assert result["subcommands"]["sub2"]["params"][0]["default"] == 5 + + @pytest.mark.parametrize("parameter", [ + click.Argument(["help"], required=True), + click.Option(["--help"], required=True, help="An explicitly declared input"), + ]) + def test_declared_parameter_named_help_is_preserved(self, parameter): + command = click.Command("cmd", params=[parameter], add_help_option=False) + params = walk_click_tree(command)["params"] + assert len(params) == 1 + assert params[0]["name"] == "help" + assert params[0]["kind"] == parameter.param_type_name + assert params[0]["required"] is True + assert params[0]["opts"] == parameter.opts + + def test_automatic_help_is_not_a_declared_parameter(self): + command = click.Command("cmd") + with click.Context(command) as context: + assert command.get_help_option(context) is not None + assert walk_click_tree(command)["params"] == [] + + def test_hidden_params_excluded(self): + @click.command("cmd") + @click.option("--visible", help="shown") + @click.option("--secret", hidden=True) + def cmd(visible, secret): + pass + + result = walk_click_tree(cmd) + names = [p["name"] for p in result["params"]] + assert "visible" in names + assert "secret" not in names + + +class TestGetPublicMethodNames: + def test_filters_private_and_base_methods(self): + names = _get_public_method_names(FakePowerClient()) + assert "on" in names + assert "off" in names + assert "cycle" in names + assert "_private" not in names + assert "call" not in names + assert "check_exporter_status" not in names + + +class TestListDrivers: + def test_flat_tree(self): + client = FakeCompositeClient() + result = list_drivers(client) + paths = [d["path"] for d in result] + assert "client" in paths + assert "client.power" in paths + assert "client.serial" in paths + + def test_driver_path_field(self): + client = FakeCompositeClient() + result = list_drivers(client) + root = next(d for d in result if d["path"] == "client") + assert root["driver_path"] == [] + power = next(d for d in result if d["path"] == "client.power") + assert power["driver_path"] == ["power"] + + def test_methods_populated(self): + client = FakeCompositeClient() + result = list_drivers(client) + power = next(d for d in result if d["path"] == "client.power") + assert "on" in power["methods"] + assert "off" in power["methods"] + + +class TestGetDriverMethods: + def test_returns_method_details(self): + client = FakeCompositeClient() + result = get_driver_methods(client, ["power"]) + assert result["driver_path"] == ["power"] + method_names = [m["name"] for m in result["methods"]] + assert "on" in method_names + assert "off" in method_names + assert "cycle" in method_names + + def test_call_example_uses_dot_notation(self): + client = FakeCompositeClient() + result = get_driver_methods(client, ["power"]) + cycle = next(m for m in result["methods"] if m["name"] == "cycle") + assert "client.power.cycle(" in cycle["call_example"] + assert 'children["power"]' not in cycle["call_example"] + ast.parse(cycle["call_example"]) + + def test_root_driver_example_is_valid_python(self): + result = get_driver_methods(FakePowerClient(), []) + cycle = next(m for m in result["methods"] if m["name"] == "cycle") + assert "client.cycle(wait=...)" in cycle["call_example"] + assert "client.." not in cycle["call_example"] + ast.parse(cycle["call_example"]) + + def test_invalid_path_raises(self): + client = FakeCompositeClient() + with pytest.raises(KeyError, match="nonexistent"): + get_driver_methods(client, ["nonexistent"]) + + def test_docstrings_captured(self): + client = FakeCompositeClient() + result = get_driver_methods(client, ["power"]) + on_method = next(m for m in result["methods"] if m["name"] == "on") + assert on_method["docstring"] == "Power on the device." + + def test_parameters_captured(self): + client = FakeCompositeClient() + result = get_driver_methods(client, ["power"]) + cycle = next(m for m in result["methods"] if m["name"] == "cycle") + assert len(cycle["parameters"]) == 1 + assert cycle["parameters"][0]["name"] == "wait" + assert cycle["parameters"][0]["default"] == "2" + + +class TestDescribeClient: + def test_returns_drivers_and_cli_tree(self): + result = describe_client(FakeCompositeClient()) + paths = [d["path"] for d in result["drivers"]] + assert "client" in paths + assert "client.power" in paths + assert result["cli_tree"]["name"] == "j" + assert "on" in result["cli_tree"]["subcommands"] + + def test_cli_tree_none_without_cli(self): + result = describe_client(FakePowerClient()) + assert result["cli_tree"] is None + assert result["drivers"][0]["path"] == "client" + + def test_an_inherited_cli_still_counts(self): + """A driver client is free to inherit its CLI rather than define one. + + QemuFlasherClient is a real example: it has no cli of its own and takes + one from FlasherClientInterface. Looking only at type(client).__dict__ + would drop the CLI tree for every such driver. + """ + + class InheritsCli(FakeCompositeClient): + pass + + assert "cli" not in InheritsCli.__dict__ + result = describe_client(InheritsCli()) + assert result["cli_tree"]["name"] == "j" + + def test_a_broken_cli_costs_only_the_cli_tree(self): + """The driver listing is worth having even when cli() raises.""" + + class BrokenCli(FakeCompositeClient): + def cli(self): + raise RuntimeError("this driver's cli is broken") + + result = describe_client(BrokenCli()) + assert result["cli_tree"] is None + assert [d["path"] for d in result["drivers"]] != [] + + def test_composite_client_e2e(self): + from jumpstarter_driver_composite.driver import Composite + from jumpstarter_driver_power.driver import MockPower + + from jumpstarter.common.utils import serve + + with serve( + Composite( + children={ + "power": MockPower(), + "nested": Composite( + children={ + "power": MockPower(), + }, + ), + }, + ) + ) as client: + result = describe_client(client) + + paths = [d["path"] for d in result["drivers"]] + assert "client" in paths + assert "client.power" in paths + assert "client.nested.power" in paths + power = next(d for d in result["drivers"] if d["path"] == "client.power") + assert power["driver_path"] == ["power"] + assert "on" in power["methods"] + assert "power" in result["cli_tree"]["subcommands"] + assert "nested" in result["cli_tree"]["subcommands"] + json.dumps(result) + + @pytest.mark.anyio + async def test_stub_client_represented(self): + from contextlib import ExitStack + + from anyio.from_thread import BlockingPortal + + from jumpstarter.client.base import StubDriverClient + + async with BlockingPortal() as portal: + stub = StubDriverClient( + labels={"jumpstarter.dev/client": "missing_pkg.client.MissingClient"}, + stub=None, + portal=portal, + stack=ExitStack(), + ) + result = describe_client(stub) + + assert result["cli_tree"] is None + assert result["drivers"][0]["class"].endswith("StubDriverClient") + + +class TestDescribeDevices: + @pytest.fixture() + def mock_config(self): + fake_lease = MagicMock() + fake_lease.allow = [] + fake_lease.unsafe = True + + @asynccontextmanager + async def fake_serve_unix_async(): + yield "/tmp/fake.sock" + + fake_lease.serve_unix_async = fake_serve_unix_async + + @asynccontextmanager + async def fake_lease_async(*args, **kwargs): + yield fake_lease + + config = MagicMock() + config.lease_async = MagicMock(side_effect=fake_lease_async) + return config + + @pytest.fixture() + def mock_client_from_path(self): + @asynccontextmanager + async def fake_client_from_path(path, portal, stack, allow, unsafe): + yield FakeCompositeClient() + + with patch("jumpstarter.client.introspect.client_from_path", fake_client_from_path): + yield + + @pytest.mark.anyio + async def test_attaches_to_named_lease(self, mock_config, mock_client_from_path): + result = await describe_devices_async(mock_config, "existing-lease") + + kwargs = mock_config.lease_async.call_args.kwargs + assert kwargs["lease_name"] == "existing-lease" + assert kwargs["selector"] is None + assert kwargs["exporter_name"] is None + # lease_async requires a duration argument, but attaching by name never + # sends it to the controller. Keep the placeholder explicitly neutral. + assert kwargs["duration"] == timedelta(0) + paths = [d["path"] for d in result["drivers"]] + assert "client.power" in paths + assert result["cli_tree"]["name"] == "j" + json.dumps(result) + + @pytest.mark.anyio + async def test_empty_lease_name_raises(self, mock_config): + with pytest.raises(ValueError, match="non-empty"): + await describe_devices_async(mock_config, "") + mock_config.lease_async.assert_not_called() + + def test_blocking_wrapper(self, mock_config, mock_client_from_path): + result = describe_devices(mock_config, "existing-lease") + + kwargs = mock_config.lease_async.call_args.kwargs + assert kwargs["lease_name"] == "existing-lease" + assert result["cli_tree"]["name"] == "j" + + +class TestParamDescription: + @pytest.mark.parametrize("mapping_type", [dict, MappingProxyType]) + def test_mapping_defaults_remain_structured(self, mapping_type): + """Preserve nested mappings while coercing keys and non-JSON leaves.""" + default = mapping_type({ + "nested": mapping_type({1: (Path("image.bin"), None, True)}), + "items": [mapping_type({"count": 2})], + }) + command = click.Command("cmd", params=[click.Option(["--config"], default=default)]) + tree = walk_click_tree(command) + assert tree["params"][0]["default"] == { + "nested": {"1": ["image.bin", None, True]}, + "items": [{"count": 2}], + } + assert json.loads(json.dumps(tree)) == tree + + @pytest.mark.parametrize("default", [ + {1: "numeric", "1": "string"}, + {"1": "string", 1: "numeric"}, + {"nested": [MappingProxyType({1: "numeric", "1": "string"})]}, + ]) + def test_mapping_key_collisions_are_rejected(self, default): + """Never report a default with values lost to stringified key collisions.""" + command = click.Command("cmd", params=[click.Option(["--config"], default=default)]) + with pytest.raises(ValueError, match="Mapping keys collide after stringification: '1'"): + walk_click_tree(command) + + def test_describes_arguments_and_options(self): + @click.group() + def root(): + pass + + @root.command() + @click.argument("port", type=int) + @click.option("--address", help="Local address to bind") + @click.option("--verbose", is_flag=True) + @click.option("--mode", type=click.Choice(["fast", "slow"]), default="fast") + @click.option("--tag", multiple=True) + def forward(port, address, verbose, mode, tag): + """Forward a port.""" + + tree = walk_click_tree(root) + params = {p["name"]: p for p in tree["subcommands"]["forward"]["params"]} + + # A positional argument is spelled by name and an option by its flag, + # so a caller building a command line has to tell them apart. + assert params["port"]["kind"] == "argument" + assert params["port"]["required"] is True + assert params["port"]["type"] == "integer" + assert params["address"]["kind"] == "option" + assert params["address"]["opts"] == ["--address"] + assert params["address"]["help"] == "Local address to bind" + assert params["verbose"]["is_flag"] is True + assert params["mode"]["choices"] == ["fast", "slow"] + assert params["tag"]["multiple"] is True + + def test_type_name_is_readable(self): + @click.group() + def root(): + pass + + @root.command() + @click.argument("path", type=click.Path()) + def send(path): + pass + + params = walk_click_tree(root)["subcommands"]["send"]["params"] + # str(click.Path()) is an object repr, which is useless in a prompt. + assert params[0]["type"] == "path"