MicroPythonOS/tests/test_download_manager.py

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