working

tests/component_tests/web_server/test_esp_idf_ota.py

@@ -0,0 +1,236 @@
+import asyncio
+import os
+import tempfile
+
+import aiohttp
+import pytest
+
+
+@pytest.fixture
+async def web_server_fixture(event_loop):
+    """Start the test device with web server"""
+    # This would be replaced with actual device setup in a real test environment
+    # For now, we'll assume the device is running at a specific address
+    base_url = "http://localhost:8080"
+
+    # Wait a bit for server to be ready
+    await asyncio.sleep(2)
+
+    yield base_url
+
+
+async def create_test_firmware():
+    """Create a dummy firmware file for testing"""
+    with tempfile.NamedTemporaryFile(suffix=".bin", delete=False) as f:
+        # Write some dummy data that looks like a firmware file
+        # ESP32 firmware files typically start with these magic bytes
+        f.write(b"\xe9\x08\x02\x20")  # ESP32 magic bytes
+        # Add some padding to make it look like a real firmware
+        f.write(b"\x00" * 1024)  # 1KB of zeros
+        f.write(b"TEST_FIRMWARE_CONTENT")
+        f.write(b"\x00" * 1024)  # More padding
+        return f.name
+
+
+@pytest.mark.asyncio
+async def test_ota_upload_multipart(web_server_fixture):
+    """Test OTA firmware upload using multipart/form-data"""
+    base_url = web_server_fixture
+    firmware_path = await create_test_firmware()
+
+    try:
+        # Create multipart form data
+        async with aiohttp.ClientSession() as session:
+            # First, check if OTA endpoint is available
+            async with session.get(f"{base_url}/") as resp:
+                assert resp.status == 200
+                content = await resp.text()
+                assert "ota" in content or "OTA" in content
+
+            # Prepare multipart upload
+            with open(firmware_path, "rb") as f:
+                data = aiohttp.FormData()
+                data.add_field(
+                    "firmware",
+                    f,
+                    filename="firmware.bin",
+                    content_type="application/octet-stream",
+                )
+
+                # Send OTA update request
+                async with session.post(f"{base_url}/ota/upload", data=data) as resp:
+                    assert resp.status in [200, 201, 204], (
+                        f"OTA upload failed with status {resp.status}"
+                    )
+
+                    # Check response
+                    if resp.status == 200:
+                        response_text = await resp.text()
+                        # The response might be JSON or plain text depending on implementation
+                        assert (
+                            "success" in response_text.lower()
+                            or "ok" in response_text.lower()
+                        )
+
+    finally:
+        # Clean up
+        os.unlink(firmware_path)
+
+
+@pytest.mark.asyncio
+async def test_ota_upload_wrong_content_type(web_server_fixture):
+    """Test that OTA upload fails with wrong content type"""
+    base_url = web_server_fixture
+
+    async with aiohttp.ClientSession() as session:
+        # Try to upload with wrong content type
+        data = b"not a firmware file"
+        headers = {"Content-Type": "text/plain"}
+
+        async with session.post(
+            f"{base_url}/ota/upload", data=data, headers=headers
+        ) as resp:
+            # Should fail with bad request or similar
+            assert resp.status >= 400, f"Expected error status, got {resp.status}"
+
+
+@pytest.mark.asyncio
+async def test_ota_upload_empty_file(web_server_fixture):
+    """Test that OTA upload fails with empty file"""
+    base_url = web_server_fixture
+
+    async with aiohttp.ClientSession() as session:
+        # Create empty multipart upload
+        data = aiohttp.FormData()
+        data.add_field(
+            "firmware",
+            b"",
+            filename="empty.bin",
+            content_type="application/octet-stream",
+        )
+
+        async with session.post(f"{base_url}/ota/upload", data=data) as resp:
+            # Should fail with bad request
+            assert resp.status >= 400, (
+                f"Expected error status for empty file, got {resp.status}"
+            )
+
+
+@pytest.mark.asyncio
+async def test_ota_multipart_boundary_parsing(web_server_fixture):
+    """Test multipart boundary parsing edge cases"""
+    base_url = web_server_fixture
+    firmware_path = await create_test_firmware()
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            # Test with custom boundary
+            with open(firmware_path, "rb") as f:
+                # Create multipart manually with specific boundary
+                boundary = "----WebKitFormBoundaryCustomTest123"
+                body = (
+                    f"--{boundary}\r\n"
+                    f'Content-Disposition: form-data; name="firmware"; filename="test.bin"\r\n'
+                    f"Content-Type: application/octet-stream\r\n"
+                    f"\r\n"
+                ).encode()
+                body += f.read()
+                body += f"\r\n--{boundary}--\r\n".encode()
+
+                headers = {
+                    "Content-Type": f"multipart/form-data; boundary={boundary}",
+                    "Content-Length": str(len(body)),
+                }
+
+                async with session.post(
+                    f"{base_url}/ota/upload", data=body, headers=headers
+                ) as resp:
+                    assert resp.status in [200, 201, 204], (
+                        f"Custom boundary upload failed with status {resp.status}"
+                    )
+
+    finally:
+        os.unlink(firmware_path)
+
+
+@pytest.mark.asyncio
+async def test_ota_concurrent_uploads(web_server_fixture):
+    """Test that concurrent OTA uploads are properly handled"""
+    base_url = web_server_fixture
+    firmware_path = await create_test_firmware()
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            # Create two concurrent upload tasks
+            async def upload_firmware():
+                with open(firmware_path, "rb") as f:
+                    data = aiohttp.FormData()
+                    data.add_field(
+                        "firmware",
+                        f.read(),  # Read to bytes to avoid file conflicts
+                        filename="firmware.bin",
+                        content_type="application/octet-stream",
+                    )
+
+                    async with session.post(
+                        f"{base_url}/ota/upload", data=data
+                    ) as resp:
+                        return resp.status
+
+            # Start two uploads concurrently
+            results = await asyncio.gather(
+                upload_firmware(), upload_firmware(), return_exceptions=True
+            )
+
+            # One should succeed, the other should fail with conflict
+            statuses = [r for r in results if isinstance(r, int)]
+            assert len(statuses) == 2
+            assert 200 in statuses or 201 in statuses or 204 in statuses
+            # The other might be 409 Conflict or similar
+
+    finally:
+        os.unlink(firmware_path)
+
+
+@pytest.mark.asyncio
+async def test_ota_large_file_upload(web_server_fixture):
+    """Test OTA upload with a larger file to test chunked processing"""
+    base_url = web_server_fixture
+
+    # Create a larger test firmware (1MB)
+    with tempfile.NamedTemporaryFile(suffix=".bin", delete=False) as f:
+        # ESP32 magic bytes
+        f.write(b"\xe9\x08\x02\x20")
+        # Write 1MB of data in chunks
+        chunk_size = 4096
+        for _ in range(256):  # 256 * 4KB = 1MB
+            f.write(b"A" * chunk_size)
+        firmware_path = f.name
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            with open(firmware_path, "rb") as f:
+                data = aiohttp.FormData()
+                data.add_field(
+                    "firmware",
+                    f,
+                    filename="large_firmware.bin",
+                    content_type="application/octet-stream",
+                )
+
+                # Use a longer timeout for large file
+                timeout = aiohttp.ClientTimeout(total=60)
+                async with session.post(
+                    f"{base_url}/ota/upload", data=data, timeout=timeout
+                ) as resp:
+                    assert resp.status in [200, 201, 204], (
+                        f"Large file OTA upload failed with status {resp.status}"
+                    )
+
+    finally:
+        os.unlink(firmware_path)
+
+
+if __name__ == "__main__":
+    # For manual testing
+    asyncio.run(test_ota_upload_multipart(asyncio.Event()))

