forked from MicroPythonOS/MicroPythonOS
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_download_manager_utils.py
More file actions
203 lines (166 loc) · 7.89 KB
/
test_download_manager_utils.py
File metadata and controls
203 lines (166 loc) · 7.89 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
"""
Unit tests for DownloadManager utility functions.
Tests the network error detection and resume position helpers.
"""
import unittest
import os
import sys
# Handle both CPython and MicroPython path handling
try:
# CPython has os.path
from os.path import join, dirname
except ImportError:
# MicroPython doesn't have os.path, use string concatenation
def join(*parts):
return '/'.join(parts)
def dirname(path):
parts = path.split('/')
return '/'.join(parts[:-1]) if len(parts) > 1 else '.'
# Add parent directory to path for imports
sys.path.insert(0, join(dirname(__file__), '..', 'internal_filesystem', 'lib'))
# Import functions directly from the module file to avoid mpos.__init__ dependencies
try:
import importlib.util
spec = importlib.util.spec_from_file_location(
"download_manager",
join(dirname(__file__), '..', 'internal_filesystem', 'lib', 'mpos', 'net', 'download_manager.py')
)
download_manager = importlib.util.module_from_spec(spec)
spec.loader.exec_module(download_manager)
except (ImportError, AttributeError):
# MicroPython doesn't have importlib.util, import directly
sys.path.insert(0, join(dirname(__file__), '..', 'internal_filesystem', 'lib', 'mpos', 'net'))
import download_manager
is_network_error = download_manager.is_network_error
get_resume_position = download_manager.get_resume_position
class TestIsNetworkError(unittest.TestCase):
"""Test network error detection utility."""
def test_detects_timeout_error_code(self):
"""Should detect OSError with -110 (ETIMEDOUT) as network error."""
error = OSError(-110, "Connection timed out")
self.assertTrue(is_network_error(error))
def test_detects_connection_aborted_error_code(self):
"""Should detect OSError with -113 (ECONNABORTED) as network error."""
error = OSError(-113, "Connection aborted")
self.assertTrue(is_network_error(error))
def test_detects_connection_reset_error_code(self):
"""Should detect OSError with -104 (ECONNRESET) as network error."""
error = OSError(-104, "Connection reset by peer")
self.assertTrue(is_network_error(error))
def test_detects_host_unreachable_error_code(self):
"""Should detect OSError with -118 (EHOSTUNREACH) as network error."""
error = OSError(-118, "No route to host")
self.assertTrue(is_network_error(error))
def test_detects_dns_error_code(self):
"""Should detect OSError with -202 (DNS/connection error) as network error."""
error = OSError(-202, "DNS lookup failed")
self.assertTrue(is_network_error(error))
def test_detects_connection_reset_message(self):
"""Should detect 'connection reset' in error message."""
error = Exception("Connection reset by peer")
self.assertTrue(is_network_error(error))
def test_detects_connection_aborted_message(self):
"""Should detect 'connection aborted' in error message."""
error = Exception("Connection aborted")
self.assertTrue(is_network_error(error))
def test_detects_broken_pipe_message(self):
"""Should detect 'broken pipe' in error message."""
error = Exception("Broken pipe")
self.assertTrue(is_network_error(error))
def test_detects_network_unreachable_message(self):
"""Should detect 'network unreachable' in error message."""
error = Exception("Network unreachable")
self.assertTrue(is_network_error(error))
def test_detects_failed_to_download_chunk_message(self):
"""Should detect 'failed to download chunk' message from download_manager."""
error = OSError(-110, "Failed to download chunk after retries")
self.assertTrue(is_network_error(error))
def test_rejects_value_error(self):
"""Should not detect ValueError as network error."""
error = ValueError("Invalid value")
self.assertFalse(is_network_error(error))
def test_rejects_http_404_error(self):
"""Should not detect HTTP 404 as network error."""
error = RuntimeError("HTTP 404")
self.assertFalse(is_network_error(error))
def test_rejects_file_not_found_error(self):
"""Should not detect ENOENT (-2) as network error."""
error = OSError(-2, "No such file or directory")
self.assertFalse(is_network_error(error))
def test_rejects_permission_error(self):
"""Should not detect permission errors as network error."""
error = OSError(-13, "Permission denied")
self.assertFalse(is_network_error(error))
def test_case_insensitive_detection(self):
"""Should detect network errors regardless of case."""
error1 = Exception("CONNECTION RESET")
error2 = Exception("connection reset")
error3 = Exception("Connection Reset")
self.assertTrue(is_network_error(error1))
self.assertTrue(is_network_error(error2))
self.assertTrue(is_network_error(error3))
class TestGetResumePosition(unittest.TestCase):
"""Test resume position utility."""
def setUp(self):
"""Create test directory."""
self.test_dir = "tmp_test_download_manager"
# Handle both CPython and MicroPython
try:
os.makedirs(self.test_dir, exist_ok=True)
except (AttributeError, TypeError):
# MicroPython doesn't have makedirs or exist_ok parameter
try:
os.mkdir(self.test_dir)
except OSError:
pass # Directory already exists
def tearDown(self):
"""Clean up test files."""
# Handle both CPython and MicroPython
try:
import shutil
if os.path.exists(self.test_dir):
shutil.rmtree(self.test_dir)
except (ImportError, AttributeError):
# MicroPython doesn't have shutil, manually remove files
try:
import os as os_module
for f in os_module.listdir(self.test_dir):
os_module.remove(join(self.test_dir, f))
os_module.rmdir(self.test_dir)
except (OSError, AttributeError):
pass # Ignore errors during cleanup
def test_returns_zero_for_nonexistent_file(self):
"""Should return 0 for files that don't exist."""
nonexistent = join(self.test_dir, "nonexistent.bin")
self.assertEqual(get_resume_position(nonexistent), 0)
def test_returns_file_size_for_existing_file(self):
"""Should return file size for existing files."""
test_file = join(self.test_dir, "test.bin")
test_data = b"x" * 1024
with open(test_file, "wb") as f:
f.write(test_data)
self.assertEqual(get_resume_position(test_file), 1024)
def test_returns_zero_for_empty_file(self):
"""Should return 0 for empty files."""
test_file = join(self.test_dir, "empty.bin")
with open(test_file, "wb") as f:
pass # Create empty file
self.assertEqual(get_resume_position(test_file), 0)
def test_returns_correct_size_for_large_file(self):
"""Should return correct size for larger files."""
test_file = join(self.test_dir, "large.bin")
test_data = b"x" * (1024 * 1024) # 1 MB (reduced from 10 MB to avoid memory issues)
with open(test_file, "wb") as f:
f.write(test_data)
self.assertEqual(get_resume_position(test_file), 1024 * 1024)
def test_returns_size_after_partial_write(self):
"""Should return current size after partial write."""
test_file = join(self.test_dir, "partial.bin")
# Write 1KB
with open(test_file, "wb") as f:
f.write(b"x" * 1024)
self.assertEqual(get_resume_position(test_file), 1024)
# Append another 1KB
with open(test_file, "ab") as f:
f.write(b"y" * 1024)
self.assertEqual(get_resume_position(test_file), 2048)