Add a new download() method for PybricksHub.

Author: 1e0ng

pybricksdev/cli/__init__.py

@@ -325,21 +325,7 @@ def is_pybricks_usb(dev):
         await hub.connect()
         try:
             with _get_script_path(args.file) as script_path:
-                # For PybricksHub, we use download_user_program to just upload without running
-                if args.conntype in ["ble", "usb"]:
-                    from pybricksdev.compile import compile_multi_file
-
-                    # Compile the script to mpy format
-                    mpy = await compile_multi_file(
-                        script_path, hub._mpy_abi_version or 6
-                    )
-                    # Upload without running
-                    await hub.download_user_program(mpy)
-                # For EV3Connection, we just use download
-                elif args.conntype == "ssh":
-                    await hub.download(script_path)
-                else:
-                    raise RuntimeError("Unsupported hub type for download command")
+                await hub.download(script_path)
         finally:
             await hub.disconnect()
 

tests/test_cli.py

@@ -64,7 +64,7 @@ async def test_download_ble(self):
         # Create a mock hub
         mock_hub = AsyncMock()
         mock_hub._mpy_abi_version = 6
-        mock_hub.download_user_program = AsyncMock()
+        mock_hub.download = AsyncMock()
 
         # Create a temporary file
         with tempfile.NamedTemporaryFile(suffix=".py", mode="w+", delete=False) as temp:
@@ -79,27 +79,20 @@ async def test_download_ble(self):
                 name="MyHub",
             )
 
-            # Mock the compilation
-            mock_mpy = MagicMock()
-            mock_mpy.__bytes__ = lambda self: b"compiled code"
-
             # Mock the hub creation
             with patch(
                 "pybricksdev.connections.pybricks.PybricksHubBLE", return_value=mock_hub
             ) as mock_hub_class:
                 with patch("pybricksdev.ble.find_device", return_value="mock_device"):
-                    with patch(
-                        "pybricksdev.compile.compile_multi_file", return_value=mock_mpy
-                    ):
-                        # Run the command
-                        download = Download()
-                        await download.run(args)
+                    # Run the command
+                    download = Download()
+                    await download.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.download_user_program.assert_called_once()
-                        mock_hub.disconnect.assert_called_once()
+                    # 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.download.assert_called_once()
+                    mock_hub.disconnect.assert_called_once()
         finally:
             os.unlink(temp_path)
 
@@ -109,7 +102,7 @@ async def test_download_usb(self):
         # Create a mock hub
         mock_hub = AsyncMock()
         mock_hub._mpy_abi_version = 6
-        mock_hub.download_user_program = AsyncMock()
+        mock_hub.download = AsyncMock()
 
         # Create a temporary file
         with tempfile.NamedTemporaryFile(suffix=".py", mode="w+", delete=False) as temp:
@@ -124,27 +117,20 @@ async def test_download_usb(self):
                 name=None,
             )
 
-            # Mock the compilation
-            mock_mpy = MagicMock()
-            mock_mpy.__bytes__ = lambda self: b"compiled code"
-
             # Mock the hub creation
             with patch(
                 "pybricksdev.connections.pybricks.PybricksHubUSB", return_value=mock_hub
             ) as mock_hub_class:
                 with patch("usb.core.find", return_value="mock_device"):
-                    with patch(
-                        "pybricksdev.compile.compile_multi_file", return_value=mock_mpy
-                    ):
-                        # Run the command
-                        download = Download()
-                        await download.run(args)
+                    # Run the command
+                    download = Download()
+                    await download.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.download_user_program.assert_called_once()
-                        mock_hub.disconnect.assert_called_once()
+                    # 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.download.assert_called_once()
+                    mock_hub.disconnect.assert_called_once()
         finally:
             os.unlink(temp_path)
 
@@ -214,7 +200,7 @@ async def test_download_stdin(self):
         # Create a mock hub
         mock_hub = AsyncMock()
         mock_hub._mpy_abi_version = 6
-        mock_hub.download_user_program = AsyncMock()
+        mock_hub.download = AsyncMock()
 
         # Create a mock stdin
         mock_stdin = io.StringIO("print('test')")
@@ -228,31 +214,22 @@ async def test_download_stdin(self):
             name="MyHub",
         )
 
-        # Mock the compilation
-        mock_mpy = MagicMock()
-        mock_mpy.__bytes__ = lambda self: b"compiled code"
-
         # Mock the hub creation and file handling
         with patch(
             "pybricksdev.connections.pybricks.PybricksHubBLE", return_value=mock_hub
         ) as mock_hub_class:
             with patch("pybricksdev.ble.find_device", return_value="mock_device"):
-                with patch(
-                    "pybricksdev.compile.compile_multi_file", return_value=mock_mpy
-                ):
-                    with patch("tempfile.NamedTemporaryFile") as mock_temp:
-                        mock_temp.return_value.__enter__.return_value.name = (
-                            "/tmp/test.py"
-                        )
-                        # Run the command
-                        download = Download()
-                        await download.run(args)
+                with patch("tempfile.NamedTemporaryFile") as mock_temp:
+                    mock_temp.return_value.__enter__.return_value.name = "/tmp/test.py"
+                    # Run the command
+                    download = Download()
+                    await download.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.download_user_program.assert_called_once()
-                        mock_hub.disconnect.assert_called_once()
+                    # 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.download.assert_called_once()
+                    mock_hub.disconnect.assert_called_once()
 
     @pytest.mark.asyncio
     async def test_download_connection_error(self):