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

Merge branch 'feature/pytest_ota_examples' into 'master'

Browse Source 
Migrate native OTA example test to pytest framework

See merge request espressif/esp-idf!17544
This commit is contained in:
Mahavir Jain
2022-03-28 13:44:51 +08:00
parent c4eb0c3584 e9ad8a63c3
commit 58b30a9423
3 changed files with 90 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.gitlab/ci/target-test.yml
View File

@@ -72,6 +72,16 @@ example_test_pytest_esp32c3_flash_suspend:
TARGET: ESP32C3
ENV_MARKER: flash_suspend
example_test_pytest_esp32_ethernet_ota:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32
needs:
- build_pytest_examples_esp32
variables:
TARGET: ESP32
ENV_MARKER: ethernet_ota
.pytest_components_dir_template:
extends: .pytest_template
variables:

184
examples/system/ota/native_ota_example/example_test.py → examples/system/ota/native_ota_example/pytest_native_ota.py
View File

@@ -1,15 +1,18 @@
# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Unlicense OR CC0-1.0
import http.server
import multiprocessing
import os
import random
import re
import socket
import ssl
import struct
import subprocess
from typing import Callable, Tuple
import ttfw_idf
from tiny_test_fw import DUT
import pexpect
import pytest
from pytest_embedded import Dut
server_cert = '-----BEGIN CERTIFICATE-----\n' \
'MIIDWDCCAkACCQCbF4+gVh/MLjANBgkqhkiG9w0BAQsFADBuMQswCQYDVQQGEwJJ\n'\
@@ -62,15 +65,16 @@ server_key = '-----BEGIN PRIVATE KEY-----\n'\
'-----END PRIVATE KEY-----\n'
def get_my_ip():
def get_my_ip() -> str:
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect(('8.8.8.8', 80))
my_ip = ''
my_ip = s1.getsockname()[0]
s1.close()
return my_ip
def get_server_status(host_ip, port):
def get_server_status(host_ip: str, port: int) -> bool:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_status = sock.connect_ex((host_ip, port))
sock.close()
@@ -79,12 +83,12 @@ def get_server_status(host_ip, port):
return False
def create_file(server_file, file_data):
def create_file(server_file: str, file_data: str) -> None:
with open(server_file, 'w+') as file:
file.write(file_data)
def get_ca_cert(ota_image_dir):
def get_ca_cert(ota_image_dir: str) -> Tuple[str, str]:
os.chdir(ota_image_dir)
server_file = os.path.join(ota_image_dir, 'server_cert.pem')
create_file(server_file, server_cert)
@@ -94,12 +98,12 @@ def get_ca_cert(ota_image_dir):
return server_file, key_file
def https_request_handler():
def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]:
"""
Returns a request handler class that handles broken pipe exception
"""
class RequestHandler(http.server.SimpleHTTPRequestHandler):
def finish(self):
def finish(self) -> None:
try:
if not self.wfile.closed:
self.wfile.flush()
@@ -108,7 +112,7 @@ def https_request_handler():
pass
self.rfile.close()
def handle(self):
def handle(self) -> None:
try:
http.server.BaseHTTPRequestHandler.handle(self)
except socket.error:
@@ -117,7 +121,7 @@ def https_request_handler():
return RequestHandler
def start_https_server(ota_image_dir, server_ip, server_port):
def start_https_server(ota_image_dir: str, server_ip: str, server_port: int) -> None:
server_file, key_file = get_ca_cert(ota_image_dir)
requestHandler = https_request_handler()
httpd = http.server.HTTPServer((server_ip, server_port), requestHandler)
@@ -128,14 +132,15 @@ def start_https_server(ota_image_dir, server_ip, server_port):
httpd.serve_forever()
def start_chunked_server(ota_image_dir, server_port):
def start_chunked_server(ota_image_dir: str, server_port: int) -> subprocess.Popen:
server_file, key_file = get_ca_cert(ota_image_dir)
chunked_server = subprocess.Popen(['openssl', 's_server', '-WWW', '-key', key_file, '-cert', server_file, '-port', str(server_port)])
return chunked_server
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
def test_examples_protocol_native_ota_example(env, extra_data):
@pytest.mark.supported_targets
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example(dut: Dut) -> None:
"""
This is a positive test case, which downloads complete binary file multiple number of times.
Number of iterations can be specified in variable iterations.
@@ -144,43 +149,35 @@ def test_examples_protocol_native_ota_example(env, extra_data):
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
server_port = 8002
# No. of times working of application to be validated
iterations = 3
# File to be downloaded. This file is generated after compilation
bin_name = 'native_ota.bin'
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1.daemon = True
thread1.start()
dut1.start_app()
for i in range(iterations):
dut1.expect('Loaded app from partition at offset', timeout=30)
dut.expect('Loaded app from partition at offset', timeout=60)
try:
ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
ip_address = dut.expect(r' (sta|eth) ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
dut1.expect('Starting OTA example', timeout=30)
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut.expect('Starting OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting OTA example', timeout=30)
dut1.reset()
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data):
@pytest.mark.supported_targets
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example_truncated_bin(dut: Dut) -> None:
"""
Working of OTA if binary file is truncated is validated in this test case.
Application should return with error message in this case.
@@ -190,7 +187,6 @@ def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data):
3. Fetch OTA image over HTTPS
4. Check working of code if bin is truncated
"""
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
server_port = 8002
# Original binary file generated after compilation
bin_name = 'native_ota.bin'
@@ -200,40 +196,38 @@ def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data):
# truncated_bin_size is set to 64000 to reduce consumed by the test case
truncated_bin_size = 64000
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
binary_file = os.path.join(dut.app.binary_path, bin_name)
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+')
fo = open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+')
fo.write(f.read(truncated_bin_size))
fo.close()
f.close()
binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect('Loaded app from partition at offset', timeout=30)
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=60)
ip_address = dut.expect(r' (sta|eth) ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
dut1.expect('Starting OTA example', timeout=30)
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut.expect('Starting OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut1.expect('native_ota_example: Image validation failed, image is corrupted', timeout=20)
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut.expect('native_ota_example: Image validation failed, image is corrupted', timeout=20)
os.remove(binary_file)
thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
def test_examples_protocol_native_ota_example_truncated_header(env, extra_data):
@pytest.mark.supported_targets
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example_truncated_header(dut: Dut) -> None:
"""
Working of OTA if headers of binary file are truncated is vaildated in this test case.
Application should return with error message in this case.
@@ -243,7 +237,6 @@ def test_examples_protocol_native_ota_example_truncated_header(env, extra_data):
3. Fetch OTA image over HTTPS
4. Check working of code if headers are not sent completely
"""
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
server_port = 8002
# Original binary file generated after compilation
bin_name = 'native_ota.bin'
@@ -252,40 +245,38 @@ def test_examples_protocol_native_ota_example_truncated_header(env, extra_data):
# Size of truncated file to be grnerated. This value should be less than 288 bytes (Image header size)
truncated_bin_size = 180
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
binary_file = os.path.join(dut.app.binary_path, bin_name)
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+')
fo = open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+')
fo.write(f.read(truncated_bin_size))
fo.close()
f.close()
binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect('Loaded app from partition at offset', timeout=30)
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=60)
ip_address = dut.expect(r' (sta|eth) ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
dut1.expect('Starting OTA example', timeout=30)
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut.expect('Starting OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut1.expect('native_ota_example: received package is not fit len', timeout=20)
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut.expect('native_ota_example: received package is not fit len', timeout=20)
os.remove(binary_file)
thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
def test_examples_protocol_native_ota_example_random(env, extra_data):
@pytest.mark.supported_targets
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example_random(dut: Dut) -> None:
"""
Working of OTA if random data is added in binary file are validated in this test case.
Magic byte verification should fail in this case.
@@ -295,14 +286,13 @@ def test_examples_protocol_native_ota_example_random(env, extra_data):
3. Fetch OTA image over HTTPS
4. Check working of code for random binary file
"""
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
server_port = 8002
# Random binary file to be generated
random_bin_name = 'random.bin'
# Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
random_bin_size = 32000
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, random_bin_name)
binary_file = os.path.join(dut.app.binary_path, random_bin_name)
fo = open(binary_file, 'wb+')
# First byte of binary file is always set to zero. If first byte is generated randomly,
# in some cases it may generate 0xE9 which will result in failure of testcase.
@@ -310,33 +300,31 @@ def test_examples_protocol_native_ota_example_random(env, extra_data):
for i in range(random_bin_size - 1):
fo.write(struct.pack('B', random.randrange(0,255,1)))
fo.close()
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect('Loaded app from partition at offset', timeout=30)
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=60)
ip_address = dut.expect(r' (sta|eth) ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
dut1.expect('Starting OTA example', timeout=30)
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut.expect('Starting OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
dut1.expect('esp_ota_ops: OTA image has invalid magic byte', timeout=20)
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
dut.expect('esp_ota_ops: OTA image has invalid magic byte', timeout=20)
os.remove(binary_file)
thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
def test_examples_protocol_native_ota_example_chunked(env, extra_data):
@pytest.mark.supported_targets
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example_chunked(dut: Dut) -> None:
"""
This is a positive test case, which downloads complete binary file multiple number of times.
Number of iterations can be specified in variable iterations.
@@ -345,37 +333,23 @@ def test_examples_protocol_native_ota_example_chunked(env, extra_data):
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
# File to be downloaded. This file is generated after compilation
bin_name = 'native_ota.bin'
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
chunked_server = start_chunked_server(dut1.app.binary_path, 8070)
dut1.start_app()
dut1.expect('Loaded app from partition at offset', timeout=30)
chunked_server = start_chunked_server(dut.app.binary_path, 8070)
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
ip_address = dut.expect(r' (sta|eth) ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting OTA example', timeout=30)
dut.expect('Starting OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
dut1.write('https://' + host_ip + ':8070/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting OTA example', timeout=30)
dut.write('https://' + host_ip + ':8070/' + bin_name)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect('Starting OTA example', timeout=30)
chunked_server.kill()
os.remove(os.path.join(dut1.app.binary_path, 'server_cert.pem'))
os.remove(os.path.join(dut1.app.binary_path, 'server_key.pem'))
if __name__ == '__main__':
test_examples_protocol_native_ota_example()
test_examples_protocol_native_ota_example_chunked()
test_examples_protocol_native_ota_example_truncated_bin()
test_examples_protocol_native_ota_example_truncated_header()
test_examples_protocol_native_ota_example_random()
os.remove(os.path.join(dut.app.binary_path, 'server_cert.pem'))
os.remove(os.path.join(dut.app.binary_path, 'server_key.pem'))

1
pytest.ini
View File

@@ -30,6 +30,7 @@ markers =
lan8720: connected via LAN8720 ethernet transceiver
octal_psram: runners with octal psram
usb_host: usb host runners
ethernet_ota: ethernet OTA runners
# log related
log_cli = True