alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2025-08-11 13:00:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: format all test scripts

Browse Source
This commit is contained in:
igor.udot
2025-02-24 10:18:03 +08:00
parent 717c18a58e
commit daf2d31008
381 changed files with 6180 additions and 4289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,6 +16,7 @@ import pytest
from common_test_methods import get_env_config_variable
from common_test_methods import get_host_ip4_by_dest_ip
from pytest_embedded import Dut
from pytest_embedded_idf.utils import idf_parametrize
from RangeHTTPServer import RangeRequestHandler
NVS_PARTITION = 'nvs'
@@ -40,10 +41,11 @@ def restart_device_with_random_delay(dut: Dut, min_delay: int = 10, max_delay: i
print('Device restarted after random delay.')
def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]:
def https_request_handler() -> Callable[..., http.server.BaseHTTPRequestHandler]:
"""
Returns a request handler class that handles broken pipe exception
"""
class RequestHandler(RangeRequestHandler):
def finish(self) -> None:
try:
@@ -77,14 +79,25 @@ def start_https_server(ota_image_dir: str, server_ip: str, server_port: int) ->
def start_chunked_server(ota_image_dir: str, server_port: int) -> subprocess.Popen:
os.chdir(ota_image_dir)
chunked_server = subprocess.Popen(['openssl', 's_server', '-WWW', '-key', key_file, '-cert', server_file, '-port', str(server_port)])
chunked_server = subprocess.Popen([
'openssl',
's_server',
'-WWW',
'-key',
key_file,
'-cert',
server_file,
'-port',
str(server_port),
])
return chunked_server
def redirect_handler_factory(url: str) -> Callable[...,http.server.BaseHTTPRequestHandler]:
def redirect_handler_factory(url: str) -> Callable[..., http.server.BaseHTTPRequestHandler]:
"""
Returns a request handler class that redirects to supplied `url`
"""
class RedirectHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self) -> None:
print('Sending resp, URL: ' + url)
@@ -103,7 +116,9 @@ def redirect_handler_factory(url: str) -> Callable[...,http.server.BaseHTTPReque
def start_redirect_server(ota_image_dir: str, server_ip: str, server_port: int, redirection_port: int) -> None:
os.chdir(ota_image_dir)
redirectHandler = redirect_handler_factory('https://' + server_ip + ':' + str(redirection_port) + '/advanced_https_ota.bin')
redirectHandler = redirect_handler_factory(
'https://' + server_ip + ':' + str(redirection_port) + '/advanced_https_ota.bin'
)
httpd = http.server.HTTPServer((server_ip, server_port), redirectHandler)
@@ -114,8 +129,8 @@ def start_redirect_server(ota_image_dir: str, server_ip: str, server_port: int,
httpd.serve_forever()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example(dut: Dut) -> None:
"""
This is a positive test case, which downloads complete binary file multiple number of times.
@@ -152,9 +167,9 @@ def test_examples_protocol_advanced_https_ota_example(dut: Dut) -> None:
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.wifi_router
@pytest.mark.parametrize('config', ['ota_resumption'], indirect=True)
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_ota_resumption(dut: Dut) -> None:
"""
This is a positive test case, which stops the download midway and resumes downloading again.
@@ -235,8 +250,8 @@ def test_examples_protocol_advanced_https_ota_example_ota_resumption(dut: Dut) -
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_truncated_bin(dut: Dut) -> None:
"""
Working of OTA if binary file is truncated is validated in this test case.
@@ -286,8 +301,8 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(dut: Dut) ->
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_truncated_header(dut: Dut) -> None:
"""
Working of OTA if headers of binary file are truncated is validated in this test case.
@@ -337,8 +352,8 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(dut: Dut)
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_random(dut: Dut) -> None:
"""
Working of OTA if random data is added in binary file are validated in this test case.
@@ -361,7 +376,7 @@ def test_examples_protocol_advanced_https_ota_example_random(dut: Dut) -> None:
# in some cases it may generate 0xE9 which will result in failure of testcase.
output_file.write(struct.pack('B', 0))
for i in range(random_bin_size - 1):
output_file.write(struct.pack('B', random.randrange(0,255,1)))
output_file.write(struct.pack('B', random.randrange(0, 255, 1)))
# Start server
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
@@ -388,8 +403,8 @@ def test_examples_protocol_advanced_https_ota_example_random(dut: Dut) -> None:
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(dut: Dut) -> None:
"""
Working of OTA if binary file have invalid chip id is validated in this test case.
@@ -412,7 +427,7 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(dut: Dut)
with open(binary_file, 'rb+') as f:
data = list(f.read(random_bin_size))
# Changing Chip id
data[13] = 0xfe
data[13] = 0xFE
with open(random_binary_file, 'wb+') as output_file:
output_file.write(bytearray(data))
# Start server
@@ -441,8 +456,8 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(dut: Dut)
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_chunked(dut: Dut) -> None:
"""
This is a positive test case, which downloads complete binary file multiple number of times.
@@ -477,8 +492,8 @@ def test_examples_protocol_advanced_https_ota_example_chunked(dut: Dut) -> None:
chunked_server.kill()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_redirect_url(dut: Dut) -> None:
"""
This is a positive test case, which starts a server and a redirection server.
@@ -508,9 +523,14 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(dut: Dut) ->
host_ip = get_host_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1.daemon = True
thread2 = multiprocessing.Process(target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port, redirection_server_port1))
thread2 = multiprocessing.Process(
target=start_redirect_server,
args=(dut.app.binary_path, host_ip, redirection_server_port, redirection_server_port1),
)
thread2.daemon = True
thread3 = multiprocessing.Process(target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port1, server_port))
thread3 = multiprocessing.Process(
target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port1, server_port)
)
thread3.daemon = True
thread1.start()
thread2.start()
@@ -518,7 +538,9 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(dut: Dut) ->
time.sleep(1)
try:
print('writing to device: {}'.format('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name))
print(
'writing to device: {}'.format('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name)
)
dut.write('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name)
dut.expect('upgrade successful. Rebooting ...', timeout=150)
# after reboot
@@ -530,10 +552,16 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(dut: Dut) ->
thread3.terminate()
@pytest.mark.esp32
@pytest.mark.flash_encryption_ota
@pytest.mark.parametrize('config', ['anti_rollback',], indirect=True)
@pytest.mark.parametrize(
'config',
[
'anti_rollback',
],
indirect=True,
)
@pytest.mark.parametrize('skip_autoflash', ['y'], indirect=True)
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_anti_rollback(dut: Dut) -> None:
"""
Working of OTA when anti_rollback is enabled and security version of new image is less than current one.
@@ -587,7 +615,9 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(dut: Dut) ->
# Negative Case
dut.expect('Starting Advanced OTA example', timeout=30)
# Use modified image with secure_version=0
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name))
print(
'writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name)
)
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name)
dut.expect('New firmware security version is less than eFuse programmed, 0 < 1', timeout=30)
try:
@@ -598,9 +628,15 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(dut: Dut) ->
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@pytest.mark.parametrize('config', ['partial_download',], indirect=True)
@pytest.mark.parametrize(
'config',
[
'partial_download',
],
indirect=True,
)
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_partial_request(dut: Dut) -> None:
"""
This is a positive test case, to test OTA workflow with Range HTTP header.
@@ -645,9 +681,15 @@ def test_examples_protocol_advanced_https_ota_example_partial_request(dut: Dut)
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.wifi_router
@pytest.mark.parametrize('config', ['ota_resumption_partial_download',], indirect=True)
@pytest.mark.parametrize(
'config',
[
'ota_resumption_partial_download',
],
indirect=True,
)
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_ota_resumption_partial_download_request(dut: Dut) -> None:
"""
This is a positive test case, to test OTA workflow with Range HTTP header.
@@ -732,11 +774,15 @@ def test_examples_protocol_advanced_https_ota_example_ota_resumption_partial_dow
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.esp32c3
@pytest.mark.esp32s3
@pytest.mark.wifi_high_traffic
@pytest.mark.parametrize('config', ['nimble',], indirect=True)
@pytest.mark.parametrize(
'config',
[
'nimble',
],
indirect=True,
)
@idf_parametrize('target', ['esp32', 'esp32c3', 'esp32s3'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_nimble_gatts(dut: Dut) -> None:
"""
Run an OTA image update while a BLE GATT Server is running in background. This GATT server will be using NimBLE Host stack.
@@ -783,11 +829,15 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(dut: Dut) ->
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.esp32c3
@pytest.mark.esp32s3
@pytest.mark.wifi_high_traffic
@pytest.mark.parametrize('config', ['bluedroid',], indirect=True)
@pytest.mark.parametrize(
'config',
[
'bluedroid',
],
indirect=True,
)
@idf_parametrize('target', ['esp32', 'esp32c3', 'esp32s3'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(dut: Dut) -> None:
"""
Run an OTA image update while a BLE GATT Server is running in background. This GATT server will be using Bluedroid Host stack.
@@ -835,8 +885,8 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(dut: Dut)
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(dut: Dut) -> None:
"""
This is a test case for esp_http_client_read with binary size multiple of 289 bytes
@@ -859,7 +909,7 @@ def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(dut: D
with open(os.path.join(dut.app.binary_path, aligned_bin_name), 'wb+') as output_file:
output_file.write(f.read(bin_size))
for _ in range(dummy_data_size):
output_file.write(struct.pack('B', random.randrange(0,255,1)))
output_file.write(struct.pack('B', random.randrange(0, 255, 1)))
# Start server
chunked_server = start_chunked_server(dut.app.binary_path, 8070)
try: