diff --git a/examples/system/ota/advanced_https_ota/pytest_advanced_ota.py b/examples/system/ota/advanced_https_ota/pytest_advanced_ota.py index 04acfeb497..c61196646e 100644 --- a/examples/system/ota/advanced_https_ota/pytest_advanced_ota.py +++ b/examples/system/ota/advanced_https_ota/pytest_advanced_ota.py @@ -4,13 +4,11 @@ import http.server import multiprocessing import os import random -import socket import ssl import struct import subprocess import time -from typing import Callable -from typing import Optional +from collections.abc import Callable import pexpect import pytest @@ -53,14 +51,14 @@ def https_request_handler() -> Callable[..., http.server.BaseHTTPRequestHandler] if not self.wfile.closed: self.wfile.flush() self.wfile.close() - except socket.error: + except OSError: pass self.rfile.close() def handle(self) -> None: try: RangeRequestHandler.handle(self) - except socket.error: + except OSError: pass return RequestHandler @@ -111,7 +109,7 @@ def redirect_handler_factory(url: str) -> Callable[..., http.server.BaseHTTPRequ def handle(self) -> None: try: http.server.BaseHTTPRequestHandler.handle(self) - except socket.error: + except OSError: pass return RedirectHandler @@ -134,7 +132,7 @@ def start_redirect_server(ota_image_dir: str, server_ip: str, server_port: int, # Function to modify chip revisions in the app header def modify_chip_revision( - app_path: str, min_rev: Optional[int] = None, max_rev: Optional[int] = None, increment_min: bool = False + app_path: str, min_rev: int | None = None, max_rev: int | None = None, increment_min: bool = False ) -> None: """ Modify min_chip_rev_full and max_chip_rev_full in the app header. @@ -170,7 +168,7 @@ def modify_chip_revision( with open(app_path, 'r+b') as f: f.write(header) - except IOError as e: + except OSError as e: raise RuntimeError(f'Failed to modify app header: {e}') @@ -199,7 +197,7 @@ def test_examples_protocol_advanced_https_ota_example(dut: Dut) -> None: dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') dut.expect('Starting Advanced OTA example', timeout=30) @@ -240,7 +238,7 @@ def test_examples_protocol_advanced_https_ota_example_ota_resumption(dut: Dut) - try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') @@ -252,19 +250,13 @@ def test_examples_protocol_advanced_https_ota_example_ota_resumption(dut: Dut) - dut.expect('Starting OTA...', timeout=60) restart_device_with_random_delay(dut, 5, 15) - thread1.terminate() - - # Start server - thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port)) - thread1.daemon = True - thread1.start() # Validate that the device restarts correctly dut.expect('Loaded app from partition at offset', timeout=180) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') @@ -316,7 +308,7 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(dut: Dut) -> dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') dut.expect('Starting Advanced OTA example', timeout=30) @@ -367,7 +359,7 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(dut: Dut) dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -418,7 +410,7 @@ def test_examples_protocol_advanced_https_ota_example_random(dut: Dut) -> None: dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -471,7 +463,7 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(dut: Dut) dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -508,7 +500,7 @@ def test_examples_protocol_advanced_https_ota_example_chunked(dut: Dut) -> None: dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -546,7 +538,7 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(dut: Dut) -> dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') dut.expect('Starting Advanced OTA example', timeout=30) @@ -631,7 +623,7 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(dut: Dut) -> dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -695,7 +687,7 @@ def test_examples_protocol_advanced_https_ota_example_partial_request(dut: Dut) dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -753,7 +745,7 @@ def test_examples_protocol_advanced_https_ota_example_ota_resumption_partial_dow try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -763,19 +755,13 @@ def test_examples_protocol_advanced_https_ota_example_ota_resumption_partial_dow dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name) restart_device_with_random_delay(dut, 5, 15) - thread1.terminate() - - # Start server - thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port)) - thread1.daemon = True - thread1.start() # Validate that the device restarts correctly dut.expect('Loaded app from partition at offset', timeout=180) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') @@ -830,7 +816,7 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(dut: Dut) -> dut.write(f'{ap_ssid} {ap_password}') try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP with IP: {}'.format(ip_address)) + print(f'Connected to AP with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -886,7 +872,7 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(dut: Dut) dut.write(f'{ap_ssid} {ap_password}') try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP with IP: {}'.format(ip_address)) + print(f'Connected to AP with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -937,7 +923,7 @@ def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(dut: D dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') host_ip = get_host_ip4_by_dest_ip(ip_address) @@ -997,7 +983,7 @@ def test_examples_protocol_advanced_https_ota_example_verify_min_chip_revision(d try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') @@ -1053,7 +1039,7 @@ def test_examples_protocol_advanced_https_ota_example_verify_max_chip_revision(d try: ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() - print('Connected to AP/Ethernet with IP: {}'.format(ip_address)) + print(f'Connected to AP/Ethernet with IP: {ip_address}') except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') diff --git a/examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption b/examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption index 5c68adc02a..420a484407 100644 --- a/examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption +++ b/examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption @@ -1,7 +1,7 @@ CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN" CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y CONFIG_EXAMPLE_SKIP_VERSION_CHECK=y -CONFIG_EXAMPLE_OTA_RECV_TIMEOUT=3000 +CONFIG_EXAMPLE_OTA_RECV_TIMEOUT=15000 CONFIG_EXAMPLE_ENABLE_OTA_RESUMPTION=y CONFIG_EXAMPLE_CONNECT_ETHERNET=y diff --git a/examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption_partial_download b/examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption_partial_download index df17ac916e..7f709d1eca 100644 --- a/examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption_partial_download +++ b/examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption_partial_download @@ -1,7 +1,7 @@ CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN" CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y CONFIG_EXAMPLE_SKIP_VERSION_CHECK=y -CONFIG_EXAMPLE_OTA_RECV_TIMEOUT=3000 +CONFIG_EXAMPLE_OTA_RECV_TIMEOUT=15000 CONFIG_ESP_HTTPS_OTA_ENABLE_PARTIAL_DOWNLOAD=y CONFIG_EXAMPLE_ENABLE_PARTIAL_HTTP_DOWNLOAD=y CONFIG_EXAMPLE_ENABLE_OTA_RESUMPTION=y