""" 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)