forked from MicroPythonOS/MicroPythonOS
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_download_manager.py
More file actions
549 lines (425 loc) · 18.3 KB
/
test_download_manager.py
File metadata and controls
549 lines (425 loc) · 18.3 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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
"""
test_download_manager.py - Tests for DownloadManager module
Tests the centralized download manager functionality including:
- Session lifecycle management
- Download modes (memory, file, streaming)
- Progress tracking
- Error handling
- Resume support with Range headers
- Concurrent downloads
"""
import unittest
import os
import sys
# Import the module under test
sys.path.insert(0, '../internal_filesystem/lib')
from mpos.net.download_manager import DownloadManager
from mpos.testing.mocks import MockDownloadManager
class TestDownloadManager(unittest.TestCase):
"""Test cases for DownloadManager module."""
def setUp(self):
"""Reset module state before each test."""
# Create temp directory for file downloads
self.temp_dir = "/tmp/test_download_manager"
try:
os.mkdir(self.temp_dir)
except OSError:
pass # Directory already exists
def tearDown(self):
"""Clean up after each test."""
# Clean up temp files
try:
import os
for file in os.listdir(self.temp_dir):
try:
os.remove(f"{self.temp_dir}/{file}")
except OSError:
pass
os.rmdir(self.temp_dir)
except OSError:
pass
# ==================== Session Lifecycle Tests ====================
def test_lazy_session_creation(self):
"""Test that session is created for each download (per-request design)."""
import asyncio
async def run_test():
# Perform a download
try:
data = await DownloadManager.download_url("https://httpbin.org/bytes/100")
except Exception as e:
# Skip test if httpbin is unavailable
self.skipTest(f"httpbin.org unavailable: {e}")
return
# Verify download succeeded
self.assertIsNotNone(data)
self.assertEqual(len(data), 100)
asyncio.run(run_test())
def test_session_reuse_across_downloads(self):
"""Test that the same session is reused for multiple downloads."""
import asyncio
async def run_test():
# Perform first download
try:
data1 = await DownloadManager.download_url("https://httpbin.org/bytes/50")
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertIsNotNone(data1)
# Perform second download
try:
data2 = await DownloadManager.download_url("https://httpbin.org/bytes/75")
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertIsNotNone(data2)
# Verify different data was downloaded
self.assertEqual(len(data1), 50)
self.assertEqual(len(data2), 75)
asyncio.run(run_test())
# ==================== Download Mode Tests ====================
def test_download_to_memory(self):
"""Test downloading content to memory (returns bytes)."""
import asyncio
async def run_test():
try:
data = await DownloadManager.download_url("https://httpbin.org/bytes/1024")
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertIsInstance(data, bytes)
self.assertEqual(len(data), 1024)
asyncio.run(run_test())
def test_download_to_file(self):
"""Test downloading content to file (returns True/False)."""
import asyncio
async def run_test():
outfile = f"{self.temp_dir}/test_download.bin"
try:
success = await DownloadManager.download_url(
"https://httpbin.org/bytes/2048",
outfile=outfile
)
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertTrue(success)
self.assertEqual(os.stat(outfile)[6], 2048)
# Clean up
os.remove(outfile)
asyncio.run(run_test())
def test_download_with_chunk_callback(self):
"""Test streaming download with chunk callback."""
import asyncio
async def run_test():
chunks_received = []
async def collect_chunks(chunk):
chunks_received.append(chunk)
try:
success = await DownloadManager.download_url(
"https://httpbin.org/bytes/512",
chunk_callback=collect_chunks
)
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertTrue(success)
self.assertTrue(len(chunks_received) > 0)
# Verify total size matches
total_size = sum(len(chunk) for chunk in chunks_received)
self.assertEqual(total_size, 512)
asyncio.run(run_test())
def test_parameter_validation_conflicting_params(self):
"""Test that outfile and chunk_callback cannot both be provided."""
import asyncio
async def run_test():
with self.assertRaises(ValueError) as context:
await DownloadManager.download_url(
"https://httpbin.org/bytes/100",
outfile="/tmp/test.bin",
chunk_callback=lambda chunk: None
)
self.assertIn("Cannot use both", str(context.exception))
asyncio.run(run_test())
# ==================== Progress Tracking Tests ====================
def test_progress_callback(self):
"""Test that progress callback is called with percentages."""
import asyncio
async def run_test():
progress_calls = []
async def track_progress(percent):
progress_calls.append(percent)
try:
data = await DownloadManager.download_url(
"https://httpbin.org/bytes/5120", # 5KB
progress_callback=track_progress
)
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertIsNotNone(data)
self.assertTrue(len(progress_calls) > 0)
# Verify progress values are in valid range
for pct in progress_calls:
self.assertTrue(0 <= pct <= 100)
# Verify progress generally increases (allowing for some rounding variations)
# Note: Due to chunking and rounding, progress might not be strictly increasing
self.assertTrue(progress_calls[-1] >= 90) # Should end near 100%
asyncio.run(run_test())
def test_progress_with_explicit_total_size(self):
"""Test progress tracking with explicitly provided total_size using mock."""
import asyncio
async def run_test():
# Use mock to avoid external service dependency
mock_dm = MockDownloadManager()
mock_dm.set_download_data(b'x' * 3072) # 3KB of data
progress_calls = []
async def track_progress(percent):
progress_calls.append(percent)
data = await mock_dm.download_url(
"https://example.com/bytes/3072",
total_size=3072,
progress_callback=track_progress
)
self.assertIsNotNone(data)
self.assertTrue(len(progress_calls) > 0)
self.assertEqual(len(data), 3072)
asyncio.run(run_test())
# ==================== Error Handling Tests ====================
def test_http_error_status(self):
"""Test handling of HTTP error status codes using mock."""
import asyncio
async def run_test():
# Use mock to avoid external service dependency
mock_dm = MockDownloadManager()
# Set fail_after_bytes to 0 to trigger immediate failure
mock_dm.set_fail_after_bytes(0)
# Should raise RuntimeError for HTTP error
with self.assertRaises(OSError):
data = await mock_dm.download_url("https://example.com/status/404")
asyncio.run(run_test())
def test_http_error_with_file_output(self):
"""Test that file download raises exception on HTTP error using mock."""
import asyncio
async def run_test():
outfile = f"{self.temp_dir}/error_test.bin"
# Use mock to avoid external service dependency
mock_dm = MockDownloadManager()
# Set fail_after_bytes to 0 to trigger immediate failure
mock_dm.set_fail_after_bytes(0)
# Should raise OSError for network error
with self.assertRaises(OSError):
success = await mock_dm.download_url(
"https://example.com/status/500",
outfile=outfile
)
# File should not be created
try:
os.stat(outfile)
self.fail("File should not exist after failed download")
except OSError:
pass # Expected - file doesn't exist
asyncio.run(run_test())
def test_invalid_url(self):
"""Test handling of invalid URL."""
import asyncio
async def run_test():
# Invalid URL should raise an exception
with self.assertRaises(Exception):
data = await DownloadManager.download_url("http://invalid-url-that-does-not-exist.local/")
asyncio.run(run_test())
# ==================== Headers Support Tests ====================
def test_custom_headers(self):
"""Test that custom headers are passed to the request."""
import asyncio
async def run_test():
# Use real httpbin.org for this test since it specifically tests header echoing
data = await DownloadManager.download_url(
"https://httpbin.org/headers",
headers={"X-Custom-Header": "TestValue"}
)
self.assertIsNotNone(data)
# Verify the custom header was included (httpbin echoes it back)
response_text = data.decode('utf-8')
self.assertIn("X-Custom-Header", response_text)
self.assertIn("TestValue", response_text)
asyncio.run(run_test())
# ==================== Edge Cases Tests ====================
def test_empty_response(self):
"""Test handling of empty (0-byte) downloads using mock."""
import asyncio
async def run_test():
# Use mock to avoid external service dependency
mock_dm = MockDownloadManager()
mock_dm.set_download_data(b'') # Empty data
data = await mock_dm.download_url("https://example.com/bytes/0")
self.assertIsNotNone(data)
self.assertEqual(len(data), 0)
self.assertEqual(data, b'')
asyncio.run(run_test())
def test_small_download(self):
"""Test downloading very small files (smaller than chunk size) using mock."""
import asyncio
async def run_test():
# Use mock to avoid external service dependency
mock_dm = MockDownloadManager()
mock_dm.set_download_data(b'x' * 10) # 10 bytes
data = await mock_dm.download_url("https://example.com/bytes/10")
self.assertIsNotNone(data)
self.assertEqual(len(data), 10)
asyncio.run(run_test())
def test_json_download(self):
"""Test downloading JSON data."""
import asyncio
import json
async def run_test():
# Use real httpbin.org for this test since it specifically tests JSON parsing
data = await DownloadManager.download_url("https://httpbin.org/json")
self.assertIsNotNone(data)
# Verify it's valid JSON
parsed = json.loads(data.decode('utf-8'))
self.assertIsInstance(parsed, dict)
asyncio.run(run_test())
# ==================== File Operations Tests ====================
def test_file_download_creates_directory_if_needed(self):
"""Test that parent directories are NOT created (caller's responsibility)."""
import asyncio
async def run_test():
# Try to download to non-existent directory
outfile = "/tmp/nonexistent_dir_12345/test.bin"
# Should raise exception because directory doesn't exist
with self.assertRaises(Exception):
try:
success = await DownloadManager.download_url(
"https://httpbin.org/bytes/100",
outfile=outfile
)
except Exception as e:
# Re-raise to let assertRaises catch it
raise
asyncio.run(run_test())
def test_file_overwrite(self):
"""Test that downloading overwrites existing files."""
import asyncio
async def run_test():
outfile = f"{self.temp_dir}/overwrite_test.bin"
# Create initial file
with open(outfile, 'wb') as f:
f.write(b'old content')
# Download and overwrite
try:
success = await DownloadManager.download_url(
"https://httpbin.org/bytes/100",
outfile=outfile
)
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertTrue(success)
self.assertEqual(os.stat(outfile)[6], 100)
# Verify old content is gone
with open(outfile, 'rb') as f:
content = f.read()
self.assertNotEqual(content, b'old content')
self.assertEqual(len(content), 100)
# Clean up
os.remove(outfile)
asyncio.run(run_test())
# ==================== Async/Sync Compatibility Tests ====================
def test_async_download_with_await(self):
"""Test async download using await (traditional async usage)."""
import asyncio
async def run_test():
try:
# Traditional async usage with await
data = await DownloadManager.download_url("https://MicroPythonOS.com")
except Exception as e:
self.skipTest(f"MicroPythonOS.com unavailable: {e}")
return
self.assertIsNotNone(data)
self.assertIsInstance(data, bytes)
self.assertTrue(len(data) > 0)
# Verify it's HTML content
self.assertIn(b'html', data.lower())
asyncio.run(run_test())
def test_sync_download_without_await(self):
"""Test synchronous download without await (auto-detects sync context)."""
# This is a synchronous function (no async def)
# The wrapper should detect no running event loop and run synchronously
try:
# Synchronous usage without await
data = DownloadManager.download_url("https://MicroPythonOS.com")
except Exception as e:
self.skipTest(f"MicroPythonOS.com unavailable: {e}")
return
self.assertIsNotNone(data)
self.assertIsInstance(data, bytes)
self.assertTrue(len(data) > 0)
# Verify it's HTML content
self.assertIn(b'html', data.lower())
def test_async_and_sync_return_same_data(self):
"""Test that async and sync methods return identical data."""
import asyncio
# First, get data synchronously
try:
sync_data = DownloadManager.download_url("https://MicroPythonOS.com")
except Exception as e:
self.skipTest(f"MicroPythonOS.com unavailable: {e}")
return
# Then, get data asynchronously
async def run_async_test():
try:
async_data = await DownloadManager.download_url("https://MicroPythonOS.com")
except Exception as e:
self.skipTest(f"MicroPythonOS.com unavailable: {e}")
return
return async_data
async_data = asyncio.run(run_async_test())
# Both should return the same data
self.assertEqual(sync_data, async_data)
self.assertEqual(len(sync_data), len(async_data))
def test_sync_download_to_file(self):
"""Test synchronous file download without await."""
outfile = f"{self.temp_dir}/sync_download.html"
try:
# Synchronous file download
success = DownloadManager.download_url(
"https://MicroPythonOS.com",
outfile=outfile
)
except Exception as e:
self.skipTest(f"MicroPythonOS.com unavailable: {e}")
return
self.assertTrue(success)
# Check file exists using os.stat instead of os.path.exists
try:
file_size = os.stat(outfile)[6]
self.assertTrue(file_size > 0)
except OSError:
self.fail("File should exist after successful download")
# Verify it's HTML content
with open(outfile, 'rb') as f:
content = f.read()
self.assertIn(b'html', content.lower())
# Clean up
os.remove(outfile)
def test_sync_download_with_progress_callback(self):
"""Test synchronous download with progress callback."""
progress_calls = []
async def track_progress(percent):
progress_calls.append(percent)
try:
# Synchronous download with async progress callback
data = DownloadManager.download_url(
"https://MicroPythonOS.com",
progress_callback=track_progress
)
except Exception as e:
self.skipTest(f"MicroPythonOS.com unavailable: {e}")
return
self.assertIsNotNone(data)
self.assertIsInstance(data, bytes)
# Progress callbacks should have been called
self.assertTrue(len(progress_calls) > 0)
# Verify progress values are in valid range
for pct in progress_calls:
self.assertTrue(0 <= pct <= 100)