tests/components/web_server/test_multipart_ota.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""
+Test script for ESP-IDF web server multipart OTA upload functionality.
+This script can be run manually to test OTA uploads to a running device.
+"""
+
+import argparse
+from pathlib import Path
+import sys
+import time
+
+import requests
+
+
+def test_multipart_ota_upload(host, port, firmware_path):
+    """Test OTA firmware upload using multipart/form-data"""
+    base_url = f"http://{host}:{port}"
+
+    print(f"Testing OTA upload to {base_url}")
+
+    # First check if server is reachable
+    try:
+        resp = requests.get(f"{base_url}/", timeout=5)
+        if resp.status_code != 200:
+            print(f"Error: Server returned status {resp.status_code}")
+            return False
+        print("✓ Server is reachable")
+    except requests.exceptions.RequestException as e:
+        print(f"Error: Cannot reach server - {e}")
+        return False
+
+    # Check if firmware file exists
+    if not Path(firmware_path).exists():
+        print(f"Error: Firmware file not found: {firmware_path}")
+        return False
+
+    # Prepare multipart upload
+    print(f"Uploading firmware: {firmware_path}")
+    print(f"File size: {Path(firmware_path).stat().st_size} bytes")
+
+    try:
+        with open(firmware_path, "rb") as f:
+            files = {"firmware": ("firmware.bin", f, "application/octet-stream")}
+
+            # Send OTA update request
+            resp = requests.post(f"{base_url}/ota/upload", files=files, timeout=60)
+
+            if resp.status_code in [200, 201, 204]:
+                print(f"✓ OTA upload successful (status: {resp.status_code})")
+                if resp.text:
+                    print(f"Response: {resp.text}")
+                return True
+            else:
+                print(f"✗ OTA upload failed with status {resp.status_code}")
+                print(f"Response: {resp.text}")
+                return False
+
+    except requests.exceptions.RequestException as e:
+        print(f"Error during upload: {e}")
+        return False
+
+
+def test_ota_with_wrong_content_type(host, port):
+    """Test that OTA upload fails gracefully with wrong content type"""
+    base_url = f"http://{host}:{port}"
+
+    print("\nTesting OTA with wrong content type...")
+
+    try:
+        # Send plain text instead of multipart
+        headers = {"Content-Type": "text/plain"}
+        resp = requests.post(
+            f"{base_url}/ota/upload",
+            data="This is not a firmware file",
+            headers=headers,
+            timeout=10,
+        )
+
+        if resp.status_code >= 400:
+            print(
+                f"✓ Server correctly rejected wrong content type (status: {resp.status_code})"
+            )
+            return True
+        else:
+            print(f"✗ Server accepted wrong content type (status: {resp.status_code})")
+            return False
+
+    except requests.exceptions.RequestException as e:
+        print(f"Error: {e}")
+        return False
+
+
+def test_ota_with_empty_file(host, port):
+    """Test that OTA upload fails gracefully with empty file"""
+    base_url = f"http://{host}:{port}"
+
+    print("\nTesting OTA with empty file...")
+
+    try:
+        # Send empty file
+        files = {"firmware": ("empty.bin", b"", "application/octet-stream")}
+        resp = requests.post(f"{base_url}/ota/upload", files=files, timeout=10)
+
+        if resp.status_code >= 400:
+            print(
+                f"✓ Server correctly rejected empty file (status: {resp.status_code})"
+            )
+            return True
+        else:
+            print(f"✗ Server accepted empty file (status: {resp.status_code})")
+            return False
+
+    except requests.exceptions.RequestException as e:
+        print(f"Error: {e}")
+        return False
+
+
+def create_test_firmware(size_kb=10):
+    """Create a dummy firmware file for testing"""
+    import tempfile
+
+    with tempfile.NamedTemporaryFile(suffix=".bin", delete=False) as f:
+        # ESP32 firmware magic bytes
+        f.write(b"\xe9\x08\x02\x20")
+        # Add padding
+        f.write(b"\x00" * (size_kb * 1024 - 4))
+        return f.name
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Test ESP-IDF web server OTA functionality"
+    )
+    parser.add_argument("--host", default="localhost", help="Device hostname or IP")
+    parser.add_argument("--port", type=int, default=8080, help="Web server port")
+    parser.add_argument(
+        "--firmware", help="Path to firmware file (if not specified, creates test file)"
+    )
+    parser.add_argument(
+        "--skip-error-tests", action="store_true", help="Skip error condition tests"
+    )
+
+    args = parser.parse_args()
+
+    # Create test firmware if not specified
+    firmware_path = args.firmware
+    if not firmware_path:
+        print("Creating test firmware file...")
+        firmware_path = create_test_firmware(100)  # 100KB test file
+        print(f"Created test firmware: {firmware_path}")
+
+    all_passed = True
+
+    # Test successful OTA upload
+    if not test_multipart_ota_upload(args.host, args.port, firmware_path):
+        all_passed = False
+
+    # Test error conditions
+    if not args.skip_error_tests:
+        time.sleep(1)  # Small delay between tests
+
+        if not test_ota_with_wrong_content_type(args.host, args.port):
+            all_passed = False
+
+        time.sleep(1)
+
+        if not test_ota_with_empty_file(args.host, args.port):
+            all_passed = False
+
+    # Clean up test firmware if we created it
+    if not args.firmware:
+        import os
+
+        os.unlink(firmware_path)
+        print("\nCleaned up test firmware")
+
+    print(f"\n{'All tests passed!' if all_passed else 'Some tests failed!'}")
+    return 0 if all_passed else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/components/web_server/test_ota.esp32-idf.yaml

@@ -1,12 +1,33 @@
+# Test configuration for ESP-IDF web server with OTA enabled
+esphome:
+  name: test-web-server-ota-idf
+
+# Force ESP-IDF framework
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
 packages:
   device_base: !include common.yaml
 
-# Enable OTA for this test
+# Enable OTA for multipart upload testing
 ota:
   - platform: esphome
     safe_mode: true
+    password: "test_ota_password"
 
+# Web server with OTA enabled
 web_server:
   port: 8080
   version: 2
   ota: true
+  include_internal: true
+
+# Enable debug logging for OTA
+logger:
+  level: DEBUG
+  logs:
+    web_server: VERBOSE
+    web_server_idf: VERBOSE
+

tests/components/web_server/test_ota_readme.md

@@ -0,0 +1,70 @@
+# Testing ESP-IDF Web Server OTA Functionality
+
+This directory contains tests for the ESP-IDF web server OTA (Over-The-Air) update functionality using multipart form uploads.
+
+## Test Files
+
+- `test_ota.esp32-idf.yaml` - ESPHome configuration with OTA enabled for ESP-IDF
+- `test_no_ota.esp32-idf.yaml` - ESPHome configuration with OTA disabled
+- `test_ota_disabled.esp32-idf.yaml` - ESPHome configuration with web_server ota: false
+- `test_multipart_ota.py` - Manual test script for OTA functionality
+- `test_esp_idf_ota.py` - Automated pytest for OTA functionality
+
+## Running the Tests
+
+### 1. Compile and Flash Test Device
+
+```bash
+# Compile the OTA-enabled configuration
+esphome compile tests/components/web_server/test_ota.esp32-idf.yaml
+
+# Flash to device
+esphome upload tests/components/web_server/test_ota.esp32-idf.yaml
+```
+
+### 2. Run Manual Tests
+
+Once the device is running, you can test OTA functionality:
+
+```bash
+# Test with default settings (creates test firmware)
+python tests/components/web_server/test_multipart_ota.py --host <device-ip>
+
+# Test with real firmware file
+python tests/components/web_server/test_multipart_ota.py --host <device-ip> --firmware <path-to-firmware.bin>
+
+# Skip error condition tests (useful for production devices)
+python tests/components/web_server/test_multipart_ota.py --host <device-ip> --skip-error-tests
+```
+
+### 3. Run Automated Tests
+
+```bash
+# Run pytest suite
+pytest tests/component_tests/web_server/test_esp_idf_ota.py
+```
+
+## What's Being Tested
+
+1. **Multipart Upload**: Tests that firmware can be uploaded using multipart/form-data
+2. **Error Handling**: 
+   - Wrong content type rejection
+   - Empty file rejection
+   - Concurrent upload handling
+3. **Large Files**: Tests chunked processing of larger firmware files
+4. **Boundary Parsing**: Tests various multipart boundary formats
+
+## Implementation Details
+
+The ESP-IDF web server uses the `multipart-parser` library to handle multipart uploads. Key components:
+
+- `MultipartReader` class for parsing multipart data
+- Chunked processing to handle large files without excessive memory use
+- Integration with ESPHome's OTA component for actual firmware updates
+
+## Troubleshooting
+
+1. **Connection Refused**: Make sure the device is on the network and the IP is correct
+2. **404 Not Found**: Ensure OTA is enabled in the configuration (`ota: true` in web_server)
+3. **Upload Fails**: Check device logs for detailed error messages
+4. **Timeout**: Large firmware files may take time, increase timeout if needed