MicroPythonOS/tests/test_download_manager_utils.py

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