11"""Tests for the pybricks connection module."""
22
33import asyncio
4+ import contextlib
5+ import os
46import tempfile
5- from unittest .mock import AsyncMock , PropertyMock
7+ from unittest .mock import AsyncMock , PropertyMock , patch
68
79import pytest
810from reactivex .subject import Subject
@@ -32,16 +34,17 @@ async def test_download_modern_protocol(self):
3234 )
3335 hub ._capability_flags = HubCapabilityFlag .USER_PROG_MULTI_FILE_MPY6
3436
35- with tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False ) as temp :
37+ with contextlib .ExitStack () as stack :
38+ # Create and manage temporary file
39+ temp = stack .enter_context (
40+ tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False )
41+ )
3642 temp .write ("print('test')" )
3743 temp_path = temp .name
38- try :
44+ stack .callback (os .unlink , temp_path )
45+
3946 await hub .download (temp_path )
4047 hub .download_user_program .assert_called_once ()
41- finally :
42- import os
43-
44- os .unlink (temp_path )
4548
4649 @pytest .mark .asyncio
4750 async def test_download_legacy_firmware (self ):
@@ -56,16 +59,17 @@ async def test_download_legacy_firmware(self):
5659 )
5760 hub ._capability_flags = HubCapabilityFlag .USER_PROG_MULTI_FILE_MPY6
5861
59- with tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False ) as temp :
62+ with contextlib .ExitStack () as stack :
63+ # Create and manage temporary file
64+ temp = stack .enter_context (
65+ tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False )
66+ )
6067 temp .write ("print('test')" )
6168 temp_path = temp .name
62- try :
69+ stack .callback (os .unlink , temp_path )
70+
6371 await hub .download (temp_path )
6472 hub .download_user_program .assert_called_once ()
65- finally :
66- import os
67-
68- os .unlink (temp_path )
6973
7074 @pytest .mark .asyncio
7175 async def test_download_unsupported_capabilities (self ):
@@ -79,19 +83,20 @@ async def test_download_unsupported_capabilities(self):
7983 )
8084 hub ._capability_flags = 0
8185
82- with tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False ) as temp :
86+ with contextlib .ExitStack () as stack :
87+ # Create and manage temporary file
88+ temp = stack .enter_context (
89+ tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False )
90+ )
8391 temp .write ("print('test')" )
8492 temp_path = temp .name
85- try :
93+ stack .callback (os .unlink , temp_path )
94+
8695 with pytest .raises (
8796 RuntimeError ,
8897 match = "Hub is not compatible with any of the supported file formats" ,
8998 ):
9099 await hub .download (temp_path )
91- finally :
92- import os
93-
94- os .unlink (temp_path )
95100
96101 @pytest .mark .asyncio
97102 async def test_download_compile_error (self ):
@@ -104,17 +109,27 @@ async def test_download_compile_error(self):
104109 return_value = ConnectionState .CONNECTED
105110 )
106111 hub ._capability_flags = HubCapabilityFlag .USER_PROG_MULTI_FILE_MPY6
112+ hub ._max_user_program_size = 1000 # Set a reasonable size limit
107113
108- with tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False ) as temp :
114+ with contextlib .ExitStack () as stack :
115+ # Create and manage temporary file
116+ temp = stack .enter_context (
117+ tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False )
118+ )
109119 temp .write ("print('test' # Missing closing parenthesis" )
110120 temp_path = temp .name
111- try :
112- with pytest .raises (SyntaxError ):
113- await hub .download (temp_path )
114- finally :
115- import os
121+ stack .callback (os .unlink , temp_path )
122+
123+ # Mock compile_multi_file to raise SyntaxError
124+ stack .enter_context (
125+ patch (
126+ "pybricksdev.connections.pybricks.compile_multi_file" ,
127+ side_effect = SyntaxError ("invalid syntax" ),
128+ )
129+ )
116130
117- os .unlink (temp_path )
131+ with pytest .raises (SyntaxError , match = "invalid syntax" ):
132+ await hub .download (temp_path )
118133
119134 @pytest .mark .asyncio
120135 async def test_run_modern_protocol (self ):
@@ -139,10 +154,15 @@ async def test_run_modern_protocol(self):
139154 hub ._stdout_line_queue = asyncio .Queue ()
140155 hub ._enable_line_handler = True
141156
142- with tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False ) as temp :
157+ with contextlib .ExitStack () as stack :
158+ # Create and manage temporary file
159+ temp = stack .enter_context (
160+ tempfile .NamedTemporaryFile (suffix = ".py" , mode = "w+" , delete = False )
161+ )
143162 temp .write ("print('test')" )
144163 temp_path = temp .name
145- try :
164+ stack .callback (os .unlink , temp_path )
165+
146166 # Start the run task
147167 run_task = asyncio .create_task (hub .run (temp_path ))
148168
@@ -160,8 +180,3 @@ async def test_run_modern_protocol(self):
160180 # Verify the expected calls were made
161181 hub .download_user_program .assert_called_once ()
162182 hub .start_user_program .assert_called_once ()
163- finally :
164- import os
165-
166- os .unlink (temp_path )
167-
0 commit comments