diff --git a/cli/commands/communication/ble.py b/cli/commands/communication/ble.py index eeb4619a..01e14bdf 100644 --- a/cli/commands/communication/ble.py +++ b/cli/commands/communication/ble.py @@ -14,7 +14,7 @@ import click from ...core.group_usage import LagerGroup -from ...core.net_helpers import resolve_box, post_box_command +from ...core.net_helpers import resolve_box, resolve_box_locked, post_box_command @click.group(name='ble', cls=LagerGroup) @@ -107,7 +107,7 @@ def scan(ctx, box, timeout, name_contains, name_exact, verbose): click.secho(f"Error: Timeout must be between {MIN_TIMEOUT} and {MAX_TIMEOUT} seconds, got {timeout}", fg='red', err=True) ctx.exit(1) - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'ble') click.secho(f"Scanning for BLE devices for {timeout} seconds...", fg='green') result = _post_ble( @@ -149,7 +149,7 @@ def scan(ctx, box, timeout, name_contains, name_exact, verbose): def _info_or_connect(ctx, box, address, connect_style: bool): """Shared body for the info and connect commands (same box action).""" _validate_ble_address(ctx, address) - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'ble') verb = "Connecting to" if connect_style else "Getting info for" click.secho(f"{verb} BLE device: {address}", fg='green') @@ -214,7 +214,7 @@ def disconnect(ctx, box, address): Disconnect from a BLE device """ _validate_ble_address(ctx, address) - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'ble') click.secho(f"Disconnecting from BLE device: {address}", fg='green') result = _post_ble(ctx, box_ip, 'disconnect', http_timeout=45.0, address=address) diff --git a/cli/commands/communication/blufi.py b/cli/commands/communication/blufi.py index be7dacfd..832dcf49 100644 --- a/cli/commands/communication/blufi.py +++ b/cli/commands/communication/blufi.py @@ -13,7 +13,7 @@ import click from ...core.group_usage import LagerGroup -from ...core.net_helpers import resolve_box, post_box_command +from ...core.net_helpers import resolve_box_locked, post_box_command @click.group(name='blufi', cls=LagerGroup) @@ -43,7 +43,7 @@ def _print_json(value: dict) -> None: @click.option('--name-contains', required=False, help='Filter devices to those whose name contains this string') def scan(ctx, box, timeout, name_contains): """Scan for BluFi-capable BLE devices""" - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'blufi') click.secho(f"Scanning for BluFi devices for {timeout} seconds...", fg='green') result = _post_blufi( @@ -79,7 +79,7 @@ def scan(ctx, box, timeout, name_contains): @click.argument('device_name', required=True) def connect(ctx, box, timeout, device_name): """Connect to a BluFi device and retrieve version and status""" - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'blufi') click.secho(f"Connecting to BluFi device: {device_name}", fg='green') result = _post_blufi( @@ -108,7 +108,7 @@ def connect(ctx, box, timeout, device_name): @click.argument('device_name', required=True) def provision(ctx, box, timeout, ssid, password, device_name): """Provision WiFi credentials to a BluFi device""" - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'blufi') click.secho(f"Provisioning '{ssid}' to BluFi device: {device_name}", fg='green') # Provisioning blocks box-side through connect + security negotiation + @@ -135,7 +135,7 @@ def provision(ctx, box, timeout, ssid, password, device_name): @click.argument('device_name', required=True) def wifi_scan(ctx, box, timeout, scan_timeout, device_name): """Scan for WiFi networks via a BluFi device""" - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'blufi') click.secho(f"Requesting WiFi scan via {device_name} (timeout={scan_timeout}s)...", fg='green') result = _post_blufi( @@ -169,7 +169,7 @@ def wifi_scan(ctx, box, timeout, scan_timeout, device_name): @click.argument('device_name', required=True) def status(ctx, box, timeout, device_name): """Get WiFi connection status from a BluFi device""" - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'blufi') result = _post_blufi( ctx, box_ip, 'status', @@ -193,7 +193,7 @@ def status(ctx, box, timeout, device_name): @click.argument('device_name', required=True) def version(ctx, box, timeout, device_name): """Get firmware version from a BluFi device""" - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'blufi') result = _post_blufi( ctx, box_ip, 'version', diff --git a/cli/commands/communication/i2c.py b/cli/commands/communication/i2c.py index 588c6122..69123650 100644 --- a/cli/commands/communication/i2c.py +++ b/cli/commands/communication/i2c.py @@ -23,7 +23,7 @@ from texttable import Texttable from ...core.net_group import NetGroupHelpMixin -from ...core.net_helpers import resolve_box, fetch_nets, post_net_command +from ...core.net_helpers import resolve_box, resolve_box_locked, fetch_nets, post_net_command from ...context import get_default_net from ...errors import net_not_specified_error @@ -56,12 +56,12 @@ def parse_args(self, ctx, args): def _resolve_box_with_name(ctx, box): """ - Resolve box parameter to IP address. + Resolve box parameter to IP address and acquire an ephemeral lock. Returns tuple of (ip_address, box_name). """ from ...box_storage import get_box_name_by_ip - resolved_ip = resolve_box(ctx, box) + resolved_ip = resolve_box_locked(ctx, box, 'i2c') if box and not box.replace('.', '').isdigit(): resolved_name = box diff --git a/cli/commands/communication/router.py b/cli/commands/communication/router.py index 6d91f540..94086812 100644 --- a/cli/commands/communication/router.py +++ b/cli/commands/communication/router.py @@ -13,7 +13,7 @@ import click from ...core.group_usage import LagerGroup -from ...core.net_helpers import resolve_box, post_net_command, NET_HTTP_PORT +from ...core.net_helpers import resolve_box, resolve_box_locked, post_net_command, NET_HTTP_PORT ROUTER_ROLE = "router" @@ -64,7 +64,7 @@ def add_net(ctx, name, address, username, password, instrument, use_ssl, box): """ import requests - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') net_data = { "name": name, @@ -117,7 +117,7 @@ def connect(ctx, netname, box): lager router connect router1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "connect", "netname": netname}) @@ -133,7 +133,7 @@ def interfaces(ctx, netname, box): lager router interfaces router1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "interfaces", "netname": netname}) @@ -149,7 +149,7 @@ def wireless_interfaces(ctx, netname, box): lager router wireless-interfaces router1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "wireless_interfaces", "netname": netname}) @@ -165,7 +165,7 @@ def wireless_clients(ctx, netname, box): lager router wireless-clients router1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "wireless_clients", "netname": netname}) @@ -181,7 +181,7 @@ def dhcp_leases(ctx, netname, box): lager router dhcp-leases router1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "dhcp_leases", "netname": netname}) @@ -197,7 +197,7 @@ def system_info(ctx, netname, box): lager router system-info router1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "system_info", "netname": netname}) @@ -218,7 +218,7 @@ def reboot(ctx, netname, yes, box): click.secho("Aborted.", fg="yellow") return - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "reboot", "netname": netname}) @@ -235,7 +235,7 @@ def enable_interface(ctx, netname, interface, box): lager router enable-interface router1 wlan1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "enable_interface", "netname": netname, "interface": interface}) @@ -253,7 +253,7 @@ def disable_interface(ctx, netname, interface, box): lager router disable-interface router1 wlan1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "disable_interface", "netname": netname, "interface": interface}) @@ -272,7 +272,7 @@ def block_internet(ctx, netname, box): lager router block-internet router1 --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "block_internet", "netname": netname}) @@ -300,7 +300,7 @@ def reset(ctx, netname, ssid, password, yes, box): click.secho("Aborted.", fg="yellow") return - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, { "action": "reset_to_defaults", "netname": netname, @@ -325,5 +325,5 @@ def run_cmd(ctx, netname, path, box): lager router run router1 /ip/address --box mybox """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'router') _run_router(ctx, box_ip, {"action": "run", "netname": netname, "path": path}) diff --git a/cli/commands/communication/spi.py b/cli/commands/communication/spi.py index 10fadacf..85e7a7f1 100644 --- a/cli/commands/communication/spi.py +++ b/cli/commands/communication/spi.py @@ -25,7 +25,7 @@ from texttable import Texttable from ...core.net_group import NetGroupHelpMixin -from ...core.net_helpers import resolve_box, fetch_nets, post_net_command +from ...core.net_helpers import resolve_box, resolve_box_locked, fetch_nets, post_net_command from ...context import get_default_net from ...errors import net_not_specified_error @@ -62,12 +62,12 @@ def parse_args(self, ctx, args): def _resolve_box_with_name(ctx, box): """ - Resolve box parameter to IP address. + Resolve box parameter to IP address and acquire an ephemeral lock. Returns tuple of (ip_address, box_name) where box_name is used for username lookup. """ from ...box_storage import get_box_name_by_ip - resolved_ip = resolve_box(ctx, box) + resolved_ip = resolve_box_locked(ctx, box, 'spi') if box and not box.replace('.', '').isdigit(): resolved_name = box diff --git a/cli/commands/communication/uart.py b/cli/commands/communication/uart.py index 664fd57f..7676502e 100644 --- a/cli/commands/communication/uart.py +++ b/cli/commands/communication/uart.py @@ -17,7 +17,7 @@ # Import consolidated helpers from cli.core.net_helpers from ...core.net_group import NetCommand, HiddenArgument -from ...core.net_helpers import resolve_box +from ...core.net_helpers import resolve_box, resolve_box_locked from ...context import get_default_net from ...errors import net_not_specified_error @@ -40,8 +40,8 @@ def _resolve_box_with_name(ctx, box): """ from ...box_storage import get_box_name_by_ip - # Use the shared resolve_box helper - resolved_ip = resolve_box(ctx, box) + # Use the shared resolve_box_locked helper (auto-acquires ephemeral lock) + resolved_ip = resolve_box_locked(ctx, box, 'uart') # Try to find box name for username lookup # If box was provided and is not an IP, it's the box name diff --git a/cli/commands/communication/usb.py b/cli/commands/communication/usb.py index 31bd2e66..0f8ffd02 100644 --- a/cli/commands/communication/usb.py +++ b/cli/commands/communication/usb.py @@ -22,6 +22,7 @@ from ...core.net_helpers import ( require_netname, resolve_box, + resolve_box_locked, list_nets_by_role, display_nets_table, validate_net_exists, @@ -129,7 +130,7 @@ def usb(ctx, netname, box): def _run_usb_action(ctx, box, action: str) -> None: """Shared body for the enable/disable/toggle subcommands.""" netname = require_netname(ctx, "usb") - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'usb') # Validate net exists before invoking remote command if _validate_usb_net(ctx, resolved_box, netname) is None: diff --git a/cli/commands/communication/wifi.py b/cli/commands/communication/wifi.py index a02edf06..26c90eba 100644 --- a/cli/commands/communication/wifi.py +++ b/cli/commands/communication/wifi.py @@ -14,7 +14,7 @@ # Import consolidated helpers from cli.core.net_helpers from ...core.group_usage import LagerGroup -from ...core.net_helpers import resolve_box, post_box_command +from ...core.net_helpers import resolve_box, resolve_box_locked, post_box_command # WiFi constraints MAX_SSID_LENGTH = 32 # IEEE 802.11 maximum SSID length @@ -117,7 +117,7 @@ def status(ctx, box): """ Get the current WiFi Status of the box """ - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'wifi') result = _post_wifi(ctx, box_ip, 'status') interfaces = (result.get('value') or {}).get('interfaces', []) @@ -145,7 +145,7 @@ def access_points(ctx, box, interface='wlan0'): # Validate interface name _validate_interface(ctx, interface) - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'wifi') click.secho(f"Scanning for WiFi networks on {interface}...", fg='green') result = _post_wifi(ctx, box_ip, 'scan', interface=interface) @@ -173,7 +173,7 @@ def connect(ctx, box, ssid, interface, password=''): _validate_password(ctx, password) _validate_interface(ctx, interface) - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'wifi') click.secho(f"Connecting to WiFi network: {ssid}", fg='green') result = _post_wifi(ctx, box_ip, 'connect', @@ -200,7 +200,7 @@ def delete_connection(ctx, box, yes, ssid): click.echo("Aborting") return - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'wifi') click.secho(f"Deleting WiFi connection: {ssid}", fg='green') result = _post_wifi(ctx, box_ip, 'delete', ssid=ssid, connection_name=ssid) diff --git a/cli/commands/development/arm.py b/cli/commands/development/arm.py index db7cafe2..4c8068d6 100644 --- a/cli/commands/development/arm.py +++ b/cli/commands/development/arm.py @@ -14,6 +14,7 @@ from ...core.net_group import NetGroup from ...core.net_helpers import ( resolve_box, + resolve_box_locked, list_nets_by_role, validate_net_exists, post_net_command, @@ -88,11 +89,16 @@ def _validate_arm_net(ctx, box, netname) -> bool: def _resolve_box_for_command(ctx, target_box): - """Resolve box from command-level --box option or group-level stored box.""" + """Resolve box from command-level --box option or group-level stored box. + Acquires an ephemeral lock for hardware interaction. + """ if target_box: - return resolve_box(ctx, target_box) + return resolve_box_locked(ctx, target_box, 'arm') # Fall back to box stored by the group command - return getattr(ctx.obj, "resolved_box", None) or get_default_box(ctx) + stored = getattr(ctx.obj, "resolved_box", None) + if stored: + return stored + return resolve_box_locked(ctx, None, 'arm') @click.group( @@ -125,7 +131,7 @@ class _Obj: pass # Only resolve box if box is provided at group level # Otherwise, let subcommands resolve it if box: - resolved = resolve_box(ctx, box) + resolved = resolve_box_locked(ctx, box, 'arm') setattr(ctx.obj, "resolved_box", resolved) else: # Don't set box - let subcommands handle it diff --git a/cli/commands/development/debug/commands.py b/cli/commands/development/debug/commands.py index b71229a4..ed625a72 100644 --- a/cli/commands/development/debug/commands.py +++ b/cli/commands/development/debug/commands.py @@ -104,6 +104,7 @@ def _resolve_box_with_username(ctx, box): """ Resolve box parameter to (IP, username) tuple. Handles both box names and direct IPs, looking up username from storage. + Also acquires an ephemeral lock for the duration of the debug session. Args: ctx: Click context @@ -112,10 +113,10 @@ def _resolve_box_with_username(ctx, box): Returns: Tuple of (ip_address, username) """ - from ....box_storage import resolve_and_validate_box + from ....core.net_helpers import resolve_box_locked - # Resolve and validate the box name/IP - box_ip = resolve_and_validate_box(ctx, box) + # Resolve, validate, and auto-lock the box + box_ip = resolve_box_locked(ctx, box, 'debug') # Determine box name for username lookup # If box was provided and is not an IP, it's the box name diff --git a/cli/commands/measurement/adc.py b/cli/commands/measurement/adc.py index 5a4c46c3..1d255fa3 100644 --- a/cli/commands/measurement/adc.py +++ b/cli/commands/measurement/adc.py @@ -17,6 +17,7 @@ from ...core.net_group import NetCommand from ...core.net_helpers import ( resolve_box, + resolve_box_locked, list_nets_by_role, display_nets_table, post_net_command, @@ -42,7 +43,7 @@ def adc(ctx, box, netname, as_json): if netname is None: netname = get_default_net(ctx, 'adc') - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'adc') # If still no netname, list available ADC nets if netname is None: diff --git a/cli/commands/measurement/dac.py b/cli/commands/measurement/dac.py index dfb2963a..5d41bc1f 100644 --- a/cli/commands/measurement/dac.py +++ b/cli/commands/measurement/dac.py @@ -17,6 +17,7 @@ from ...core.net_group import NetCommand from ...core.net_helpers import ( resolve_box, + resolve_box_locked, list_nets_by_role, display_nets_table, post_net_command, @@ -44,7 +45,7 @@ def dac(ctx, box, netname, voltage, as_json): if netname is None: netname = get_default_net(ctx, 'dac') - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'dac') # If still no netname, list available DAC nets if netname is None: diff --git a/cli/commands/measurement/energy.py b/cli/commands/measurement/energy.py index d23bf1dd..20c9bd55 100644 --- a/cli/commands/measurement/energy.py +++ b/cli/commands/measurement/energy.py @@ -13,6 +13,7 @@ from ...core.net_group import NetGroup from ...core.net_helpers import ( resolve_box, + resolve_box_locked, display_nets, fmt_si, post_net_command, @@ -56,7 +57,7 @@ def _print_stats(netname, result): def _run_energy(ctx, box, duration, netname, mode, as_json=False): """Shared implementation for energy and stats commands.""" - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'energy') if netname is None: display_nets(ctx, box_ip, None, ENERGY_ROLE, "energy analyzer") diff --git a/cli/commands/measurement/gpi.py b/cli/commands/measurement/gpi.py index 47b8aaf2..35fa6863 100644 --- a/cli/commands/measurement/gpi.py +++ b/cli/commands/measurement/gpi.py @@ -17,6 +17,7 @@ from ...core.net_group import NetCommand from ...core.net_helpers import ( resolve_box, + resolve_box_locked, list_nets_by_role, display_nets_table, post_net_command, @@ -61,7 +62,7 @@ def gpi(ctx, box, wait_for, timeout, scan_rate, scans_per_read, poll_interval, if netname is None: netname = get_default_net(ctx, 'gpio') - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'gpi') # If still no netname, list available GPIO nets if netname is None: diff --git a/cli/commands/measurement/gpo.py b/cli/commands/measurement/gpo.py index 3b1bf9b2..ee78ec0a 100644 --- a/cli/commands/measurement/gpo.py +++ b/cli/commands/measurement/gpo.py @@ -17,6 +17,7 @@ from ...core.net_group import NetCommand from ...core.net_helpers import ( resolve_box, + resolve_box_locked, list_nets_by_role, display_nets_table, post_net_command, @@ -51,7 +52,7 @@ def gpo(ctx, box, netname, level, hold, as_json): if netname is None: netname = get_default_net(ctx, 'gpio') - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'gpo') # If still no netname, list available GPIO nets if netname is None: diff --git a/cli/commands/measurement/logic.py b/cli/commands/measurement/logic.py index 26b1975d..aca3de41 100644 --- a/cli/commands/measurement/logic.py +++ b/cli/commands/measurement/logic.py @@ -12,6 +12,7 @@ from ...core.net_helpers import ( require_netname, resolve_box, + resolve_box_locked, validate_net, display_nets, run_backend, @@ -27,7 +28,7 @@ def _require_netname(ctx) -> str: def _resolve_box(ctx, box): - return resolve_box(ctx, box) + return resolve_box_locked(ctx, box, 'logic') def _validate_logic_net(ctx, box_ip: str, netname: str) -> bool: diff --git a/cli/commands/measurement/scope.py b/cli/commands/measurement/scope.py index 8d8d2c25..a4816ef4 100644 --- a/cli/commands/measurement/scope.py +++ b/cli/commands/measurement/scope.py @@ -15,6 +15,7 @@ from ...core.net_helpers import ( require_netname, resolve_box, + resolve_box_locked, run_net_py, validate_net, validate_net_exists, @@ -90,7 +91,7 @@ def _require_netname(ctx) -> str: def _resolve_box(ctx, box): - return resolve_box(ctx, box) + return resolve_box_locked(ctx, box, 'scope') def _validate_scope_net(ctx, box_ip: str, netname: str) -> dict | None: diff --git a/cli/commands/measurement/thermocouple.py b/cli/commands/measurement/thermocouple.py index e169da90..7c48b8bf 100644 --- a/cli/commands/measurement/thermocouple.py +++ b/cli/commands/measurement/thermocouple.py @@ -13,6 +13,7 @@ from ...core.net_group import NetCommand from ...core.net_helpers import ( resolve_box, + resolve_box_locked, display_nets, post_net_command, validate_net_exists, @@ -32,7 +33,7 @@ def thermocouple(ctx, box, netname, as_json): if netname is None: netname = get_default_net(ctx, 'thermocouple') - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'thermocouple') # If still no netname, list available thermocouple nets if netname is None: diff --git a/cli/commands/measurement/watt.py b/cli/commands/measurement/watt.py index ca293347..545e2e33 100644 --- a/cli/commands/measurement/watt.py +++ b/cli/commands/measurement/watt.py @@ -13,6 +13,7 @@ from ...core.net_group import NetGroup from ...core.net_helpers import ( resolve_box, + resolve_box_locked, display_nets, post_net_command, validate_net_exists, @@ -62,7 +63,7 @@ def _run_watt(ctx, box, netname, mode, duration, as_json): if box is None: box = getattr(ctx.obj, "watt_box", None) - box_ip = resolve_box(ctx, box) + box_ip = resolve_box_locked(ctx, box, 'watt') # No net resolved (and none configured) -> list available watt meter nets. if netname is None: diff --git a/cli/commands/power/battery.py b/cli/commands/power/battery.py index d6a6f888..ea70ea4e 100644 --- a/cli/commands/power/battery.py +++ b/cli/commands/power/battery.py @@ -23,6 +23,7 @@ from ...core.net_helpers import ( require_netname, resolve_box, + resolve_box_locked, validate_net, display_nets, echo_box_request_failure, @@ -174,7 +175,7 @@ def battery(ctx, box, netname): @click.option("--box", required=False, help="Lagerbox name or IP") def mode(ctx, box, mode_type): """Set (or read) battery simulation mode type""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_mode', netname=netname, mode_type=mode_type) @@ -184,7 +185,7 @@ def mode(ctx, box, mode_type): @click.option("--box", required=False, help="Lagerbox name or IP") def set_mode(ctx, box): """Initialize battery simulator mode""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_to_battery_mode', netname=netname) @@ -207,7 +208,7 @@ def soc(ctx, box, value): click.secho(f"Error: SOC must be between 0 and 100%, got {parsed_value}%", fg='red', err=True) ctx.exit(1) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_soc', netname=netname, value=parsed_value) @@ -230,7 +231,7 @@ def voc(ctx, box, value): click.secho(f"Error: VOC must be positive, got {parsed_value} V", fg='red', err=True) ctx.exit(1) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_voc', netname=netname, value=parsed_value) @@ -253,7 +254,7 @@ def batt_full(ctx, box, value): click.secho(f"Error: Battery full voltage must be positive, got {parsed_value} V", fg='red', err=True) ctx.exit(1) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_volt_full', netname=netname, value=parsed_value) @@ -276,7 +277,7 @@ def batt_empty(ctx, box, value): click.secho(f"Error: Battery empty voltage must be positive, got {parsed_value} V", fg='red', err=True) ctx.exit(1) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_volt_empty', netname=netname, value=parsed_value) @@ -299,7 +300,7 @@ def capacity(ctx, box, value): click.secho(f"Error: Capacity must be positive, got {parsed_value} Ah", fg='red', err=True) ctx.exit(1) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_capacity', netname=netname, value=parsed_value) @@ -328,7 +329,7 @@ def current_limit(ctx, box, value): click.secho(f"Error: Current limit must not exceed {MAX_CURRENT} A, got {parsed_value} A", fg='red', err=True) ctx.exit(1) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_current_limit', netname=netname, value=parsed_value) @@ -351,7 +352,7 @@ def ovp(ctx, box, value): click.secho(f"Error: OVP must be positive, got {parsed_value} V", fg='red', err=True) ctx.exit(1) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_ovp', netname=netname, value=parsed_value) @@ -374,7 +375,7 @@ def ocp(ctx, box, value): click.secho(f"Error: OCP must be positive, got {parsed_value} A", fg='red', err=True) ctx.exit(1) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_ocp', netname=netname, value=parsed_value) @@ -385,7 +386,7 @@ def ocp(ctx, box, value): @click.option("--box", required=False, help="Lagerbox name or IP") def model(ctx, box, partnumber): """Set (or read) battery model (18650, nimh, lead-acid, etc.)""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'set_model', netname=netname, partnumber=partnumber) @@ -395,7 +396,7 @@ def model(ctx, box, partnumber): @click.option("--box", required=False, help="Lagerbox name or IP") def models(ctx, box): """List battery models saved on the instrument""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'list_models', netname=netname) @@ -429,7 +430,7 @@ def model_create(ctx, box, slot, csv_path, force): "(an existing slot makes a good template)"], ) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") if not force: @@ -467,7 +468,7 @@ def model_export(ctx, box, slot, csv_path): accepts, for the export → edit → create round-trip. Read-only: exporting does not change the active model. """ - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") result = _battery_command_request( ctx, resolved_box, 'export_model', netname=netname, slot=slot) @@ -489,7 +490,7 @@ def model_export(ctx, box, slot, csv_path): @click.option("--box", required=False, help="Lagerbox name or IP") def state(ctx, box): """Get battery state (comprehensive status)""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'print_state', netname=netname) @@ -500,7 +501,7 @@ def state(ctx, box): @click.option('--yes', is_flag=True, help='Confirm the action without prompting.') def enable(ctx, box, yes): """Enable battery simulator output""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") if yes or click.confirm(f"Enable Net?", default=False): @@ -518,7 +519,7 @@ def enable(ctx, box, yes): @click.option('--yes', is_flag=True, help='Confirm the action without prompting.') def disable(ctx, box, yes): """Disable battery simulator output""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") if yes or click.confirm(f"Disable Net?", default=True): @@ -537,7 +538,7 @@ def disable(ctx, box, yes): @click.option("--box", required=False, help="Lagerbox name or IP") def clear_both(ctx, box): """Clear protection trip conditions (OVP/OCP)""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'clear', netname=netname) @@ -547,7 +548,7 @@ def clear_both(ctx, box): @click.option("--box", required=False, help="Lagerbox name or IP") def clear_ovp(ctx, box): """Clear OVP trip condition""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'clear_ovp', netname=netname) @@ -557,7 +558,7 @@ def clear_ovp(ctx, box): @click.option("--box", required=False, help="Lagerbox name or IP") def clear_ocp(ctx, box): """Clear OCP trip condition""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") _run_backend(ctx, resolved_box, 'clear_ocp', netname=netname) @@ -567,7 +568,7 @@ def clear_ocp(ctx, box): @click.option("--box", required=False, help="Lagerbox name or IP") def tui(ctx, box): """Launch interactive battery control TUI""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'battery') netname = require_netname(ctx, "battery") if not validate_net(ctx, resolved_box, netname, BATTERY_ROLE): diff --git a/cli/commands/power/eload.py b/cli/commands/power/eload.py index 8c749585..ba10e93d 100644 --- a/cli/commands/power/eload.py +++ b/cli/commands/power/eload.py @@ -23,6 +23,7 @@ from ...core.net_helpers import ( require_netname, resolve_box, + resolve_box_locked, display_nets, post_net_command, NET_ROLES, @@ -188,7 +189,7 @@ def _mode_options(func): def cc(ctx, value, box, as_json): """Set (or read) constant current mode in amps (A)""" _validate_eload_value(ctx, "cc", value) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'eload') netname = require_netname(ctx, "eload") _run_eload(ctx, resolved_box, netname, "cc", value, as_json) @@ -200,7 +201,7 @@ def cc(ctx, value, box, as_json): def cv(ctx, value, box, as_json): """Set (or read) constant voltage mode in volts (V)""" _validate_eload_value(ctx, "cv", value) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'eload') netname = require_netname(ctx, "eload") _run_eload(ctx, resolved_box, netname, "cv", value, as_json) @@ -212,7 +213,7 @@ def cv(ctx, value, box, as_json): def cr(ctx, value, box, as_json): """Set (or read) constant resistance mode in ohms""" _validate_eload_value(ctx, "cr", value) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'eload') netname = require_netname(ctx, "eload") _run_eload(ctx, resolved_box, netname, "cr", value, as_json) @@ -224,7 +225,7 @@ def cr(ctx, value, box, as_json): def cp(ctx, value, box, as_json): """Set (or read) constant power mode in watts (W)""" _validate_eload_value(ctx, "cp", value) - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'eload') netname = require_netname(ctx, "eload") _run_eload(ctx, resolved_box, netname, "cp", value, as_json) @@ -236,7 +237,7 @@ def cp(ctx, value, box, as_json): @click.pass_context def state(ctx, box, as_json): """Display electronic load state""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'eload') netname = require_netname(ctx, "eload") result = post_net_command(ctx, resolved_box, netname, "state", role="eload", quiet=True) diff --git a/cli/commands/power/solar.py b/cli/commands/power/solar.py index 4cedaa60..d3f9dd1f 100644 --- a/cli/commands/power/solar.py +++ b/cli/commands/power/solar.py @@ -18,12 +18,12 @@ # Import consolidated helpers from cli.core.net_helpers from ...core.net_helpers import ( resolve_box, + resolve_box_locked, display_nets, post_net_command, NET_ROLES, ) from ...context import get_default_net -from ...box_storage import resolve_and_validate_box SOLAR_ROLE = NET_ROLES["solar"] # "solar" @@ -52,8 +52,8 @@ def _run_backend(ctx: click.Context, box: str | None, action: str, **params) -> if box is None: box = ctx.obj.box - # Resolve and validate the box name - resolved_box = resolve_and_validate_box(ctx, box) + # Resolve and validate the box name, acquiring an ephemeral lock + resolved_box = resolve_box_locked(ctx, box, 'solar') # Every solar action re-asserts PV mode on the instrument (enable() with # settle retries), so widen the HTTP budget past the box-side proxy diff --git a/cli/commands/power/supply.py b/cli/commands/power/supply.py index 48083d34..2bda8320 100644 --- a/cli/commands/power/supply.py +++ b/cli/commands/power/supply.py @@ -26,6 +26,7 @@ from ...core.net_helpers import ( require_netname, resolve_box, + resolve_box_locked, validate_net, validate_net_exists, display_nets, @@ -173,7 +174,7 @@ def supply(ctx, box, netname): @click.option("--yes", is_flag=True, default=False, help="Confirm the action without prompting") def voltage(ctx, box, value, ocp, ovp, yes): """Set (or read) voltage in volts (V)""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") # Validate net exists BEFORE prompting for confirmation @@ -210,7 +211,7 @@ def voltage(ctx, box, value, ocp, ovp, yes): @click.option("--yes", is_flag=True, default=False, help="Confirm the action without prompting") def current(ctx, box, value, ocp, ovp, yes): """Set (or read) current in amps (A)""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") # Validate net exists BEFORE prompting for confirmation @@ -247,7 +248,7 @@ def disable(ctx, box, yes): if not yes and not click.confirm("Disable Net?", default=False): click.echo("Aborting") return - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") _run_backend(ctx, resolved_box, action="disable", netname=netname) @@ -261,7 +262,7 @@ def enable(ctx, box, yes): if not yes and not click.confirm("Enable Net?", default=False): click.echo("Aborting") return - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") _run_backend(ctx, resolved_box, action="enable", netname=netname) @@ -271,7 +272,7 @@ def enable(ctx, box, yes): @click.option("--box", required=False, help="Lagerbox name or IP") def state(ctx, box): """Read power state""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") _run_backend(ctx, resolved_box, action="state", netname=netname) @@ -283,7 +284,7 @@ def set_mode(ctx, box): """ Set power supply mode """ - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") _run_backend(ctx, resolved_box, action="set_mode", netname=netname) @@ -293,7 +294,7 @@ def set_mode(ctx, box): @click.option("--box", required=False, help="Lagerbox name or IP") def clear_ovp(ctx, box): """Clear over-voltage protection trip condition""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") _run_backend(ctx, resolved_box, action="clear_ovp", netname=netname) @@ -303,7 +304,7 @@ def clear_ovp(ctx, box): @click.option("--box", required=False, help="Lagerbox name or IP") def clear_ocp(ctx, box): """Clear over-current protection trip condition""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") _run_backend(ctx, resolved_box, action="clear_ocp", netname=netname) @@ -313,7 +314,7 @@ def clear_ocp(ctx, box): @click.option("--box", required=False, help="Lagerbox name or IP") def tui(ctx, box): """Launch interactive supply control TUI""" - resolved_box = resolve_box(ctx, box) + resolved_box = resolve_box_locked(ctx, box, 'supply') netname = require_netname(ctx, "supply") if not validate_net(ctx, box, netname, SUPPLY_ROLE): diff --git a/cli/commands/utility/webcam.py b/cli/commands/utility/webcam.py index 87c4ea77..0778e818 100644 --- a/cli/commands/utility/webcam.py +++ b/cli/commands/utility/webcam.py @@ -25,7 +25,7 @@ def _get_box_ip_address(ctx: click.Context, box: str = None) -> str: """ - Get the box IP address from various sources. + Get the box IP address from various sources, acquiring an ephemeral lock. Priority: 1. Explicit --box option (check local boxes first) @@ -34,9 +34,9 @@ def _get_box_ip_address(ctx: click.Context, box: str = None) -> str: Returns: IP address string """ - from ...box_storage import resolve_and_validate_box + from ...core.net_helpers import resolve_box_locked - return resolve_and_validate_box(ctx, box) + return resolve_box_locked(ctx, box, 'webcam') def _resolve_box(ctx, box): diff --git a/cli/core/net_helpers.py b/cli/core/net_helpers.py index 65223857..b83d9564 100644 --- a/cli/core/net_helpers.py +++ b/cli/core/net_helpers.py @@ -70,6 +70,57 @@ def resolve_box(ctx: click.Context, box: str | None) -> str: return resolve_and_validate_box(ctx, box) +def resolve_box_locked( + ctx: click.Context, + box: str | None, + command_name: str, +) -> str: + """Resolve box IP *and* acquire an ephemeral auto-lock for this command. + + Behaves identically to :func:`resolve_box` but additionally acquires a + short-lived (TTL-based, heartbeat-refreshed) lock for the duration of the + CLI process. The lock is released on normal exit, exception, or signal. + + If the box is already locked by the same user (e.g. via ``lager boxes + lock``), the pre-existing lock is preserved and no release is registered. + + Args: + ctx: Click context object. + box: Box name or IP, or None for the default box. + command_name: Human-readable command name for heartbeat warnings + (e.g. ``'gpi'``, ``'debug flash'``). + + Returns: + Resolved and validated box IP address. + """ + from ..box_storage import ( + resolve_and_validate_box_with_name, + auto_lock_acquire_for_command, + ) + import os + + ip, box_name = resolve_and_validate_box_with_name(ctx, box) + + if not os.getenv('LAGER_AUTO_LOCK_DISABLE'): + release = auto_lock_acquire_for_command( + ip, box_name or ip, command_name, + ) + # Stash the release callable on the Click context so downstream code + # (or test fixtures) can access it if needed. The atexit hook inside + # auto_lock_acquire_for_command handles process-death paths. + if not hasattr(ctx, 'obj') or ctx.obj is None: + ctx.ensure_object(dict) + if isinstance(ctx.obj, dict): + ctx.obj.setdefault('_lock_releases', []).append(release) + else: + existing = getattr(ctx.obj, '_lock_releases', None) + if existing is None: + ctx.obj._lock_releases = [] + ctx.obj._lock_releases.append(release) + + return ip + + # ============================================================================= # Netname Handling # ============================================================================= diff --git a/docs/source/reference/cli/locking.mdx b/docs/source/reference/cli/locking.mdx index 93396ea0..11baea8b 100644 --- a/docs/source/reference/cli/locking.mdx +++ b/docs/source/reference/cli/locking.mdx @@ -24,20 +24,24 @@ TTL reap. ### Which commands auto-lock -| Command | Lock window | Why | -|------------------------|----------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------| -| `lager python` | Full test run (acquire → heartbeat → release) | Canonical test runner. | -| `lager install` | The `setup_and_deploy_box.sh` step (the part that restarts the container) | Container restart mid-test would kill the test outright. | -| `lager uninstall` | Container teardown, image wipe, `~/box` and `/etc/lager` removal | Same — destructive on-box mutation. | -| `lager update` | Container stop → image rebuild → restart → health check | Container restart is the test-clobbering action. Read-only probe / fetch are deliberately outside the lock. | -| `lager install-wheel` | The `pip install` invocation inside the container | `pip install` mutates the container's Python environment; a concurrent test could race on imports. | +| Category | Commands | Lock window | Why | +|----------|----------|-------------|-----| +| Test runner | `lager python` | Full test run (acquire → heartbeat → release) | Canonical test runner. | +| Measurement | `gpi`, `gpo`, `adc`, `dac`, `thermocouple`, `watt`, `energy`, `scope`, `logic` | Single command invocation | Hardware I/O — interleaved access corrupts readings or pin state. | +| Communication | `spi`, `i2c`, `uart`, `wifi`, `ble`, `blufi`, `usb`, `router` | Single command invocation | Bus/protocol transactions must not interleave across users. | +| Power | `supply`, `battery`, `eload`, `solar` | Single command invocation (subcommands only, not listing) | Concurrent voltage/current changes are dangerous. | +| Development | `debug` (flash/connect/erase/reset/etc.), `arm`, `webcam` | Single command invocation | Flashing or debug sessions must not collide. | +| Admin | `install`, `uninstall`, `update`, `install-wheel` | Destructive section only | Container restart/mutation mid-test would kill the test. | Read-only commands (`lager hello`, `lager boxes list`, `lager boxes lock` / -`unlock` itself, status / dry-run paths, etc.) do **not** acquire the -auto-lock. Note: this is intentionally narrower than the v0.12–0.13.3 -behavior, which slapped a `--force-command`-overridable lock on every -single CLI command. See *Backward compatibility* below for the -v0.13.4 history. +`unlock` itself, net-listing paths like `lager supply --box X` with no +subcommand, status / dry-run paths, etc.) do **not** acquire the auto-lock. + +Unlike the v0.12–0.13.3 design (which used a shared decorator on every +command and was reverted in v0.13.4), this implementation uses the same +TTL + heartbeat + atexit infrastructure as `lager python`, avoiding the +three corner cases that motivated the revert. See *Backward +compatibility* below for the full history. The lock identity is **CI-aware** so concurrent test runs in CI mutually exclude correctly. Holder formats: @@ -232,11 +236,11 @@ v0.13.0 added an ephemeral "command-in-progress" lock that fired on `--force-command` flag. v0.13.4 removed it because three corner cases were unfixable in that design: -| v0.13.4 corner case | How this PR avoids it | +| v0.13.4 corner case | How the current design avoids it | |--------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| *"Supply commands never released the lock"* | The auto-lock is only attached to **5 commands** (`python`, `install`, `uninstall`, `update`, `install-wheel`), not every CLI surface. Supply commands etc. don't touch the lock — no decorator-on-everything to leak from. | -| *"Long-running commands blocked all other commands on the same box"* | Only those 5 commands check the lock; status / list / read-only paths are unaffected. For genuine concurrent test runs, dev gets fail-fast in <5s and CI gets a queue (default 60s, configurable). That's the *desired* policy. | -| *"Detached processes left stale locks"* | `--detach` is **opt-in** for a long-lived hold (`ttl_seconds: null` is intentional). Non-detached runs have heartbeat + TTL reap, so an abnormal CLI exit self-recovers in ≤ TTL + grace. | +| *"Supply commands never released the lock"* | Every auto-lock uses `auto_lock_acquire_for_command` which registers an `atexit` handler + heartbeat thread. The lock is released on normal exit, exception, SIGINT, or (worst case) reaped by the server TTL after SIGKILL. The v0.13 bug was a decorator that swallowed exceptions before release ran — the atexit safety net makes that impossible. | +| *"Long-running commands blocked all other commands on the same box"* | Read-only / listing paths (`lager supply --box X` with no subcommand, `lager gpi --box X` with no netname, `lager boxes`, `lager hello`) use plain `resolve_box()` which only checks the lock passively. Only hardware-*interacting* subcommands acquire. For concurrent hardware use, dev gets fail-fast in <1s and CI gets a queue (default 30min, configurable via `LAGER_LOCK_WAIT`). | +| *"Detached processes left stale locks"* | All ephemeral locks have `ttl_seconds=1800` + heartbeat every 60s. If the CLI dies, the box reaps the lock within one TTL. `--detach` (on `lager python` only) is an opt-in eternal hold (`ttl_seconds: null`) that must be released manually — this is intentional and documented. | `--force-command` is **gone**. Collision policy is structured (fail-fast in dev, queue in CI) and the existing `lager boxes lock --force` is the diff --git a/test/unit/cli/test_resolve_box_locked.py b/test/unit/cli/test_resolve_box_locked.py new file mode 100644 index 00000000..8c763876 --- /dev/null +++ b/test/unit/cli/test_resolve_box_locked.py @@ -0,0 +1,143 @@ +# Copyright 2024-2026 Lager Data +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for cli.core.net_helpers.resolve_box_locked. + +Verifies that hardware-interacting CLI commands: +- Acquire an ephemeral lock on box resolution. +- Release the lock on exit (via the atexit registered inside + auto_lock_acquire_for_command). +- Pass through cleanly when LAGER_AUTO_LOCK_DISABLE is set. +- Respect pre-existing user locks (do NOT release them). +""" + +from __future__ import annotations + +import os +from unittest import mock + +import click +import pytest + + +@pytest.fixture(autouse=True) +def _clear_env(monkeypatch): + for key in ('LAGER_AUTO_LOCK_DISABLE', 'LAGER_LOCK_HOLDER', 'LAGER_USER'): + monkeypatch.delenv(key, raising=False) + + +class _FakeResp: + def __init__(self, status_code, json_body=None): + self.status_code = status_code + self._json = json_body or {} + self.text = '' + self.headers = {} + + def json(self): + return self._json + + +def _make_ctx(): + """Build a minimal Click context with obj namespace.""" + ctx = click.Context(click.Command('test')) + ctx.obj = type('Obj', (), {})() + return ctx + + +class TestResolveBoxLocked: + """Test resolve_box_locked acquires a lock.""" + + @mock.patch('requests.post') + @mock.patch('requests.get') + @mock.patch('cli.box_storage.resolve_and_validate_box_with_name') + def test_acquires_lock_on_resolve(self, mock_resolve, mock_get, mock_post): + """Calling resolve_box_locked should POST /lock on port 9000.""" + mock_resolve.return_value = ('10.0.0.1', 'test-box') + # GET /lock returns unlocked + mock_get.return_value = _FakeResp(200, {'locked': False}) + # POST /lock returns acquired + mock_post.return_value = _FakeResp(200, { + 'locked': True, + 'user': 'alice', + 'holder_type': 'ephemeral', + 'locked_at': '2026-01-01T00:00:00Z', + 'last_heartbeat': '2026-01-01T00:00:00Z', + 'ttl_seconds': 1800, + 'previous_user': None, + }) + + from cli.core.net_helpers import resolve_box_locked + + ctx = _make_ctx() + with mock.patch('cli.box_storage.get_lager_user', return_value='alice'): + ip = resolve_box_locked(ctx, 'test-box', 'gpi') + + assert ip == '10.0.0.1' + # Verify POST /lock was called + assert mock_post.call_count >= 1 + lock_call = mock_post.call_args_list[-1] + assert '/lock' in lock_call.args[0] or '/lock' in str(lock_call) + + @mock.patch('cli.box_storage.resolve_and_validate_box_with_name') + def test_skips_lock_when_disabled(self, mock_resolve, monkeypatch): + """LAGER_AUTO_LOCK_DISABLE skips the lock entirely.""" + monkeypatch.setenv('LAGER_AUTO_LOCK_DISABLE', '1') + mock_resolve.return_value = ('10.0.0.2', 'lab-box') + + from cli.core.net_helpers import resolve_box_locked + + ctx = _make_ctx() + ip = resolve_box_locked(ctx, 'lab-box', 'spi') + assert ip == '10.0.0.2' + + @mock.patch('requests.post') + @mock.patch('requests.get') + @mock.patch('cli.box_storage.resolve_and_validate_box_with_name') + def test_stashes_release_on_ctx(self, mock_resolve, mock_get, mock_post): + """The release callable is stashed on ctx.obj._lock_releases.""" + mock_resolve.return_value = ('10.0.0.3', 'my-box') + mock_get.return_value = _FakeResp(200, {'locked': False}) + mock_post.return_value = _FakeResp(200, { + 'locked': True, + 'user': 'bob', + 'holder_type': 'ephemeral', + 'locked_at': '2026-01-01T00:00:00Z', + 'last_heartbeat': '2026-01-01T00:00:00Z', + 'ttl_seconds': 1800, + 'previous_user': None, + }) + + from cli.core.net_helpers import resolve_box_locked + + ctx = _make_ctx() + with mock.patch('cli.box_storage.get_lager_user', return_value='bob'): + resolve_box_locked(ctx, 'my-box', 'debug') + + releases = getattr(ctx.obj, '_lock_releases', []) + assert len(releases) == 1 + assert callable(releases[0]) + + @mock.patch('requests.get') + @mock.patch('cli.box_storage.resolve_and_validate_box_with_name') + def test_already_ours_no_release(self, mock_resolve, mock_get): + """If the lock is already ours, release.state == 'already_ours'.""" + mock_resolve.return_value = ('10.0.0.4', 'shared-box') + # GET /lock shows already locked by us + mock_get.return_value = _FakeResp(200, { + 'locked': True, + 'user': 'carol', + 'holder_type': 'user', + 'ttl_seconds': None, + }) + + from cli.core.net_helpers import resolve_box_locked + + ctx = _make_ctx() + with mock.patch('cli.box_storage.get_lager_user', return_value='carol'): + ip = resolve_box_locked(ctx, 'shared-box', 'uart') + + assert ip == '10.0.0.4' + releases = getattr(ctx.obj, '_lock_releases', []) + # Release is stashed but its .state should be 'already_ours' + if releases: + assert releases[0].state == 'already_ours'