From d1c1309106897c899dd011ae1a1f00a0b6ec39fa Mon Sep 17 00:00:00 2001 From: shaggy Date: Fri, 31 Oct 2025 16:16:16 -0500 Subject: [PATCH 01/19] cli/__init__.py: catch a syntax error when re-compiling the target file --- pybricksdev/cli/__init__.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index 8f4a1e4..bc0c734 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -359,6 +359,10 @@ async def reconnect_hub(): case _: return + except SyntaxError as e: + print("\nA syntax error occurred when parsing your program:") + print(e, "\n") + except HubPowerButtonPressedError: # This means the user pressed the button on the hub to re-start the # current program, so the menu was canceled and we are now printing From 8b5eb9c183be1bc2be703bb8bcc8f77bdf552cb7 Mon Sep 17 00:00:00 2001 From: shaggy Date: Fri, 31 Oct 2025 16:25:18 -0500 Subject: [PATCH 02/19] cli/__init__.py: slightly modify the error message warning --- pybricksdev/cli/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index bc0c734..6beedf8 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -360,7 +360,7 @@ async def reconnect_hub(): return except SyntaxError as e: - print("\nA syntax error occurred when parsing your program:") + print("\nA syntax error occurred while parsing your program:") print(e, "\n") except HubPowerButtonPressedError: From e95fc0e313ca4b8002a5ab861287e606503d495a Mon Sep 17 00:00:00 2001 From: shaggy Date: Sat, 1 Nov 2025 10:47:12 -0500 Subject: [PATCH 03/19] cli/__init__.py: refactor to allow the catching of syntax errors on the first download to the hub --- pybricksdev/cli/__init__.py | 139 +++++++++++++++++++----------------- 1 file changed, 75 insertions(+), 64 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index 6beedf8..42c2ae6 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -25,6 +25,7 @@ from pybricksdev.connections.pybricks import ( HubDisconnectError, HubPowerButtonPressedError, + PybricksHub, ) PROG_NAME = ( @@ -184,70 +185,7 @@ def add_parser(self, subparsers: argparse._SubParsersAction): async def run(self, args: argparse.Namespace): - # Pick the right connection - if args.conntype == "ble": - from pybricksdev.ble import find_device as find_ble - from pybricksdev.connections.pybricks import PybricksHubBLE - - # It is a Pybricks Hub with BLE. Device name or address is given. - print(f"Searching for {args.name or 'any hub with Pybricks service'}...") - device_or_address = await find_ble(args.name) - hub = PybricksHubBLE(device_or_address) - elif args.conntype == "usb": - from usb.core import find as find_usb - - from pybricksdev.connections.pybricks import PybricksHubUSB - from pybricksdev.usb import ( - EV3_USB_PID, - LEGO_USB_VID, - MINDSTORMS_INVENTOR_USB_PID, - NXT_USB_PID, - SPIKE_ESSENTIAL_USB_PID, - SPIKE_PRIME_USB_PID, - ) - - def is_pybricks_usb(dev): - return ( - (dev.idVendor == LEGO_USB_VID) - and ( - dev.idProduct - in [ - NXT_USB_PID, - EV3_USB_PID, - SPIKE_PRIME_USB_PID, - SPIKE_ESSENTIAL_USB_PID, - MINDSTORMS_INVENTOR_USB_PID, - ] - ) - and dev.product.endswith("Pybricks") - ) - - device_or_address = find_usb(custom_match=is_pybricks_usb) - - if device_or_address is None: - print("Pybricks Hub not found.", file=sys.stderr) - exit(1) - - hub = PybricksHubUSB(device_or_address) - else: - raise ValueError(f"Unknown connection type: {args.conntype}") - - # Connect to the address and run the script - await hub.connect() - try: - with _get_script_path(args.file) as script_path: - if args.start: - await hub.run(script_path, args.wait or args.stay_connected) - else: - if args.stay_connected: - # if the user later starts the program by pressing the button on the hub, - # we still want the hub stdout to print to Python's stdout - hub.print_output = True - hub._enable_line_handler = True - await hub.download(script_path) - - if not args.stay_connected: - return + async def stay_connected_menu(hub: PybricksHub): class ResponseOptions(IntEnum): RECOMPILE_RUN = 0 @@ -379,6 +317,79 @@ async def reconnect_hub(): await asyncio.sleep(0.3) hub = await reconnect_hub() + # Pick the right connection + if args.conntype == "ble": + from pybricksdev.ble import find_device as find_ble + from pybricksdev.connections.pybricks import PybricksHubBLE + + # It is a Pybricks Hub with BLE. Device name or address is given. + print(f"Searching for {args.name or 'any hub with Pybricks service'}...") + device_or_address = await find_ble(args.name) + hub = PybricksHubBLE(device_or_address) + elif args.conntype == "usb": + from usb.core import find as find_usb + + from pybricksdev.connections.pybricks import PybricksHubUSB + from pybricksdev.usb import ( + EV3_USB_PID, + LEGO_USB_VID, + MINDSTORMS_INVENTOR_USB_PID, + NXT_USB_PID, + SPIKE_ESSENTIAL_USB_PID, + SPIKE_PRIME_USB_PID, + ) + + def is_pybricks_usb(dev): + return ( + (dev.idVendor == LEGO_USB_VID) + and ( + dev.idProduct + in [ + NXT_USB_PID, + EV3_USB_PID, + SPIKE_PRIME_USB_PID, + SPIKE_ESSENTIAL_USB_PID, + MINDSTORMS_INVENTOR_USB_PID, + ] + ) + and dev.product.endswith("Pybricks") + ) + + device_or_address = find_usb(custom_match=is_pybricks_usb) + + if device_or_address is None: + print("Pybricks Hub not found.", file=sys.stderr) + exit(1) + + hub = PybricksHubUSB(device_or_address) + else: + raise ValueError(f"Unknown connection type: {args.conntype}") + + # Connect to the address and run the script + await hub.connect() + try: + with _get_script_path(args.file) as script_path: + if args.start: + await hub.run(script_path, args.wait or args.stay_connected) + else: + if args.stay_connected: + # if the user later starts the program by pressing the button on the hub, + # we still want the hub stdout to print to Python's stdout + hub.print_output = True + hub._enable_line_handler = True + await hub.download(script_path) + + if not args.stay_connected: + return + + await stay_connected_menu(hub) + + except SyntaxError as e: + print("\nA syntax error occurred while parsing your program:") + print(e, "\n") + if args.stay_connected: + await stay_connected_menu(hub) + finally: await hub.disconnect() From c3f3c2016d68d57bc4b9f5968dd2050096faf293 Mon Sep 17 00:00:00 2001 From: shaggy Date: Sat, 1 Nov 2025 10:53:30 -0500 Subject: [PATCH 04/19] cli/__init__.py: clean up an if statement --- pybricksdev/cli/__init__.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index 42c2ae6..7730ce0 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -379,10 +379,8 @@ def is_pybricks_usb(dev): hub._enable_line_handler = True await hub.download(script_path) - if not args.stay_connected: - return - - await stay_connected_menu(hub) + if args.stay_connected: + await stay_connected_menu(hub) except SyntaxError as e: print("\nA syntax error occurred while parsing your program:") From 709e6f11d4b6d71a84af0a0b80f69c89e25c9a00 Mon Sep 17 00:00:00 2001 From: shaggy Date: Tue, 25 Nov 2025 22:48:36 -0600 Subject: [PATCH 05/19] CHANGELOG.md: add an entry for the new changes to the run command --- CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ace572..d7fc60e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed +- The `run` command now catches syntax errors in the input file. ([pybricksdev#126]) + +[pybricksdev#126]: https://github.com/pybricks/pybricksdev/pull/126 + ## [2.3.0] - 2025-10-31 ### Added From aedee305e9527fa4641ec9a61235ffb1691c696f Mon Sep 17 00:00:00 2001 From: shaggy Date: Fri, 28 Nov 2025 13:23:09 -0600 Subject: [PATCH 06/19] cli/__init__.py: avoid using newline escape characters --- pybricksdev/cli/__init__.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index 7730ce0..bfb6be6 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -298,8 +298,10 @@ async def reconnect_hub(): return except SyntaxError as e: - print("\nA syntax error occurred while parsing your program:") - print(e, "\n") + print() + print("A syntax error occurred while parsing your program:") + print(e) + print() except HubPowerButtonPressedError: # This means the user pressed the button on the hub to re-start the From 7a1ca00cbcce7434fc3f7e6126bdfd76fd80b1f3 Mon Sep 17 00:00:00 2001 From: shaggy Date: Fri, 28 Nov 2025 13:44:35 -0600 Subject: [PATCH 07/19] cli/__init__.py: move stay_connected_menu to separate function --- pybricksdev/cli/__init__.py | 274 ++++++++++++++++++++---------------- 1 file changed, 154 insertions(+), 120 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index bfb6be6..b0eda49 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -183,142 +183,174 @@ def add_parser(self, subparsers: argparse._SubParsersAction): default=False, ) - async def run(self, args: argparse.Namespace): - - async def stay_connected_menu(hub: PybricksHub): + async def stay_connected_menu(self, hub: PybricksHub, args: argparse.Namespace): - class ResponseOptions(IntEnum): - RECOMPILE_RUN = 0 - RECOMPILE_DOWNLOAD = 1 - RUN_STORED = 2 - CHANGE_TARGET_FILE = 3 - EXIT = 4 + if args.conntype == "ble": + from pybricksdev.ble import find_device as find_ble + from pybricksdev.connections.pybricks import PybricksHubBLE + else: + from usb.core import find as find_usb - async def reconnect_hub(): - if not await questionary.confirm( - "\nThe hub has been disconnected. Would you like to re-connect?" - ).ask_async(): - exit() + from pybricksdev.connections.pybricks import PybricksHubUSB + from pybricksdev.usb import ( + EV3_USB_PID, + LEGO_USB_VID, + MINDSTORMS_INVENTOR_USB_PID, + NXT_USB_PID, + SPIKE_ESSENTIAL_USB_PID, + SPIKE_PRIME_USB_PID, + ) - if args.conntype == "ble": - print( - f"Searching for {args.name or 'any hub with Pybricks service'}..." + def is_pybricks_usb(dev): + return ( + (dev.idVendor == LEGO_USB_VID) + and ( + dev.idProduct + in [ + NXT_USB_PID, + EV3_USB_PID, + SPIKE_PRIME_USB_PID, + SPIKE_ESSENTIAL_USB_PID, + MINDSTORMS_INVENTOR_USB_PID, + ] ) - device_or_address = await find_ble(args.name) - hub = PybricksHubBLE(device_or_address) - elif args.conntype == "usb": - device_or_address = find_usb(custom_match=is_pybricks_usb) - hub = PybricksHubUSB(device_or_address) - - await hub.connect() - # re-enable echoing of the hub's stdout - hub._enable_line_handler = True - hub.print_output = True - return hub - - response_options = [ - "Recompile and Run", - "Recompile and Download", - "Run Stored Program", - "Change Target File", - "Exit", - ] - # the entry that is selected by default when the menu opens - # this is overridden after the user picks an option - # so that the default option is always the one that was last chosen - default_response_option = ( - ResponseOptions.RECOMPILE_RUN - if args.start - else ResponseOptions.RECOMPILE_DOWNLOAD - ) + and dev.product.endswith("Pybricks") + ) - while True: - try: - if args.file is sys.stdin: - await hub.race_disconnect( - hub.race_power_button_press( - questionary.press_any_key_to_continue( - "The hub will stay connected and echo its output to the terminal. Press any key to exit." - ).ask_async() - ) - ) - return - response = await hub.race_disconnect( + class ResponseOptions(IntEnum): + RECOMPILE_RUN = 0 + RECOMPILE_DOWNLOAD = 1 + RUN_STORED = 2 + CHANGE_TARGET_FILE = 3 + EXIT = 4 + + async def reconnect_hub(): + if not await questionary.confirm( + "\nThe hub has been disconnected. Would you like to re-connect?" + ).ask_async(): + exit() + + if args.conntype == "ble": + print( + f"Searching for {args.name or 'any hub with Pybricks service'}..." + ) + device_or_address = await find_ble(args.name) + hub = PybricksHubBLE(device_or_address) + elif args.conntype == "usb": + device_or_address = find_usb(custom_match=is_pybricks_usb) + hub = PybricksHubUSB(device_or_address) + + await hub.connect() + # re-enable echoing of the hub's stdout + hub._enable_line_handler = True + hub.print_output = True + return hub + + response_options = [ + "Recompile and Run", + "Recompile and Download", + "Run Stored Program", + "Change Target File", + "Exit", + ] + # the entry that is selected by default when the menu opens + # this is overridden after the user picks an option + # so that the default option is always the one that was last chosen + default_response_option = ( + ResponseOptions.RECOMPILE_RUN + if args.start + else ResponseOptions.RECOMPILE_DOWNLOAD + ) + + while True: + try: + if args.file is sys.stdin: + await hub.race_disconnect( hub.race_power_button_press( - questionary.select( - f"Would you like to re-compile {os.path.basename(args.file.name)}?", - response_options, - default=(response_options[default_response_option]), + questionary.press_any_key_to_continue( + "The hub will stay connected and echo its output to the terminal. Press any key to exit." ).ask_async() ) ) + return + response = await hub.race_disconnect( + hub.race_power_button_press( + questionary.select( + f"Would you like to re-compile {os.path.basename(args.file.name)}?", + response_options, + default=(response_options[default_response_option]), + ).ask_async() + ) + ) - default_response_option = response_options.index(response) + default_response_option = response_options.index(response) - match response_options.index(response): + match response_options.index(response): - case ResponseOptions.RECOMPILE_RUN: - with _get_script_path(args.file) as script_path: - await hub.run(script_path, wait=True) + case ResponseOptions.RECOMPILE_RUN: + with _get_script_path(args.file) as script_path: + await hub.run(script_path, wait=True) - case ResponseOptions.RECOMPILE_DOWNLOAD: - with _get_script_path(args.file) as script_path: - await hub.download(script_path) + case ResponseOptions.RECOMPILE_DOWNLOAD: + with _get_script_path(args.file) as script_path: + await hub.download(script_path) - case ResponseOptions.RUN_STORED: - if hub.fw_version < Version("3.2.0-beta.4"): - print( - "Running a stored program remotely is only supported in the hub firmware version >= v3.2.0." - ) - else: - await hub.start_user_program() - await hub._wait_for_user_program_stop() - - case ResponseOptions.CHANGE_TARGET_FILE: - args.file.close() - while True: - try: - args.file = open( - await hub.race_disconnect( - hub.race_power_button_press( - questionary.path( - "What file would you like to use?" - ).ask_async() - ) + case ResponseOptions.RUN_STORED: + if hub.fw_version < Version("3.2.0-beta.4"): + print( + "Running a stored program remotely is only supported in the hub firmware version >= v3.2.0." + ) + else: + await hub.start_user_program() + await hub._wait_for_user_program_stop() + + case ResponseOptions.CHANGE_TARGET_FILE: + args.file.close() + while True: + try: + args.file = open( + await hub.race_disconnect( + hub.race_power_button_press( + questionary.path( + "What file would you like to use?" + ).ask_async() ) ) - break - except FileNotFoundError: - print("The file was not found. Please try again.") - # send the new target file to the hub - with _get_script_path(args.file) as script_path: - await hub.download(script_path) - - case _: - return - - except SyntaxError as e: - print() - print("A syntax error occurred while parsing your program:") - print(e) - print() - - except HubPowerButtonPressedError: - # This means the user pressed the button on the hub to re-start the - # current program, so the menu was canceled and we are now printing - # the hub stdout until the user program ends on the hub. - try: - await hub._wait_for_power_button_release() - await hub._wait_for_user_program_stop() - - except HubDisconnectError: - hub = await reconnect_hub() + ) + break + except FileNotFoundError: + print("The file was not found. Please try again.") + # send the new target file to the hub + with _get_script_path(args.file) as script_path: + await hub.download(script_path) + + case _: + return + + except SyntaxError as e: + print() + print("A syntax error occurred while parsing your program:") + print(e) + print() + + except HubPowerButtonPressedError: + # This means the user pressed the button on the hub to re-start the + # current program, so the menu was canceled and we are now printing + # the hub stdout until the user program ends on the hub. + try: + await hub._wait_for_power_button_release() + await hub._wait_for_user_program_stop() except HubDisconnectError: - # let terminal cool off before making a new prompt - await asyncio.sleep(0.3) hub = await reconnect_hub() + except HubDisconnectError: + # let terminal cool off before making a new prompt + await asyncio.sleep(0.3) + hub = await reconnect_hub() + + async def run(self, args: argparse.Namespace): + # Pick the right connection if args.conntype == "ble": from pybricksdev.ble import find_device as find_ble @@ -382,13 +414,15 @@ def is_pybricks_usb(dev): await hub.download(script_path) if args.stay_connected: - await stay_connected_menu(hub) + await self.stay_connected_menu(hub) except SyntaxError as e: - print("\nA syntax error occurred while parsing your program:") - print(e, "\n") + print() + print("A syntax error occurred while parsing your program:") + print(e) + print() if args.stay_connected: - await stay_connected_menu(hub) + await self.stay_connected_menu(hub) finally: await hub.disconnect() From 9f7101335d90cbbea2fdbbfb4b6632e1e088f96e Mon Sep 17 00:00:00 2001 From: shaggy Date: Sat, 29 Nov 2025 17:37:07 -0600 Subject: [PATCH 08/19] explicitly pass args to the stay-connected menu from the run function --- pybricksdev/cli/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index b0eda49..b5e70bc 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -414,7 +414,7 @@ def is_pybricks_usb(dev): await hub.download(script_path) if args.stay_connected: - await self.stay_connected_menu(hub) + await self.stay_connected_menu(hub, args) except SyntaxError as e: print() @@ -422,7 +422,7 @@ def is_pybricks_usb(dev): print(e) print() if args.stay_connected: - await self.stay_connected_menu(hub) + await self.stay_connected_menu(hub, args) finally: await hub.disconnect() From 1e0e8bc73646f94b803dac2b9d3a1b59b21aafda Mon Sep 17 00:00:00 2001 From: shaggy Date: Sat, 29 Nov 2025 17:59:00 -0600 Subject: [PATCH 09/19] catch a CalledProcessError and decode stderr --- pybricksdev/cli/__init__.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index b5e70bc..7fdfb54 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -8,6 +8,7 @@ import contextlib import logging import os +import subprocess import sys from abc import ABC, abstractmethod from enum import IntEnum @@ -327,11 +328,10 @@ async def reconnect_hub(): case _: return - except SyntaxError as e: + except subprocess.CalledProcessError as e : print() print("A syntax error occurred while parsing your program:") - print(e) - print() + print(e.stderr.decode()) except HubPowerButtonPressedError: # This means the user pressed the button on the hub to re-start the @@ -416,11 +416,10 @@ def is_pybricks_usb(dev): if args.stay_connected: await self.stay_connected_menu(hub, args) - except SyntaxError as e: + except subprocess.CalledProcessError as e: print() print("A syntax error occurred while parsing your program:") - print(e) - print() + print(e.stderr.decode()) if args.stay_connected: await self.stay_connected_menu(hub, args) From d7cd4bd7a9f99fef68c59aa46542a7112efd8e46 Mon Sep 17 00:00:00 2001 From: shaggy Date: Sat, 29 Nov 2025 18:00:50 -0600 Subject: [PATCH 10/19] fix typo in except block --- pybricksdev/cli/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index 7fdfb54..8432d99 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -328,7 +328,7 @@ async def reconnect_hub(): case _: return - except subprocess.CalledProcessError as e : + except subprocess.CalledProcessError as e: print() print("A syntax error occurred while parsing your program:") print(e.stderr.decode()) From 681aceebb55d2120af2992f98af854f20fe5bed9 Mon Sep 17 00:00:00 2001 From: shaggy Date: Sun, 30 Nov 2025 21:40:53 -0600 Subject: [PATCH 11/19] add a few unit tests related to stay_connected_menu --- tests/test_cli.py | 313 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 313 insertions(+) diff --git a/tests/test_cli.py b/tests/test_cli.py index 51d58cd..556af54 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,15 +1,22 @@ """Tests for the pybricksdev CLI commands.""" import argparse +import asyncio import contextlib import io import os +import subprocess import tempfile from unittest.mock import AsyncMock, Mock, mock_open, patch import pytest +from packaging.version import Version from pybricksdev.cli import Compile, Run, Tool, Udev +from pybricksdev.connections.pybricks import ( + HubDisconnectError, + HubPowerButtonPressedError, +) class TestTool: @@ -442,6 +449,312 @@ async def test_run_connection_error(self): # Verify disconnect was not called since connection failed mock_hub.disconnect.assert_not_called() + @pytest.mark.asyncio + async def test_run_syntax_error(self): + """Test that the stay connected menu is called upon a syntax error when the appropriate flag is active.""" + + # Create a mock hub + mock_hub = AsyncMock() + mock_hub.run = AsyncMock( + side_effect=subprocess.CalledProcessError( + returncode=1, cmd="test", stderr=b"test" + ) + ) + mock_hub.connect = AsyncMock() + + # Set up mocks using ExitStack + with contextlib.ExitStack() as stack: + # Create and manage temporary file + temp = stack.enter_context( + tempfile.NamedTemporaryFile( + suffix=".py", mode="w+", delete=False, encoding="utf-8" + ) + ) + temp.write("print('test')") + temp_path = temp.name + stack.callback(os.unlink, temp_path) + + # Create args + args = argparse.Namespace( + conntype="ble", + file=stack.enter_context(open(temp_path, "r", encoding="utf-8")), + name="MyHub", + start=True, + wait=True, + stay_connected=True, + ) + + mock_hub_class = stack.enter_context( + patch( + "pybricksdev.connections.pybricks.PybricksHubBLE", + return_value=mock_hub, + ) + ) + stack.enter_context( + patch("pybricksdev.ble.find_device", return_value="mock_device") + ) + mock_menu = stack.enter_context( + patch("pybricksdev.cli.Run.stay_connected_menu") + ) + + # Run the command + run_cmd = Run() + await run_cmd.run(args) + + # Verify the hub was created and used correctly + mock_hub_class.assert_called_once_with("mock_device") + mock_hub.connect.assert_called_once() + mock_hub.run.assert_called_once_with(temp_path, True) + mock_menu.assert_called_once_with(mock_hub, args) + mock_hub.disconnect.assert_called_once() + + @pytest.mark.asyncio + async def test_stay_connected_menu_integration(self): + """Test all of the basic options in the stay_connected menu.""" + + async def passthrough_awaitable(awaitable): + return await awaitable + + # Create a mock hub + mock_hub = AsyncMock() + mock_hub.fw_version = Version("3.2.0-beta.4") + mock_hub.run = AsyncMock() + mock_hub.connect = AsyncMock() + mock_hub.start_user_program = AsyncMock() + mock_hub._wait_for_user_program_stop = AsyncMock() + mock_hub.race_disconnect = mock_hub.race_power_button_press = AsyncMock( + side_effect=passthrough_awaitable + ) + mock_hub.download = AsyncMock() + + # create a mock questionary menu + mock_selector = AsyncMock() + mock_selector.ask_async.side_effect = [ + "Recompile and Run", + "Recompile and Download", + "Run Stored Program", + "Exit", + ] + + # Set up mocks using ExitStack + with contextlib.ExitStack() as stack: + # Create and manage temporary file + temp = stack.enter_context( + tempfile.NamedTemporaryFile( + suffix=".py", mode="w+", delete=False, encoding="utf-8" + ) + ) + temp.write("print('test')") + temp_path = temp.name + stack.callback(os.unlink, temp_path) + + # Create args + args = argparse.Namespace( + conntype="ble", + file=stack.enter_context(open(temp_path, "r", encoding="utf-8")), + name="MyHub", + start=True, + wait=True, + stay_connected=True, + ) + + mock_hub_class = stack.enter_context( + patch( + "pybricksdev.connections.pybricks.PybricksHubBLE", + return_value=mock_hub, + ) + ) + stack.enter_context( + patch("pybricksdev.ble.find_device", return_value="mock_device") + ) + mock_selector = stack.enter_context( + patch("questionary.select", return_value=mock_selector) + ) + + # Run the command + run_cmd = Run() + await run_cmd.run(args) + + # Verify the hub was created and used correctly + mock_hub_class.assert_called_once_with("mock_device") + mock_hub.connect.assert_called_once() + assert mock_hub.run.call_count == 2 + mock_hub.run.assert_called_with(temp_path, wait=True) + mock_hub.download.assert_called_once_with(temp_path) + mock_hub.start_user_program.assert_called_once() + mock_hub._wait_for_user_program_stop.assert_called_once() + assert mock_selector.call_count == 4 + mock_hub.disconnect.assert_called_once() + + @pytest.mark.asyncio + async def test_stay_connected_menu_change_target_file(self): + """Test the change target file option.""" + + async def passthrough_awaitable(awaitable): + return await awaitable + + # Create a mock hub + mock_hub = AsyncMock() + mock_hub.run = AsyncMock() + mock_hub.connect = AsyncMock() + mock_hub.race_disconnect = mock_hub.race_power_button_press = AsyncMock( + side_effect=passthrough_awaitable + ) + mock_hub.download = AsyncMock() + + # create a mock questionary menu + mock_menu = AsyncMock() + mock_menu.ask_async.side_effect = [ + "Change Target File", + "Recompile and Run", + "Exit", + ] + + # Set up mocks using ExitStack + with contextlib.ExitStack() as stack: + # Create and manage temporary file + temp = stack.enter_context( + tempfile.NamedTemporaryFile( + suffix=".py", mode="w+", delete=False, encoding="utf-8" + ) + ) + new_target_file = stack.enter_context( + tempfile.NamedTemporaryFile( + suffix=".py", mode="w+", delete=False, encoding="utf-8" + ) + ) + temp.write("print('test')") + temp_path = temp.name + stack.callback(os.unlink, temp_path) + new_target_file.write("print('test')") + new_path = new_target_file.name + stack.callback(os.unlink, new_path) + + # Create args + args = argparse.Namespace( + conntype="ble", + file=stack.enter_context(open(temp_path, "r", encoding="utf-8")), + name="MyHub", + start=True, + wait=True, + stay_connected=True, + ) + + stack.enter_context( + patch("pybricksdev.ble.find_device", return_value="mock_device") + ) + mock_selector = stack.enter_context( + patch("questionary.select", return_value=mock_menu) + ) + mock_file_selector = stack.enter_context(patch("questionary.path")) + path_obj = AsyncMock() + path_obj.ask_async = AsyncMock(return_value=new_path) + mock_file_selector.return_value = path_obj + + # Run the command + run_cmd = Run() + await run_cmd.stay_connected_menu(mock_hub, args) + + assert mock_selector.call_count == 3 + mock_file_selector.assert_called_once() + mock_hub.download.assert_called_once_with(new_path) + mock_hub.run.assert_called_once_with(new_path, wait=True) + + @pytest.mark.asyncio + async def test_stay_connected_menu_interruptions(self): + """Test the stay_connected_menu being interrupted by a power button press or hub disconnect.""" + disconnect_call_count = 0 + power_call_count = 0 + + # simulates the hub disconnecting on the first call, + async def mock_race_disconnect(awaitable): + task = asyncio.ensure_future(awaitable) + nonlocal disconnect_call_count + disconnect_call_count += 1 + if disconnect_call_count == 1: + task.cancel() + raise HubDisconnectError("hub disconnected") + return await awaitable + + async def mock_race_power_button_press(awaitable): + task = asyncio.ensure_future(awaitable) + nonlocal power_call_count + power_call_count += 1 + if power_call_count == 2: + task.cancel() + raise HubPowerButtonPressedError("power button pressed") + return await awaitable + + # Create a mock hub + mock_hub = AsyncMock() + mock_hub.run = AsyncMock() + mock_hub.connect = AsyncMock() + mock_hub.disconnect = AsyncMock() + mock_hub.race_disconnect = AsyncMock( + side_effect=mock_race_disconnect, + ) + mock_hub.race_power_button_press = AsyncMock( + side_effect=mock_race_power_button_press, + ) + mock_hub._wait_for_power_button_release = AsyncMock() + mock_hub._wait_for_user_program_stop = AsyncMock() + # create a mock questionary menu + mock_menu = AsyncMock() + mock_menu.ask_async.side_effect = [ + "Recompile and Run", + "Exit", + ] + mock_confirm = AsyncMock() + mock_confirm.ask_async.return_value = True + + # Set up mocks using ExitStack + with contextlib.ExitStack() as stack: + # Create and manage temporary file + temp = stack.enter_context( + tempfile.NamedTemporaryFile( + suffix=".py", mode="w+", delete=False, encoding="utf-8" + ) + ) + temp.write("print('test')") + temp_path = temp.name + stack.callback(os.unlink, temp_path) + + # Create args + args = argparse.Namespace( + conntype="ble", + file=stack.enter_context(open(temp_path, "r", encoding="utf-8")), + name="MyHub", + start=True, + wait=True, + stay_connected=True, + ) + + mock_hub_class = stack.enter_context( + patch( + "pybricksdev.connections.pybricks.PybricksHubBLE", + return_value=mock_hub, + ) + ) + + stack.enter_context( + patch("pybricksdev.ble.find_device", return_value="mock_device") + ) + mock_selector = stack.enter_context( + patch("questionary.select", return_value=mock_menu) + ) + stack.enter_context(patch("questionary.confirm", return_value=mock_confirm)) + + # Run the command + run_cmd = Run() + await run_cmd.stay_connected_menu(mock_hub, args) + + assert mock_selector.call_count == 4 + mock_hub_class.assert_called_once() + mock_hub.connect.assert_called_once() + mock_hub._wait_for_power_button_release.assert_called_once() + mock_hub._wait_for_user_program_stop.assert_called_once() + mock_hub.run.assert_called_once_with(temp_path, wait=True) + class TestCompile: """Tests for the Compile command.""" From ec917ab3d3c42767b2bbde7f68657f9ea95a44d2 Mon Sep 17 00:00:00 2001 From: shaggy Date: Sun, 30 Nov 2025 21:52:06 -0600 Subject: [PATCH 12/19] make the open() call in the "Change Target File" option explicitly specify utf-8 encoding --- pybricksdev/cli/__init__.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index 8432d99..5c73be1 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -316,7 +316,8 @@ async def reconnect_hub(): "What file would you like to use?" ).ask_async() ) - ) + ), + encoding="utf-8", ) break except FileNotFoundError: From 45fed26f59c4f834a2fefaf185dfd289dd4a3cbe Mon Sep 17 00:00:00 2001 From: shaggy Date: Mon, 1 Dec 2025 09:05:07 -0600 Subject: [PATCH 13/19] fix leaking coroutines in test_stay_connected_menu_interruptions It looks like the mock menu had to have the side effect of an actual async function for it to return a proper coroutine. When its side effect was hardcoded values, the AsyncMock function was unable to be cancelled properly, causing coroutines to leak. --- tests/test_cli.py | 66 ++++++++++++++++++++++++++++++++--------------- 1 file changed, 45 insertions(+), 21 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 556af54..30edfa2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -663,27 +663,44 @@ async def passthrough_awaitable(awaitable): @pytest.mark.asyncio async def test_stay_connected_menu_interruptions(self): """Test the stay_connected_menu being interrupted by a power button press or hub disconnect.""" - disconnect_call_count = 0 - power_call_count = 0 + mock_menu_call_count = 0 # simulates the hub disconnecting on the first call, async def mock_race_disconnect(awaitable): - task = asyncio.ensure_future(awaitable) - nonlocal disconnect_call_count - disconnect_call_count += 1 - if disconnect_call_count == 1: + task = asyncio.create_task(awaitable) + try: + if mock_menu_call_count == 1: + await asyncio.sleep(0.01) + task.cancel() + raise HubDisconnectError("hub disconnected") + return await task + except BaseException: + await asyncio.sleep(0.01) task.cancel() - raise HubDisconnectError("hub disconnected") - return await awaitable + raise + # simulate the power button being pressed on the second call async def mock_race_power_button_press(awaitable): - task = asyncio.ensure_future(awaitable) - nonlocal power_call_count - power_call_count += 1 - if power_call_count == 2: + task = asyncio.create_task(awaitable) + try: + if mock_menu_call_count == 2: + await asyncio.sleep(0.01) + task.cancel() + raise HubPowerButtonPressedError("power button pressed") + return await task + except BaseException: + await asyncio.sleep(0.01) task.cancel() - raise HubPowerButtonPressedError("power button pressed") - return await awaitable + raise + + # should be called but cancelled twice, returning "Recompile and Run" the third time + async def mock_menu_function(): + nonlocal mock_menu_call_count + mock_menu_call_count += 1 + if mock_menu_call_count <= 3: + return "Recompile and Run" + else: + return "Exit" # Create a mock hub mock_hub = AsyncMock() @@ -700,12 +717,11 @@ async def mock_race_power_button_press(awaitable): mock_hub._wait_for_user_program_stop = AsyncMock() # create a mock questionary menu mock_menu = AsyncMock() - mock_menu.ask_async.side_effect = [ - "Recompile and Run", - "Exit", - ] - mock_confirm = AsyncMock() - mock_confirm.ask_async.return_value = True + mock_menu.ask_async.side_effect = mock_menu_function + + # create a mock confirmation menu to reconnect to the hub + mock_confirm_menu = AsyncMock() + mock_confirm_menu.ask_async.return_value = True # Set up mocks using ExitStack with contextlib.ExitStack() as stack: @@ -742,17 +758,25 @@ async def mock_race_power_button_press(awaitable): mock_selector = stack.enter_context( patch("questionary.select", return_value=mock_menu) ) - stack.enter_context(patch("questionary.confirm", return_value=mock_confirm)) + mock_confirm = stack.enter_context( + patch("questionary.confirm", return_value=mock_confirm_menu) + ) # Run the command run_cmd = Run() await run_cmd.stay_connected_menu(mock_hub, args) assert mock_selector.call_count == 4 + # a confirmation menu should be triggered and the hub should be re-instantiated upon a HubDisconnectError + mock_confirm.assert_called_once() mock_hub_class.assert_called_once() mock_hub.connect.assert_called_once() + + # these functions should be triggered upon a HubPowerButtonPressedError mock_hub._wait_for_power_button_release.assert_called_once() mock_hub._wait_for_user_program_stop.assert_called_once() + + # this should only be called once because the menu was canceled the first two times it was called mock_hub.run.assert_called_once_with(temp_path, wait=True) From 95775506e49407d0e0ea30a010329a4b8ddc01da Mon Sep 17 00:00:00 2001 From: shaggy Date: Thu, 4 Dec 2025 09:25:55 -0600 Subject: [PATCH 14/19] add a test for the `Run Stored Program` option The function that this option calls is not available on hub firmware versions prior to 3.2.0-beta.3, so this test confirms that it is not called when the hub's firmware version is too old. --- tests/test_cli.py | 55 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/tests/test_cli.py b/tests/test_cli.py index 30edfa2..58a3274 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -578,11 +578,13 @@ async def passthrough_awaitable(awaitable): # Verify the hub was created and used correctly mock_hub_class.assert_called_once_with("mock_device") mock_hub.connect.assert_called_once() + assert mock_hub.run.call_count == 2 mock_hub.run.assert_called_with(temp_path, wait=True) mock_hub.download.assert_called_once_with(temp_path) mock_hub.start_user_program.assert_called_once() mock_hub._wait_for_user_program_stop.assert_called_once() + assert mock_selector.call_count == 4 mock_hub.disconnect.assert_called_once() @@ -779,6 +781,59 @@ async def mock_menu_function(): # this should only be called once because the menu was canceled the first two times it was called mock_hub.run.assert_called_once_with(temp_path, wait=True) + @pytest.mark.asyncio + async def test_stay_connected_menu_run_stored(self): + """Test that the run_stored program option doesn't call an inaccessible method on an old hub.""" + + async def passthrough_awaitable(awaitable): + return await awaitable + + # Create a mock hub + old_mock_hub = AsyncMock() + + # simulate an old hub that can't handle the run_stored_program option + old_mock_hub.fw_version = Version("3.2.0-beta.3") + old_mock_hub.run = AsyncMock() + old_mock_hub.connect = AsyncMock() + old_mock_hub.start_user_program = AsyncMock() + old_mock_hub.race_disconnect = old_mock_hub.race_power_button_press = AsyncMock( + side_effect=passthrough_awaitable + ) + + # create a mock questionary menu + mock_selector = AsyncMock() + mock_selector.ask_async.side_effect = [ + "Run Stored Program", + "Exit", + ] + + # Set up mocks using ExitStack + with contextlib.ExitStack() as stack: + # Create and manage temporary file + + # Create args + args = argparse.Namespace( + conntype="ble", + file=stack.enter_context( + tempfile.NamedTemporaryFile( + suffix=".py", mode="w+", delete=False, encoding="utf-8" + ) + ), + name="MyHub", + start=True, + wait=True, + stay_connected=True, + ) + + stack.enter_context(patch("questionary.select", return_value=mock_selector)) + + # Run the command + run_cmd = Run() + await run_cmd.stay_connected_menu(old_mock_hub, args) + + old_mock_hub.start_user_program.assert_not_called() + old_mock_hub._wait_for_user_program_stop.assert_not_called() + class TestCompile: """Tests for the Compile command.""" From 4e3f4b8676238e2b7488086afc500988bf4be1d9 Mon Sep 17 00:00:00 2001 From: shaggy Date: Thu, 4 Dec 2025 09:27:48 -0600 Subject: [PATCH 15/19] rearrange test order --- tests/test_cli.py | 106 +++++++++++++++++++++++----------------------- 1 file changed, 53 insertions(+), 53 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 58a3274..39e6746 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -662,6 +662,59 @@ async def passthrough_awaitable(awaitable): mock_hub.download.assert_called_once_with(new_path) mock_hub.run.assert_called_once_with(new_path, wait=True) + @pytest.mark.asyncio + async def test_stay_connected_menu_run_stored(self): + """Test that the run_stored program option doesn't call an inaccessible method on an old hub.""" + + async def passthrough_awaitable(awaitable): + return await awaitable + + # Create a mock hub + old_mock_hub = AsyncMock() + + # simulate an old hub that can't handle the run_stored_program option + old_mock_hub.fw_version = Version("3.2.0-beta.3") + old_mock_hub.run = AsyncMock() + old_mock_hub.connect = AsyncMock() + old_mock_hub.start_user_program = AsyncMock() + old_mock_hub.race_disconnect = old_mock_hub.race_power_button_press = AsyncMock( + side_effect=passthrough_awaitable + ) + + # create a mock questionary menu + mock_selector = AsyncMock() + mock_selector.ask_async.side_effect = [ + "Run Stored Program", + "Exit", + ] + + # Set up mocks using ExitStack + with contextlib.ExitStack() as stack: + # Create and manage temporary file + + # Create args + args = argparse.Namespace( + conntype="ble", + file=stack.enter_context( + tempfile.NamedTemporaryFile( + suffix=".py", mode="w+", delete=False, encoding="utf-8" + ) + ), + name="MyHub", + start=True, + wait=True, + stay_connected=True, + ) + + stack.enter_context(patch("questionary.select", return_value=mock_selector)) + + # Run the command + run_cmd = Run() + await run_cmd.stay_connected_menu(old_mock_hub, args) + + old_mock_hub.start_user_program.assert_not_called() + old_mock_hub._wait_for_user_program_stop.assert_not_called() + @pytest.mark.asyncio async def test_stay_connected_menu_interruptions(self): """Test the stay_connected_menu being interrupted by a power button press or hub disconnect.""" @@ -781,59 +834,6 @@ async def mock_menu_function(): # this should only be called once because the menu was canceled the first two times it was called mock_hub.run.assert_called_once_with(temp_path, wait=True) - @pytest.mark.asyncio - async def test_stay_connected_menu_run_stored(self): - """Test that the run_stored program option doesn't call an inaccessible method on an old hub.""" - - async def passthrough_awaitable(awaitable): - return await awaitable - - # Create a mock hub - old_mock_hub = AsyncMock() - - # simulate an old hub that can't handle the run_stored_program option - old_mock_hub.fw_version = Version("3.2.0-beta.3") - old_mock_hub.run = AsyncMock() - old_mock_hub.connect = AsyncMock() - old_mock_hub.start_user_program = AsyncMock() - old_mock_hub.race_disconnect = old_mock_hub.race_power_button_press = AsyncMock( - side_effect=passthrough_awaitable - ) - - # create a mock questionary menu - mock_selector = AsyncMock() - mock_selector.ask_async.side_effect = [ - "Run Stored Program", - "Exit", - ] - - # Set up mocks using ExitStack - with contextlib.ExitStack() as stack: - # Create and manage temporary file - - # Create args - args = argparse.Namespace( - conntype="ble", - file=stack.enter_context( - tempfile.NamedTemporaryFile( - suffix=".py", mode="w+", delete=False, encoding="utf-8" - ) - ), - name="MyHub", - start=True, - wait=True, - stay_connected=True, - ) - - stack.enter_context(patch("questionary.select", return_value=mock_selector)) - - # Run the command - run_cmd = Run() - await run_cmd.stay_connected_menu(old_mock_hub, args) - - old_mock_hub.start_user_program.assert_not_called() - old_mock_hub._wait_for_user_program_stop.assert_not_called() - class TestCompile: """Tests for the Compile command.""" From 884adce5f9f102031fb3f4ff87d961f72c6698ca Mon Sep 17 00:00:00 2001 From: shaggy Date: Wed, 17 Dec 2025 11:08:26 -0600 Subject: [PATCH 16/19] tweak mpy-cross error message and control flow --- pybricksdev/cli/__init__.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/pybricksdev/cli/__init__.py b/pybricksdev/cli/__init__.py index 5c73be1..cacfd33 100644 --- a/pybricksdev/cli/__init__.py +++ b/pybricksdev/cli/__init__.py @@ -331,7 +331,7 @@ async def reconnect_hub(): except subprocess.CalledProcessError as e: print() - print("A syntax error occurred while parsing your program:") + print("mpy-cross failed to compile the program:") print(e.stderr.decode()) except HubPowerButtonPressedError: @@ -414,17 +414,14 @@ def is_pybricks_usb(dev): hub._enable_line_handler = True await hub.download(script_path) - if args.stay_connected: - await self.stay_connected_menu(hub, args) - except subprocess.CalledProcessError as e: print() - print("A syntax error occurred while parsing your program:") + print("mpy-cross failed to compile the program:") print(e.stderr.decode()) - if args.stay_connected: - await self.stay_connected_menu(hub, args) finally: + if args.stay_connected: + await self.stay_connected_menu(hub, args) await hub.disconnect() From 6c083d309964b0cc4ddc0276ffcd15e3d518338f Mon Sep 17 00:00:00 2001 From: shaggy Date: Wed, 17 Dec 2025 11:12:12 -0600 Subject: [PATCH 17/19] create a syntax error organically in the unit test --- tests/test_cli.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 39e6746..91ad47e 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -455,11 +455,6 @@ async def test_run_syntax_error(self): # Create a mock hub mock_hub = AsyncMock() - mock_hub.run = AsyncMock( - side_effect=subprocess.CalledProcessError( - returncode=1, cmd="test", stderr=b"test" - ) - ) mock_hub.connect = AsyncMock() # Set up mocks using ExitStack @@ -470,7 +465,7 @@ async def test_run_syntax_error(self): suffix=".py", mode="w+", delete=False, encoding="utf-8" ) ) - temp.write("print('test')") + temp.write("syntax error") temp_path = temp.name stack.callback(os.unlink, temp_path) From ff6799b30469102eab00c003891be713cc4752f5 Mon Sep 17 00:00:00 2001 From: shaggy Date: Wed, 17 Dec 2025 11:13:32 -0600 Subject: [PATCH 18/19] remove unused subprocess import --- tests/test_cli.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 91ad47e..c30d5d0 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -5,7 +5,6 @@ import contextlib import io import os -import subprocess import tempfile from unittest.mock import AsyncMock, Mock, mock_open, patch From 11c7fdb0f0b07be3f9540f96426d5f06c1d259c1 Mon Sep 17 00:00:00 2001 From: shaggy Date: Wed, 17 Dec 2025 11:21:56 -0600 Subject: [PATCH 19/19] un-mock the stay-connected menu in the syntax error test --- tests/test_cli.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index c30d5d0..5152492 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -452,9 +452,18 @@ async def test_run_connection_error(self): async def test_run_syntax_error(self): """Test that the stay connected menu is called upon a syntax error when the appropriate flag is active.""" + async def passthrough_awaitable(awaitable): + return await awaitable + # Create a mock hub mock_hub = AsyncMock() mock_hub.connect = AsyncMock() + mock_hub.race_disconnect = mock_hub.race_power_button_press = AsyncMock( + side_effect=passthrough_awaitable + ) + + mock_selector = AsyncMock() + mock_selector.ask_async.side_effect = ["Exit"] # Set up mocks using ExitStack with contextlib.ExitStack() as stack: @@ -487,8 +496,8 @@ async def test_run_syntax_error(self): stack.enter_context( patch("pybricksdev.ble.find_device", return_value="mock_device") ) - mock_menu = stack.enter_context( - patch("pybricksdev.cli.Run.stay_connected_menu") + mock_selector = stack.enter_context( + patch("questionary.select", return_value=mock_selector) ) # Run the command @@ -499,7 +508,7 @@ async def test_run_syntax_error(self): mock_hub_class.assert_called_once_with("mock_device") mock_hub.connect.assert_called_once() mock_hub.run.assert_called_once_with(temp_path, True) - mock_menu.assert_called_once_with(mock_hub, args) + mock_selector.assert_called_once() mock_hub.disconnect.assert_called_once() @pytest.mark.asyncio