espressif/esp-idf
1
0
Fork 1
mirror of https://github.com/espressif/esp-idf.git synced 2025-10-03 02:20:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix: ota: increase timeout and remove redundant server restart in tests

Browse Source 
This comit updated pytest file in advanced ota example and also updates
timeout for example to avoid recurring timeout errors in CI
This commit is contained in:
nilesh.kale
2025-08-18 16:22:48 +05:30
parent 924540d861
commit 7fc57dbef5
3 changed files with 26 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
examples/system/ota/advanced_https_ota/pytest_advanced_ota.py
View File

@@ -4,13 +4,11 @@ import http.server
import multiprocessing import multiprocessing
import os import os
import random import random
import socket
import ssl import ssl
import struct import struct
import subprocess import subprocess
import time import time
from typing import Callable from collections.abc import Callable
from typing import Optional
import pexpect import pexpect
import pytest import pytest
@@ -53,14 +51,14 @@ def https_request_handler() -> Callable[..., http.server.BaseHTTPRequestHandler]
if not self.wfile.closed: if not self.wfile.closed:
self.wfile.flush() self.wfile.flush()
self.wfile.close() self.wfile.close()
except socket.error: except OSError:
pass pass
self.rfile.close() self.rfile.close()
def handle(self) -> None: def handle(self) -> None:
try: try:
RangeRequestHandler.handle(self) RangeRequestHandler.handle(self)
except socket.error: except OSError:
pass pass
return RequestHandler return RequestHandler
@@ -111,7 +109,7 @@ def redirect_handler_factory(url: str) -> Callable[..., http.server.BaseHTTPRequ
def handle(self) -> None: def handle(self) -> None:
try: try:
http.server.BaseHTTPRequestHandler.handle(self) http.server.BaseHTTPRequestHandler.handle(self)
except socket.error: except OSError:
pass pass
return RedirectHandler 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 # Function to modify chip revisions in the app header
def modify_chip_revision( 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: ) -> None:
""" """
Modify min_chip_rev_full and max_chip_rev_full in the app header. 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: with open(app_path, 'r+b') as f:
f.write(header) f.write(header)
except IOError as e: except OSError as e:
raise RuntimeError(f'Failed to modify app header: {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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
dut.expect('Starting Advanced OTA example', timeout=30) 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: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') 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) dut.expect('Starting OTA...', timeout=60)
restart_device_with_random_delay(dut, 5, 15) 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 # Validate that the device restarts correctly
dut.expect('Loaded app from partition at offset', timeout=180) dut.expect('Loaded app from partition at offset', timeout=180)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
dut.expect('Starting Advanced OTA example', timeout=30) 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
dut.expect('Starting Advanced OTA example', timeout=30) 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
restart_device_with_random_delay(dut, 5, 15) 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 # Validate that the device restarts correctly
dut.expect('Loaded app from partition at offset', timeout=180) dut.expect('Loaded app from partition at offset', timeout=180)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') 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}') dut.write(f'{ap_ssid} {ap_password}')
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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}') dut.write(f'{ap_ssid} {ap_password}')
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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) dut.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address) 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: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') 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: try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=30)[1].decode() 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: except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')

2
examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption
View File

@@ -1,7 +1,7 @@
CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN" CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN"
CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y
CONFIG_EXAMPLE_SKIP_VERSION_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_ENABLE_OTA_RESUMPTION=y
CONFIG_EXAMPLE_CONNECT_ETHERNET=y CONFIG_EXAMPLE_CONNECT_ETHERNET=y

2
examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption_partial_download
View File

@@ -1,7 +1,7 @@
CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN" CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN"
CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y
CONFIG_EXAMPLE_SKIP_VERSION_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_ESP_HTTPS_OTA_ENABLE_PARTIAL_DOWNLOAD=y
CONFIG_EXAMPLE_ENABLE_PARTIAL_HTTP_DOWNLOAD=y CONFIG_EXAMPLE_ENABLE_PARTIAL_HTTP_DOWNLOAD=y
CONFIG_EXAMPLE_ENABLE_OTA_RESUMPTION=y CONFIG_EXAMPLE_ENABLE_OTA_RESUMPTION=y