Skip to content

Commit 555a3d1

Browse files
1e0ng1e0ng
committed
Add unit tests for pybricks download function.
1 parent e213691 commit 555a3d1

File tree

1 file changed

+167
-0
lines changed

1 file changed

+167
-0
lines changed

‎tests/connections/test_pybricks.py‎

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
"""Tests for the pybricks connection module."""
2+
3+
import asyncio
4+
import tempfile
5+
from unittest.mock import AsyncMock, PropertyMock
6+
7+
import pytest
8+
from reactivex.subject import Subject
9+
10+
from pybricksdev.connections.pybricks import (
11+
ConnectionState,
12+
HubCapabilityFlag,
13+
HubKind,
14+
PybricksHubBLE,
15+
StatusFlag,
16+
)
17+
18+
19+
class TestPybricksHub:
20+
"""Tests for the PybricksHub base class functionality."""
21+
22+
@pytest.mark.asyncio
23+
async def test_download_modern_protocol(self):
24+
"""Test downloading with modern protocol and capability flags."""
25+
hub = PybricksHubBLE("mock_device")
26+
hub._mpy_abi_version = 6
27+
hub._client = AsyncMock()
28+
hub.get_capabilities = AsyncMock(return_value={"pybricks": {"mpy": True}})
29+
hub.download_user_program = AsyncMock()
30+
type(hub.connection_state_observable).value = PropertyMock(
31+
return_value=ConnectionState.CONNECTED
32+
)
33+
hub._capability_flags = HubCapabilityFlag.USER_PROG_MULTI_FILE_MPY6
34+
35+
with tempfile.NamedTemporaryFile(suffix=".py", mode="w+", delete=False) as temp:
36+
temp.write("print('test')")
37+
temp_path = temp.name
38+
try:
39+
await hub.download(temp_path)
40+
hub.download_user_program.assert_called_once()
41+
finally:
42+
import os
43+
44+
os.unlink(temp_path)
45+
46+
@pytest.mark.asyncio
47+
async def test_download_legacy_firmware(self):
48+
"""Test downloading with legacy firmware."""
49+
hub = PybricksHubBLE("mock_device")
50+
hub._mpy_abi_version = None # Legacy firmware
51+
hub._client = AsyncMock()
52+
hub.download_user_program = AsyncMock()
53+
hub.hub_kind = HubKind.BOOST
54+
type(hub.connection_state_observable).value = PropertyMock(
55+
return_value=ConnectionState.CONNECTED
56+
)
57+
hub._capability_flags = HubCapabilityFlag.USER_PROG_MULTI_FILE_MPY6
58+
59+
with tempfile.NamedTemporaryFile(suffix=".py", mode="w+", delete=False) as temp:
60+
temp.write("print('test')")
61+
temp_path = temp.name
62+
try:
63+
await hub.download(temp_path)
64+
hub.download_user_program.assert_called_once()
65+
finally:
66+
import os
67+
68+
os.unlink(temp_path)
69+
70+
@pytest.mark.asyncio
71+
async def test_download_unsupported_capabilities(self):
72+
"""Test downloading when hub doesn't support required capabilities."""
73+
hub = PybricksHubBLE("mock_device")
74+
hub._mpy_abi_version = 6
75+
hub._client = AsyncMock()
76+
hub.get_capabilities = AsyncMock(return_value={"pybricks": {"mpy": False}})
77+
type(hub.connection_state_observable).value = PropertyMock(
78+
return_value=ConnectionState.CONNECTED
79+
)
80+
hub._capability_flags = 0
81+
82+
with tempfile.NamedTemporaryFile(suffix=".py", mode="w+", delete=False) as temp:
83+
temp.write("print('test')")
84+
temp_path = temp.name
85+
try:
86+
with pytest.raises(
87+
RuntimeError,
88+
match="Hub is not compatible with any of the supported file formats",
89+
):
90+
await hub.download(temp_path)
91+
finally:
92+
import os
93+
94+
os.unlink(temp_path)
95+
96+
@pytest.mark.asyncio
97+
async def test_download_compile_error(self):
98+
"""Test handling compilation errors."""
99+
hub = PybricksHubBLE("mock_device")
100+
hub._mpy_abi_version = 6
101+
hub._client = AsyncMock()
102+
hub.get_capabilities = AsyncMock(return_value={"pybricks": {"mpy": True}})
103+
type(hub.connection_state_observable).value = PropertyMock(
104+
return_value=ConnectionState.CONNECTED
105+
)
106+
hub._capability_flags = HubCapabilityFlag.USER_PROG_MULTI_FILE_MPY6
107+
108+
with tempfile.NamedTemporaryFile(suffix=".py", mode="w+", delete=False) as temp:
109+
temp.write("print('test' # Missing closing parenthesis")
110+
temp_path = temp.name
111+
try:
112+
with pytest.raises(SyntaxError):
113+
await hub.download(temp_path)
114+
finally:
115+
import os
116+
117+
os.unlink(temp_path)
118+
119+
@pytest.mark.asyncio
120+
async def test_run_modern_protocol(self):
121+
"""Test running a program with modern protocol."""
122+
hub = PybricksHubBLE("mock_device")
123+
hub._mpy_abi_version = None # Use modern protocol
124+
hub._client = AsyncMock()
125+
hub.client = AsyncMock()
126+
hub.get_capabilities = AsyncMock(return_value={"pybricks": {"mpy": True}})
127+
hub.download_user_program = AsyncMock()
128+
hub.start_user_program = AsyncMock()
129+
hub.write_gatt_char = AsyncMock()
130+
type(hub.connection_state_observable).value = PropertyMock(
131+
return_value=ConnectionState.CONNECTED
132+
)
133+
hub._capability_flags = HubCapabilityFlag.USER_PROG_MULTI_FILE_MPY6
134+
hub.hub_kind = HubKind.BOOST
135+
136+
# Mock the status observable to simulate program start and stop
137+
status_subject = Subject()
138+
hub.status_observable = status_subject
139+
hub._stdout_line_queue = asyncio.Queue()
140+
hub._enable_line_handler = True
141+
142+
with tempfile.NamedTemporaryFile(suffix=".py", mode="w+", delete=False) as temp:
143+
temp.write("print('test')")
144+
temp_path = temp.name
145+
try:
146+
# Start the run task
147+
run_task = asyncio.create_task(hub.run(temp_path))
148+
149+
# Simulate program start
150+
await asyncio.sleep(0.1)
151+
status_subject.on_next(StatusFlag.USER_PROGRAM_RUNNING)
152+
153+
# Simulate program stop after a short delay
154+
await asyncio.sleep(0.1)
155+
status_subject.on_next(0) # Clear all flags
156+
157+
# Wait for run task to complete
158+
await run_task
159+
160+
# Verify the expected calls were made
161+
hub.download_user_program.assert_called_once()
162+
hub.start_user_program.assert_called_once()
163+
finally:
164+
import os
165+
166+
os.unlink(temp_path)
167+

0 commit comments

Comments
 (0)