m5/MicroPythonOS
0
0
Fork 0
mirror of https://github.com/m5stack/MicroPythonOS.git synced 2026-05-20 11:51:27 -07:00
Code Issues Packages Projects Releases Wiki Activity
Files
d7e49d04dcdb08cf353bc85fd43c6fdcf96bd135
MicroPythonOS/tests/test_download_manager.py
T
Thomas Farstrike 30b3764710 Harmonize frameworks 
All frameworks now follow the same singleton class pattern with class methods:

AudioFlinger (already had this pattern)
DownloadManager (refactored)
ConnectivityManager (refactored)
CameraManager (refactored)
SensorManager (refactored)
Pattern Structure:

class FrameworkName:
    _initialized = False
    _instance_data = {}

    @classmethod
    def init(cls, *args, **kwargs):
        """Initialize the framework"""
        cls._initialized = True
        # initialization logic

    @classmethod
    def is_available(cls):
        """Check if framework is available"""
        return cls._initialized

    @classmethod
    def method_name(cls, *args):
        """Framework methods as class methods"""
        # implementation

2. Standardized Imports in __init__.py
All frameworks are now imported consistently as classes:

from .content.package_manager import PackageManager
from .config import SharedPreferences
from .net.connectivity_manager import ConnectivityManager
from .net.wifi_service import WifiService
from .audio.audioflinger import AudioFlinger
from .net.download_manager import DownloadManager
from .task_manager import TaskManager
from .camera_manager import CameraManager
from .sensor_manager import SensorManager
3. Updated Board Initialization Files
Fixed imports in all board files to use the new class-based pattern:

linux.py
fri3d_2024.py
fri3d_2026.py
waveshare_esp32_s3_touch_lcd_2.py
4. Updated UI Components
Fixed topmenu.py to import SensorManager as a class instead of a module.

5. Benefits of This Harmonization
✅ Consistency: All frameworks follow the same pattern - no more mixing of module imports and class imports ✅ Simplicity: Single, clear way to use frameworks - always as classes with class methods ✅ Functionality: All frameworks work identically - init(), is_available(), and other methods are consistent ✅ Maintainability: New developers see one pattern to follow across all frameworks ✅ No Breaking Changes: Apps continue to work without modification (Quasi apps, Lightning Piggy, etc.)

6. Testing
All tests pass successfully, confirming:

Framework initialization works correctly
Board hardware detection functions properly
UI components render without errors
No regressions in existing functionality
The harmonization is complete and production-ready. All frameworks now provide a unified, predictable interface that's easy to understand and extend.
2026-01-23 15:31:47 +01:00

590 lines
20 KiB
Python
Raw Blame History

"""
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."""
# Close any open sessions
import asyncio
try:
asyncio.run(DownloadManager.close_session())
except Exception:
pass # Session might not be open
# 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 lazily on first download."""
import asyncio
async def run_test():
# Verify no session exists initially
self.assertFalse(DownloadManager.is_session_active())
# 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 session was created
# Note: Session may be closed immediately after download if refcount == 0
# So we can't reliably check is_session_active() here
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())
def test_explicit_session_close(self):
"""Test explicit session closure."""
import asyncio
async def run_test():
# Create session by downloading
try:
data = await DownloadManager.download_url("https://httpbin.org/bytes/10")
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertIsNotNone(data)
# Explicitly close session
await DownloadManager.close_session()
# Verify session is closed
self.assertFalse(DownloadManager.is_session_active())
# Verify new download recreates session
try:
data2 = await DownloadManager.download_url("https://httpbin.org/bytes/20")
except Exception as e:
self.skipTest(f"httpbin.org unavailable: {e}")
return
self.assertIsNotNone(data2)
self.assertEqual(len(data2), 20)
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)
self.assertTrue(os.path.exists(outfile))
# Verify file has content
file_size = os.stat(outfile)[6]
self.assertTrue(file_size > 0)
# 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)
Reference in New Issue View Git Blame Copy Permalink