feedc0de/esp-idf
1
0
Fork 0
forked from espressif/esp-idf
Code Pull Requests Activity

Merge branch 'feature/ota_examples_pytest_migration' into 'master'

Browse Source 
Migrate OTA example test to pytest framework

Closes IDFCI-1146

See merge request espressif/esp-idf!17767
This commit is contained in:
Mahavir Jain
2022-05-24 14:28:15 +08:00
parent 60d51bf6c8 448b98d3bc
commit 5607018f28
10 changed files with 447 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -91,6 +91,56 @@ example_test_pytest_esp32_ethernet_ota:
TARGET: ESP32 TARGET: ESP32
ENV_MARKER: ethernet_ota ENV_MARKER: ethernet_ota
example_test_pytest_esp32_wifi_ota:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32
needs:
- build_pytest_examples_esp32
variables:
TARGET: ESP32
ENV_MARKER: wifi_ota
example_test_pytest_esp32_flash_encryption_ota:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32
needs:
- build_pytest_examples_esp32
variables:
TARGET: ESP32
ENV_MARKER: flash_encryption_ota
example_test_pytest_esp32c3_flash_encryption_wifi_ota:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32c3
needs:
- build_pytest_examples_esp32c3
variables:
TARGET: ESP32C3
ENV_MARKER: flash_encryption_wifi_ota
example_test_pytest_esp32_ethernet:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32
needs:
- build_pytest_examples_esp32
variables:
TARGET: ESP32
ENV_MARKER: ethernet
example_test_pytest_esp32_8mb_flash:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32
needs:
- build_pytest_examples_esp32
variables:
TARGET: ESP32
ENV_MARKER: ethernet_flash_8m
example_test_pytest_esp32_wifi: example_test_pytest_esp32_wifi:
extends: extends:
- .pytest_examples_dir_template - .pytest_examples_dir_template

4
examples/system/ota/README.md
View File

@@ -74,7 +74,7 @@ After a successful build, we need to create a self-signed certificate and run a
(Some windows builds of openssl translate CR/LF sequences to LF in the served files, leading to corrupted images received by the OTA client; others interpret the `0x1a`/`SUB` character in a binary as an escape sequence, i.e. end of file, and close the connection prematurely thus preventing the OTA client from receiving a complete image). (Some windows builds of openssl translate CR/LF sequences to LF in the served files, leading to corrupted images received by the OTA client; others interpret the `0x1a`/`SUB` character in a binary as an escape sequence, i.e. end of file, and close the connection prematurely thus preventing the OTA client from receiving a complete image).
* We recommend using the `openssl` binary bundled in `Git For Windows` from the [ESP-IDF Tool installer](https://docs.espressif.com/projects/esp-idf/en/latest/get-started/windows-setup.html): * We recommend using the `openssl` binary bundled in `Git For Windows` from the [ESP-IDF Tool installer](https://docs.espressif.com/projects/esp-idf/en/latest/get-started/windows-setup.html):
Open the ESP-IDF command prompt and add the internal openssl binary to your path: `set PATH=%LocalAppData%\Git\usr\bin;%PATH%` and run openssl's http server command as above. Open the ESP-IDF command prompt and add the internal openssl binary to your path: `set PATH=%LocalAppData%\Git\usr\bin;%PATH%` and run openssl's http server command as above.
* Alternatively, use any windows based openssl with version `v1.1.1i` or greater built on the `Msys-x86_64` platform, or a simple python https server -- see `start_https_server` in the [example_test](simple_ota_example/example_test.py) script. * Alternatively, use any windows based openssl with version `v1.1.1i` or greater built on the `Msys-x86_64` platform, or a simple python https server -- see `start_https_server` in the [example_test](simple_ota_example/pytest_simple_ota.py) script.
### Flash Certificate to "ESP-Dev-Board" ### Flash Certificate to "ESP-Dev-Board"
@@ -147,7 +147,7 @@ Running a local https server might be tricky in some cases (due to self signed c
- Execute `python -m http.server 8070` in the directory with the firmware image - Execute `python -m http.server 8070` in the directory with the firmware image
- Use http://<host-ip>:8070/<firmware-name> as the firmware upgrade URL - Use http://<host-ip>:8070/<firmware-name> as the firmware upgrade URL
- Enable *Allow HTTP for OTA* (`CONFIG_ESP_HTTPS_OTA_ALLOW_HTTP`) in `Component config -> ESP HTTPS OTA` so the URI without TLS is accepted - Enable *Allow HTTP for OTA* (`CONFIG_ESP_HTTPS_OTA_ALLOW_HTTP`) in `Component config -> ESP HTTPS OTA` so the URI without TLS is accepted
* Start the HTTPS server using [example_test](simple_ota_example/example_test.py) with two or more parameters: `example_test.py <BIN_DIR> <PORT> [CERT_DIR]`, where: * Start the HTTPS server using [example_test](simple_ota_example/pytest_simple_ota.py) with two or more parameters: `pytest_simple_ota.py <BIN_DIR> <PORT> [CERT_DIR]`, where:
- `<BIN_DIR>` is a directory containing the image and by default also the certificate and key files:`ca_cert.pem` and `ca_key.pem` - `<BIN_DIR>` is a directory containing the image and by default also the certificate and key files:`ca_cert.pem` and `ca_key.pem`
- `<PORT>` is the server's port, here `8070` - `<PORT>` is the server's port, here `8070`
- `[CERT_DIR]` is an optional argument pointing to a specific directory with the certificate and key file. - `[CERT_DIR]` is an optional argument pointing to a specific directory with the certificate and key file.

2
examples/system/ota/advanced_https_ota/README.md
View File

@@ -1,3 +1,5 @@
| Supported Targets | ESP32 | ESP32-S2 | ESP32-S3 | ESP32-C3 |
| ----------------- | ----- | -------- | -------- | -------- |
# Advanced HTTPS OTA example # Advanced HTTPS OTA example
This example is based on `esp_https_ota` component's APIs. This example is based on `esp_https_ota` component's APIs.

446
examples/system/ota/advanced_https_ota/example_test.py → examples/system/ota/advanced_https_ota/pytest_advanced_ota.py
View File

@@ -1,30 +1,34 @@
# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Unlicense OR CC0-1.0
import http.server import http.server
import multiprocessing import multiprocessing
import os import os
import random import random
import re
import socket import socket
import ssl import ssl
import struct import struct
import subprocess import subprocess
from typing import Callable
import ttfw_idf import pexpect
import pytest
from pytest_embedded import Dut
from RangeHTTPServer import RangeRequestHandler from RangeHTTPServer import RangeRequestHandler
from tiny_test_fw import DUT, Utility
server_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_certs/server_cert.pem') server_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_certs/server_cert.pem')
key_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_certs/server_key.pem') key_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_certs/server_key.pem')
def get_my_ip(): def get_my_ip() -> str:
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect(('8.8.8.8', 80)) s1.connect(('8.8.8.8', 80))
my_ip = ''
my_ip = s1.getsockname()[0] my_ip = s1.getsockname()[0]
s1.close() s1.close()
return my_ip 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) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_status = sock.connect_ex((host_ip, port)) server_status = sock.connect_ex((host_ip, port))
sock.close() sock.close()
@@ -33,12 +37,12 @@ def get_server_status(host_ip, port):
return False return False
def https_request_handler(): def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]:
""" """
Returns a request handler class that handles broken pipe exception Returns a request handler class that handles broken pipe exception
""" """
class RequestHandler(RangeRequestHandler): class RequestHandler(RangeRequestHandler):
def finish(self): def finish(self) -> None:
try: try:
if not self.wfile.closed: if not self.wfile.closed:
self.wfile.flush() self.wfile.flush()
@@ -47,7 +51,7 @@ def https_request_handler():
pass pass
self.rfile.close() self.rfile.close()
def handle(self): def handle(self) -> None:
try: try:
RangeRequestHandler.handle(self) RangeRequestHandler.handle(self)
except socket.error: except socket.error:
@@ -56,7 +60,7 @@ def https_request_handler():
return RequestHandler 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:
os.chdir(ota_image_dir) os.chdir(ota_image_dir)
requestHandler = https_request_handler() requestHandler = https_request_handler()
httpd = http.server.HTTPServer((server_ip, server_port), requestHandler) httpd = http.server.HTTPServer((server_ip, server_port), requestHandler)
@@ -67,24 +71,24 @@ def start_https_server(ota_image_dir, server_ip, server_port):
httpd.serve_forever() 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:
os.chdir(ota_image_dir) 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 return chunked_server
def redirect_handler_factory(url): def redirect_handler_factory(url: str) -> Callable[...,http.server.BaseHTTPRequestHandler]:
""" """
Returns a request handler class that redirects to supplied `url` Returns a request handler class that redirects to supplied `url`
""" """
class RedirectHandler(http.server.SimpleHTTPRequestHandler): class RedirectHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self): def do_GET(self) -> None:
print('Sending resp, URL: ' + url) print('Sending resp, URL: ' + url)
self.send_response(301) self.send_response(301)
self.send_header('Location', url) self.send_header('Location', url)
self.end_headers() self.end_headers()
def handle(self): def handle(self) -> None:
try: try:
http.server.BaseHTTPRequestHandler.handle(self) http.server.BaseHTTPRequestHandler.handle(self)
except socket.error: except socket.error:
@@ -93,7 +97,7 @@ def redirect_handler_factory(url):
return RedirectHandler return RedirectHandler
def start_redirect_server(ota_image_dir, server_ip, server_port, redirection_port): def start_redirect_server(ota_image_dir: str, server_ip: str, server_port: int, redirection_port: int) -> None:
os.chdir(ota_image_dir) 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')
@@ -105,8 +109,12 @@ def start_redirect_server(ota_image_dir, server_ip, server_port, redirection_por
httpd.serve_forever() httpd.serve_forever()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
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. This is a positive test case, which downloads complete binary file multiple number of times.
Number of iterations can be specified in variable iterations. Number of iterations can be specified in variable iterations.
@@ -115,43 +123,37 @@ def test_examples_protocol_advanced_https_ota_example(env, extra_data):
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
# Number of iterations to validate OTA # Number of iterations to validate OTA
iterations = 3 iterations = 3
server_port = 8001 server_port = 8001
# File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_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('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app()
for i in range(iterations): for i in range(iterations):
dut1.expect('Loaded app from partition at offset', timeout=30) dut.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting Advanced OTA example', timeout=30)
dut1.reset()
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
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. Working of OTA if binary file is truncated is validated in this test case.
Application should return with error message in this case. Application should return with error message in this case.
@@ -161,7 +163,6 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_d
3. Fetch OTA image over HTTPS 3. Fetch OTA image over HTTPS
4. Check working of code if bin is truncated 4. Check working of code if bin is truncated
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001 server_port = 8001
# Original binary file generated after compilation # Original binary file generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_ota.bin'
@@ -170,34 +171,30 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_d
# Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file # Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
# truncated_bin_size is set to 64000 to reduce consumed by the test case # truncated_bin_size is set to 64000 to reduce consumed by the test case
truncated_bin_size = 64000 truncated_bin_size = 64000
# check and log bin size binary_file = os.path.join(dut.app.binary_path, bin_name)
binary_file = os.path.join(dut1.app.binary_path, bin_name)
with open(binary_file, 'rb+') as f: with open(binary_file, 'rb+') as f:
with open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+') as fo: with open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+') as fo:
fo.write(f.read(truncated_bin_size)) fo.write(f.read(truncated_bin_size))
binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name) binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)) 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) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut1.expect('Image validation failed, image is corrupted', timeout=30) dut.expect('Image validation failed, image is corrupted', timeout=30)
try: try:
os.remove(binary_file) os.remove(binary_file)
except OSError: except OSError:
@@ -205,8 +202,12 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_d
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
def test_examples_protocol_advanced_https_ota_example_truncated_header(dut: Dut) -> None:
""" """
Working of OTA if headers of binary file are truncated is vaildated in this test case. 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. Application should return with error message in this case.
@@ -216,42 +217,38 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extr
3. Fetch OTA image over HTTPS 3. Fetch OTA image over HTTPS
4. Check working of code if headers are not sent completely 4. Check working of code if headers are not sent completely
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001 server_port = 8001
# Original binary file generated after compilation # Original binary file generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_ota.bin'
# Truncated binary file to be generated from original binary file # Truncated binary file to be generated from original binary file
truncated_bin_name = 'truncated_header.bin' truncated_bin_name = 'truncated_header.bin'
# Size of truncated file to be grnerated. This value should be less than 288 bytes (Image header size) # Size of truncated file to be generated. This value should be less than 288 bytes (Image header size)
truncated_bin_size = 180 truncated_bin_size = 180
# check and log bin size # 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)
with open(binary_file, 'rb+') as f: with open(binary_file, 'rb+') as f:
with open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+') as fo: with open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+') as fo:
fo.write(f.read(truncated_bin_size)) fo.write(f.read(truncated_bin_size))
binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name) binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)) 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) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut1.expect('advanced_https_ota_example: esp_https_ota_read_img_desc failed', timeout=30) dut.expect('advanced_https_ota_example: esp_https_ota_read_img_desc failed', timeout=30)
try: try:
os.remove(binary_file) os.remove(binary_file)
except OSError: except OSError:
@@ -259,8 +256,12 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extr
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_random(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
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. 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. Magic byte verification should fail in this case.
@@ -270,14 +271,13 @@ def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
3. Fetch OTA image over HTTPS 3. Fetch OTA image over HTTPS
4. Check working of code for random binary file 4. Check working of code for random binary file
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001 server_port = 8001
# Random binary file to be generated # Random binary file to be generated
random_bin_name = 'random.bin' random_bin_name = 'random.bin'
# Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case # Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
random_bin_size = 32000 random_bin_size = 32000
# check and log bin size # 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)
with open(binary_file, 'wb+') as fo: with open(binary_file, 'wb+') as fo:
# First byte of binary file is always set to zero. If first byte is generated randomly, # 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. # in some cases it may generate 0xE9 which will result in failure of testcase.
@@ -285,27 +285,24 @@ def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
for i in range(random_bin_size - 1): for i in range(random_bin_size - 1):
fo.write(struct.pack('B', random.randrange(0,255,1))) fo.write(struct.pack('B', random.randrange(0,255,1)))
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)) 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) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
dut1.expect(re.compile(r'esp_https_ota: Incorrect app descriptor magic'), timeout=10) dut.expect(r'esp_https_ota: Incorrect app descriptor magic', timeout=10)
try: try:
os.remove(binary_file) os.remove(binary_file)
except OSError: except OSError:
@@ -313,8 +310,12 @@ def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
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. Working of OTA if binary file have invalid chip id is validated in this test case.
Chip id verification should fail in this case. Chip id verification should fail in this case.
@@ -324,16 +325,15 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(env, extra
3. Fetch OTA image over HTTPS 3. Fetch OTA image over HTTPS
4. Check working of code for random binary file 4. Check working of code for random binary file
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001 server_port = 8001
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_ota.bin'
# Random binary file to be generated # Random binary file to be generated
random_bin_name = 'random.bin' random_bin_name = 'random.bin'
random_binary_file = os.path.join(dut1.app.binary_path, random_bin_name) random_binary_file = os.path.join(dut.app.binary_path, random_bin_name)
# Size of random binary file. 2000 is choosen, to reduce the time required to run the test-case # Size of random binary file. 2000 is choosen, to reduce the time required to run the test-case
random_bin_size = 2000 random_bin_size = 2000
binary_file = os.path.join(dut1.app.binary_path, bin_name) binary_file = os.path.join(dut.app.binary_path, bin_name)
with open(binary_file, 'rb+') as f: with open(binary_file, 'rb+') as f:
data = list(f.read(random_bin_size)) data = list(f.read(random_bin_size))
# Changing Chip id # Changing Chip id
@@ -344,22 +344,21 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(env, extra
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)) 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) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
dut1.expect(re.compile(r'esp_https_ota: Mismatch chip id, expected 0, found \d'), timeout=10) dut.expect(r'esp_https_ota: Mismatch chip id, expected 0, found \d', timeout=10)
try: try:
os.remove(random_binary_file) os.remove(random_binary_file)
except OSError: except OSError:
@@ -367,8 +366,12 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(env, extra
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_chunked(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
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. This is a positive test case, which downloads complete binary file multiple number of times.
Number of iterations can be specified in variable iterations. Number of iterations can be specified in variable iterations.
@@ -377,34 +380,32 @@ def test_examples_protocol_advanced_https_ota_example_chunked(env, extra_data):
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
# File to be downloaded. This file is generated after compilation # File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_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('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
chunked_server = start_chunked_server(dut1.app.binary_path, 8070) chunked_server = start_chunked_server(dut.app.binary_path, 8070)
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) 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') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name)) print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
dut1.write('https://' + host_ip + ':8070/' + bin_name) dut.write('https://' + host_ip + ':8070/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60) dut.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
chunked_server.kill() chunked_server.kill()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
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. This is a positive test case, which starts a server and a redirection server.
Redirection server redirects http_request to different port Redirection server redirects http_request to different port
@@ -414,53 +415,52 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_da
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001 server_port = 8001
# Port to which the request should be redirected # Port to which the request should be redirected
redirection_server_port = 8081 redirection_server_port = 8081
redirection_server_port1 = 8082 redirection_server_port1 = 8082
# File to be downloaded. This file is generated after compilation # File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_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('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
thread2 = multiprocessing.Process(target=start_redirect_server, args=(dut1.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 thread2.daemon = True
thread2.start() thread2.start()
thread3 = multiprocessing.Process(target=start_redirect_server, args=(dut1.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 thread3.daemon = True
thread3.start() thread3.start()
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
thread2.terminate() thread2.terminate()
thread3.terminate() thread3.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
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))
dut1.write('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name) dut.write('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60) dut.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
dut1.reset()
thread1.terminate() thread1.terminate()
thread2.terminate() thread2.terminate()
thread3.terminate() thread3.terminate()
@ttfw_idf.idf_example_test(env_tag='Example_8Mflash_Ethernet') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_flash_8m
@pytest.mark.parametrize('config', ['anti_rollback',], indirect=True)
@pytest.mark.parametrize('skip_autoflash', ['y'], indirect=True)
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. Working of OTA when anti_rollback is enabled and security version of new image is less than current one.
Application should return with error message in this case. Application should return with error message in this case.
@@ -470,57 +470,52 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_d
3. Fetch OTA image over HTTPS 3. Fetch OTA image over HTTPS
4. Check working of anti_rollback feature 4. Check working of anti_rollback feature
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='anti_rollback') dut.serial.erase_flash()
Utility.console_log('Erasing the flash on the chip') dut.serial.flash()
# erase the flash
dut1.erase_flash()
server_port = 8001 server_port = 8001
# Original binary file generated after compilation # Original binary file generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_ota.bin'
# Modified firmware image to lower security version in its header. This is to enable negative test case # Modified firmware image to lower security version in its header. This is to enable negative test case
anti_rollback_bin_name = 'advanced_https_ota_lower_sec_version.bin' anti_rollback_bin_name = 'advanced_https_ota_lower_sec_version.bin'
# check and log bin size # 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)
file_size = os.path.getsize(binary_file) file_size = os.path.getsize(binary_file)
with open(binary_file, 'rb+') as f: with open(binary_file, 'rb+') as f:
with open(os.path.join(dut1.app.binary_path, anti_rollback_bin_name), 'wb+') as fo: with open(os.path.join(dut.app.binary_path, anti_rollback_bin_name), 'wb+') as fo:
fo.write(f.read(file_size)) fo.write(f.read(file_size))
# Change security_version to 0 for negative test case # Change security_version to 0 for negative test case
fo.seek(36) fo.seek(36)
fo.write(b'\x00') fo.write(b'\x00')
binary_file = os.path.join(dut1.app.binary_path, anti_rollback_bin_name) binary_file = os.path.join(dut.app.binary_path, anti_rollback_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app()
# Positive Case # Positive Case
dut1.expect('Loaded app from partition at offset', timeout=30) dut.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
# Use originally generated image with secure_version=1 # Use originally generated image with secure_version=1
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60) dut.expect('Loaded app from partition at offset', timeout=60)
dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30) dut.expect(r' (sta|eth) ip: ([^,]+),', timeout=30)
dut1.expect('App is valid, rollback cancelled successfully', 30) dut.expect(r'App is valid, rollback cancelled successfully', timeout=30)
# Negative Case # Negative Case
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
# Use modified image with secure_version=0 # 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))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name)
dut1.expect('New firmware security version is less than eFuse programmed, 0 < 1', timeout=30) dut.expect('New firmware security version is less than eFuse programmed, 0 < 1', timeout=30)
try: try:
os.remove(binary_file) os.remove(binary_file)
except OSError: except OSError:
@@ -528,8 +523,13 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_d
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_partial_request(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
@pytest.mark.parametrize('config', ['partial_download',], indirect=True)
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. This is a positive test case, to test OTA workflow with Range HTTP header.
steps: | steps: |
@@ -537,46 +537,46 @@ def test_examples_protocol_advanced_https_ota_example_partial_request(env, extra
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='partial_download')
server_port = 8001 server_port = 8001
# Size of partial HTTP request # Size of partial HTTP request
request_size = 16384 request_size = 16384
# File to be downloaded. This file is generated after compilation # File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_ota.bin'
# check and log bin size binary_file = os.path.join(dut.app.binary_path, bin_name)
binary_file = os.path.join(dut1.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file) bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
http_requests = int((bin_size / request_size) - 1) http_requests = int((bin_size / request_size) - 1)
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
Utility.console_log('ENV_TEST_FAILURE: Cannot connect to AP') print('ENV_TEST_FAILURE: Cannot connect to AP')
thread1.terminate() thread1.terminate()
raise raise
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
for _ in range(http_requests): for _ in range(http_requests):
dut1.expect('Connection closed', timeout=60) dut.expect('Connection closed', timeout=60)
dut1.expect('Loaded app from partition at offset', timeout=60) dut.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
dut1.reset()
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='Example_WIFI_OTA', nightly_run=True) @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_nimble_gatts(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_ota
@pytest.mark.parametrize('config', ['nimble',], indirect=True)
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. Run an OTA image update while a BLE GATT Server is running in background. This GATT server will be using NimBLE Host stack.
steps: | steps: |
@@ -585,42 +585,40 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(env, extra_da
3. Fetch OTA image over HTTPS 3. Fetch OTA image over HTTPS
4. Reboot with the new OTA image 4. Reboot with the new OTA image
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='nimble')
server_port = 8001 server_port = 8001
# File to be downloaded. This file is generated after compilation # File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_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('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30) ip_address = dut.expect(r' sta ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
print('Started GAP advertising.') print('Started GAP advertising.')
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60) dut.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
dut1.reset()
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='Example_WIFI_OTA', nightly_run=True) @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_ota
@pytest.mark.parametrize('config', ['bluedroid',], indirect=True)
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. Run an OTA image update while a BLE GATT Server is running in background. This GATT server will be using Bluedroid Host stack.
steps: | steps: |
@@ -629,43 +627,40 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(env, extra
3. Fetch OTA image over HTTPS 3. Fetch OTA image over HTTPS
4. Reboot with the new OTA image 4. Reboot with the new OTA image
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='bluedroid')
server_port = 8001 server_port = 8001
# File to be downloaded. This file is generated after compilation # File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_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('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: try:
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30) ip_address = dut.expect(r' sta ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut1.expect('Started advertising.', timeout=30) dut.expect('Started advertising.', timeout=30)
print('Started GAP advertising.') print('Started GAP advertising.')
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60) dut.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
dut1.reset()
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
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 This is a test case for esp_http_client_read with binary size multiple of 289 bytes
steps: | steps: |
@@ -673,55 +668,38 @@ def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(env, e
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
# Original binary file generated after compilation # Original binary file generated after compilation
bin_name = 'advanced_https_ota.bin' bin_name = 'advanced_https_ota.bin'
# Binary file aligned to DEFAULT_OTA_BUF_SIZE(289 bytes) boundary # Binary file aligned to DEFAULT_OTA_BUF_SIZE(289 bytes) boundary
aligned_bin_name = 'aligned.bin' aligned_bin_name = 'aligned.bin'
# check and log bin size # 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)
# Original binary size # Original binary size
bin_size = os.path.getsize(binary_file) bin_size = os.path.getsize(binary_file)
# Dummy data required to align binary size to 289 bytes boundary # Dummy data required to align binary size to 289 bytes boundary
dummy_data_size = 289 - (bin_size % 289) dummy_data_size = 289 - (bin_size % 289)
with open(binary_file, 'rb+') as f: with open(binary_file, 'rb+') as f:
with open(os.path.join(dut1.app.binary_path, aligned_bin_name), 'wb+') as fo: with open(os.path.join(dut.app.binary_path, aligned_bin_name), 'wb+') as fo:
fo.write(f.read(bin_size)) fo.write(f.read(bin_size))
for _ in range(dummy_data_size): for _ in range(dummy_data_size):
fo.write(struct.pack('B', random.randrange(0,255,1))) fo.write(struct.pack('B', random.randrange(0,255,1)))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
chunked_server = start_chunked_server(dut1.app.binary_path, 8070) chunked_server = start_chunked_server(dut.app.binary_path, 8070)
dut1.start_app() dut.expect('Loaded app from partition at offset', timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) 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') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8070/' + aligned_bin_name)) print('writing to device: {}'.format('https://' + host_ip + ':8070/' + aligned_bin_name))
dut1.write('https://' + host_ip + ':8070/' + aligned_bin_name) dut.write('https://' + host_ip + ':8070/' + aligned_bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60) dut.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting Advanced OTA example', timeout=30) dut.expect('Starting Advanced OTA example', timeout=30)
chunked_server.kill() chunked_server.kill()
try: try:
os.remove(aligned_bin_name) os.remove(aligned_bin_name)
except OSError: except OSError:
pass pass
if __name__ == '__main__':
test_examples_protocol_advanced_https_ota_example()
test_examples_protocol_advanced_https_ota_example_chunked()
test_examples_protocol_advanced_https_ota_example_redirect_url()
test_examples_protocol_advanced_https_ota_example_truncated_bin()
test_examples_protocol_advanced_https_ota_example_truncated_header()
test_examples_protocol_advanced_https_ota_example_random()
test_examples_protocol_advanced_https_ota_example_invalid_chip_id()
test_examples_protocol_advanced_https_ota_example_anti_rollback()
test_examples_protocol_advanced_https_ota_example_partial_request()
test_examples_protocol_advanced_https_ota_example_nimble_gatts()
test_examples_protocol_advanced_https_ota_example_bluedroid_gatts()
test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin()

23
examples/system/ota/otatool/example_test.py → examples/system/ota/otatool/pytest_otatool.py
View File

@@ -1,35 +1,30 @@
# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Unlicense OR CC0-1.0
from __future__ import print_function from __future__ import print_function
import os import os
import subprocess import subprocess
import sys import sys
import ttfw_idf import pytest
from pytest_embedded import Dut
@ttfw_idf.idf_example_test(env_tag='Example_WIFI') @pytest.mark.supported_targets
def test_otatool_example(env, extra_data): @pytest.mark.wifi
dut = env.get_dut('otatool', 'examples/system/ota/otatool', dut_class=ttfw_idf.ESP32DUT) def test_otatool_example(dut: Dut) -> None:
# Verify factory firmware # Verify factory firmware
dut.start_app()
dut.expect('OTA Tool Example') dut.expect('OTA Tool Example')
dut.expect('Example end') dut.expect('Example end')
# Close connection to DUT # Close connection to DUT
dut.receive_thread.exit() dut.serial.proc.close()
dut.port_inst.close()
script_path = os.path.join(os.getenv('IDF_PATH'), 'examples', 'system', 'ota', 'otatool', 'otatool_example.py') script_path = os.path.join(str(os.getenv('IDF_PATH')), 'examples', 'system', 'ota', 'otatool', 'otatool_example.py')
binary_path = '' binary_path = ''
for flash_file in dut.app.flash_files: for flash_file in dut.app.flash_files:
if 'otatool.bin' in flash_file[1]: if 'otatool.bin' in flash_file[1]:
binary_path = flash_file[1] binary_path = flash_file[1]
break break
subprocess.check_call([sys.executable, script_path, '--binary', binary_path]) subprocess.check_call([sys.executable, script_path, '--binary', binary_path])
if __name__ == '__main__':
test_otatool_example()

3
examples/system/ota/pre_encrypted_ota/README.md
View File

@@ -1,4 +1,5 @@
| Supported Targets | ESP32 | ESP32-S2 | ESP32-S3 | ESP32-C3 |
| ----------------- | ----- | -------- | -------- | -------- |
# Encrypted Binary OTA # Encrypted Binary OTA
This example demonstrates OTA updates with pre-encrypted binary using `esp_encrypted_img` component's APIs and tool. This example demonstrates OTA updates with pre-encrypted binary using `esp_encrypted_img` component's APIs and tool.

57
examples/system/ota/pre_encrypted_ota/example_test.py → examples/system/ota/pre_encrypted_ota/pytest_pre_encrypted_ota.py
View File

@@ -3,36 +3,37 @@
import http.server import http.server
import multiprocessing import multiprocessing
import os import os
import re
import socket import socket
import ssl import ssl
from typing import Any from typing import Callable
import ttfw_idf import pexpect
import pytest
from pytest_embedded import Dut
from RangeHTTPServer import RangeRequestHandler from RangeHTTPServer import RangeRequestHandler
from tiny_test_fw import DUT
server_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'server_certs/ca_cert.pem') server_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'server_certs/ca_cert.pem')
key_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'server_certs/server_key.pem') key_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'server_certs/server_key.pem')
enc_bin_name = 'pre_encrypted_ota_secure.bin' enc_bin_name = 'pre_encrypted_ota_secure.bin'
def get_my_ip() -> Any: def get_my_ip() -> str:
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect(('8.8.8.8', 80)) s1.connect(('8.8.8.8', 80))
my_ip = ''
my_ip = s1.getsockname()[0] my_ip = s1.getsockname()[0]
s1.close() s1.close()
return my_ip return my_ip
def get_server_status(host_ip: Any, port: int) -> bool: def get_server_status(host_ip: str, port: int) -> bool:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_status = sock.connect_ex((host_ip, port)) server_status = sock.connect_ex((host_ip, port))
sock.close() sock.close()
return server_status == 0 return server_status == 0
def https_request_handler(): # type: ignore def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]:
""" """
Returns a request handler class that handles broken pipe exception Returns a request handler class that handles broken pipe exception
""" """
@@ -55,7 +56,7 @@ def https_request_handler(): # type: ignore
return RequestHandler return RequestHandler
def start_https_server(ota_image_dir: str, server_ip: Any, server_port: int) -> None: def start_https_server(ota_image_dir: str, server_ip: str, server_port: int) -> None:
os.chdir(ota_image_dir) os.chdir(ota_image_dir)
requestHandler = https_request_handler() requestHandler = https_request_handler()
httpd = http.server.HTTPServer((server_ip, server_port), requestHandler) httpd = http.server.HTTPServer((server_ip, server_port), requestHandler)
@@ -66,42 +67,32 @@ def start_https_server(ota_image_dir: str, server_ip: Any, server_port: int) ->
httpd.serve_forever() httpd.serve_forever()
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA') @pytest.mark.esp32
def test_examples_protocol_pre_encrypted_ota_example(env, extra_data): # type: ignore @pytest.mark.esp32c3
dut1 = env.get_dut('pre_encrypted_ota_example', 'examples/system/ota/pre_encrypted_ota', dut_class=ttfw_idf.ESP32DUT) @pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
def test_examples_protocol_pre_encrypted_ota_example(dut: Dut) -> None:
server_port = 8001 server_port = 8001
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False): 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.daemon = True
thread1.start() thread1.start()
artifacts = dut1.app.artifact_cls(dut1.app.idf_path, dut.expect('Loaded app from partition at offset', timeout=30)
dut1.app.case_group.get_artifact_index_file(),
dut1.app.app_path, dut1.app.config_name, dut1.app.target)
artifacts.download_artifact_files(file_names=['pre_encrypted_ota_secure.bin'])
dut1.start_app()
dut1.expect('Loaded app from partition at offset', timeout=30)
try: 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)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except pexpect.exceptions.TIMEOUT:
thread1.terminate() thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
dut1.expect('Starting Pre Encrypted OTA example', timeout=30) dut.expect('Starting Pre Encrypted OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + enc_bin_name)) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + enc_bin_name))
dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + enc_bin_name) dut.write('https://' + host_ip + ':' + str(server_port) + '/' + enc_bin_name)
dut1.expect('Magic Verified', timeout=30) dut.expect('Magic Verified', timeout=30)
dut1.expect('Reading RSA private key', timeout=30) dut.expect('Reading RSA private key', timeout=30)
dut1.expect('upgrade successful. Rebooting', timeout=30) dut.expect('upgrade successful. Rebooting', timeout=30)
thread1.terminate() thread1.terminate()
if __name__ == '__main__':
test_examples_protocol_pre_encrypted_ota_example()

2
examples/system/ota/simple_ota_example/README.md
View File

@@ -1,3 +1,5 @@
| Supported Targets | ESP32 | ESP32-S2 | ESP32-S3 | ESP32-C3 |
| ----------------- | ----- | -------- | -------- | -------- |
# Simple OTA example # Simple OTA example
This example is based on `esp_https_ota` component's APIs. This example is based on `esp_https_ota` component's APIs.

295
examples/system/ota/simple_ota_example/example_test.py → examples/system/ota/simple_ota_example/pytest_simple_ota.py
View File

@@ -1,13 +1,16 @@
# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Unlicense OR CC0-1.0
import http.server import http.server
import multiprocessing import multiprocessing
import os import os
import re
import socket import socket
import ssl import ssl
import sys import sys
from typing import Tuple
import ttfw_idf import pexpect
from tiny_test_fw import DUT, Utility import pytest
from pytest_embedded import Dut
server_cert = '-----BEGIN CERTIFICATE-----\n' \ server_cert = '-----BEGIN CERTIFICATE-----\n' \
'MIIDWDCCAkACCQCbF4+gVh/MLjANBgkqhkiG9w0BAQsFADBuMQswCQYDVQQGEwJJ\n'\ 'MIIDWDCCAkACCQCbF4+gVh/MLjANBgkqhkiG9w0BAQsFADBuMQswCQYDVQQGEwJJ\n'\
@@ -60,15 +63,16 @@ server_key = '-----BEGIN PRIVATE KEY-----\n'\
'-----END 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 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect(('8.8.8.8', 80)) s1.connect(('8.8.8.8', 80))
my_ip = ''
my_ip = s1.getsockname()[0] my_ip = s1.getsockname()[0]
s1.close() s1.close()
return my_ip return my_ip
def get_server_status(host_ip, server_port): def get_server_status(host_ip: str, server_port: int) -> bool:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_status = sock.connect_ex((host_ip, server_port)) server_status = sock.connect_ex((host_ip, server_port))
sock.close() sock.close()
@@ -77,7 +81,7 @@ def get_server_status(host_ip, server_port):
return False return False
def start_https_server(ota_image_dir, server_ip, server_port, server_file=None, key_file=None): def start_https_server(ota_image_dir: str, server_ip: str, server_port: int, server_file: str = None, key_file: str = None) -> None:
os.chdir(ota_image_dir) os.chdir(ota_image_dir)
if server_file is None: if server_file is None:
@@ -100,274 +104,260 @@ def start_https_server(ota_image_dir, server_ip, server_port, server_file=None,
httpd.serve_forever() httpd.serve_forever()
def check_sha256(sha256_expected, sha256_reported): def check_sha256(sha256_expected: str, sha256_reported: str) -> None:
Utility.console_log('sha256_expected: %s' % (sha256_expected)) print('sha256_expected: %s' % (sha256_expected))
Utility.console_log('sha256_reported: %s' % (sha256_reported)) print('sha256_reported: %s' % (sha256_reported))
if sha256_reported not in sha256_expected: if sha256_expected not in sha256_reported:
raise ValueError('SHA256 mismatch') raise ValueError('SHA256 mismatch')
else: else:
Utility.console_log('SHA256 expected and reported are the same') print('SHA256 expected and reported are the same')
def calc_all_sha256(dut): def calc_all_sha256(dut: Dut) -> Tuple[str, str]:
bootloader_path = os.path.join(dut.app.binary_path, 'bootloader', 'bootloader.bin') bootloader_path = os.path.join(dut.app.binary_path, 'bootloader', 'bootloader.bin')
output = dut.image_info(bootloader_path) sha256_bootloader = dut.app.get_sha256(bootloader_path)
sha256_bootloader = re.search(r'Validation Hash:\s+([a-f0-9]+)', output).group(1)
Utility.console_log('bootloader SHA256: %s' % sha256_bootloader)
app_path = os.path.join(dut.app.binary_path, 'simple_ota.bin') app_path = os.path.join(dut.app.binary_path, 'simple_ota.bin')
output = dut.image_info(app_path) sha256_app = dut.app.get_sha256(app_path)
sha256_app = re.search(r'Validation Hash:\s+([a-f0-9]+)', output).group(1)
Utility.console_log('app SHA256: %s' % sha256_app)
return sha256_bootloader, sha256_app return str(sha256_bootloader), str(sha256_app)
@ttfw_idf.idf_example_test(env_tag='Example_WIFI_OTA', nightly_run=True) @pytest.mark.esp32
def test_examples_protocol_simple_ota_example(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_ota
def test_examples_protocol_simple_ota_example(dut: Dut) -> None:
""" """
steps: | steps: |
1. join AP 1. join AP
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', dut_class=ttfw_idf.ESP32DUT) sha256_bootloader, sha256_app = calc_all_sha256(dut)
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('simple_ota_bin_size', '{}KB'.format(bin_size // 1024))
sha256_bootloader, sha256_app = calc_all_sha256(dut1)
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, 8000) is False): if (get_server_status(host_ip, 8000) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1.daemon = True thread1.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset 0x10000', timeout=30)
dut1.expect('Loaded app from partition at offset 0x10000', timeout=30) check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
check_sha256(sha256_bootloader, dut1.expect(re.compile(r'SHA-256 for bootloader:\s+([a-f0-9]+)'))[0]) check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
check_sha256(sha256_app, dut1.expect(re.compile(r'SHA-256 for current firmware:\s+([a-f0-9]+)'))[0])
try: try:
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30) ip_address = dut.expect(r' sta ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address)) 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')
thread1.terminate() 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 + ':8000/simple_ota.bin')) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut1.write('https://' + host_ip + ':8000/simple_ota.bin') dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut1.expect('Loaded app from partition at offset 0x110000', timeout=60) dut.expect('Loaded app from partition at offset 0x110000', timeout=60)
dut1.expect('Starting OTA example', timeout=30) dut.expect('Starting OTA example', timeout=30)
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1') @pytest.mark.esp32
def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
@pytest.mark.parametrize('config', ['spiram',], indirect=True)
def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(dut: Dut) -> None:
""" """
steps: | steps: |
1. join AP 1. join AP
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', dut_class=ttfw_idf.ESP32DUT, app_config_name='spiram')
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('simple_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, 8000) is False): if (get_server_status(host_ip, 8000) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1.daemon = True thread1.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset 0x10000', timeout=30)
dut1.expect('Loaded app from partition at offset 0x10000', timeout=30)
try: try:
ip_address = dut1.expect(re.compile(r' eth ip: ([^,]+),'), timeout=30) ip_address = dut.expect(r' eth ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address)) 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')
thread1.terminate() 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 + ':8000/simple_ota.bin')) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut1.write('https://' + host_ip + ':8000/simple_ota.bin') dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut1.expect('Loaded app from partition at offset 0x110000', timeout=60) dut.expect('Loaded app from partition at offset 0x110000', timeout=60)
dut1.expect('Starting OTA example', timeout=30) dut.expect('Starting OTA example', timeout=30)
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='Example_Flash_Encryption_OTA') @pytest.mark.esp32
def test_examples_protocol_simple_ota_example_with_flash_encryption(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.flash_encryption_ota
@pytest.mark.parametrize('config', ['flash_enc',], indirect=True)
@pytest.mark.parametrize('skip_autoflash', ['y'], indirect=True)
def test_examples_protocol_simple_ota_example_with_flash_encryption(dut: Dut) -> None:
""" """
steps: | steps: |
1. join AP 1. join AP
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', dut_class=ttfw_idf.ESP32DUT, app_config_name='flash_enc') # Erase flash
# check and log bin size dut.serial.erase_flash()
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin') dut.serial.flash()
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('simple_ota_bin_size', '{}KB'.format(bin_size // 1024))
# erase flash on the device
print('Erasing the flash in order to have an empty NVS key partiton')
dut1.erase_flash()
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, 8000) is False): if (get_server_status(host_ip, 8000) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1.daemon = True thread1.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
dut1.expect('Loaded app from partition at offset 0x20000', timeout=30) dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
dut1.expect('Flash encryption mode is DEVELOPMENT (not secure)', timeout=10)
try: try:
ip_address = dut1.expect(re.compile(r' eth ip: ([^,]+),'), timeout=30) ip_address = dut.expect(r' eth ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address)) 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')
thread1.terminate() 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 + ':8000/simple_ota.bin')) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut1.write('https://' + host_ip + ':8000/simple_ota.bin') dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut1.expect('Loaded app from partition at offset 0x120000', timeout=60) dut.expect('Loaded app from partition at offset 0x120000', timeout=60)
dut1.expect('Flash encryption mode is DEVELOPMENT (not secure)', timeout=10) dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
dut1.expect('Starting OTA example', timeout=30) dut.expect('Starting OTA example', timeout=30)
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='Example_Flash_Encryption_OTA_WiFi', target=['esp32c3'], nightly_run=True) @pytest.mark.esp32c3
def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(env, extra_data): @pytest.mark.flash_encryption_wifi_ota
@pytest.mark.parametrize('config', ['flash_enc_wifi',], indirect=True)
@pytest.mark.parametrize('skip_autoflash', ['y'], indirect=True)
def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(dut: Dut) -> None:
""" """
steps: | steps: |
1. join AP 1. join AP
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', app_config_name='flash_enc_wifi')
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('simple_ota_bin_size', '{}KB'.format(bin_size // 1024))
# erase flash on the device
print('Erasing the flash in order to have an empty NVS key partiton')
dut1.erase_flash()
# start test # start test
# Erase flash
dut.serial.erase_flash()
dut.serial.flash()
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, 8000) is False): if (get_server_status(host_ip, 8000) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1.daemon = True thread1.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
dut1.expect('Loaded app from partition at offset 0x20000', timeout=30) dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
dut1.expect('Flash encryption mode is DEVELOPMENT (not secure)', timeout=10)
try: try:
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30) ip_address = dut.expect(r' sta ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address)) 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')
thread1.terminate() 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 + ':8000/simple_ota.bin')) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut1.write('https://' + host_ip + ':8000/simple_ota.bin') dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut1.expect('Loaded app from partition at offset 0x120000', timeout=60) dut.expect('Loaded app from partition at offset 0x120000', timeout=60)
dut1.expect('Flash encryption mode is DEVELOPMENT (not secure)', timeout=10) dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
dut1.expect('Starting OTA example', timeout=30) dut.expect('Starting OTA example', timeout=30)
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1') @pytest.mark.esp32
def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_update_no_secure_boot_ecdsa(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_ota
@pytest.mark.parametrize('config', ['on_update_no_sb_ecdsa',], indirect=True)
def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_update_no_secure_boot_ecdsa(dut: Dut) -> None:
""" """
steps: | steps: |
1. join AP 1. join AP
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', dut_class=ttfw_idf.ESP32DUT, sha256_bootloader, sha256_app = calc_all_sha256(dut)
app_config_name='on_update_no_sb_ecdsa')
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('simple_ota_bin_size', '{}KB'.format(bin_size // 1024))
sha256_bootloader, sha256_app = calc_all_sha256(dut1)
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, 8000) is False): if (get_server_status(host_ip, 8000) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1.daemon = True thread1.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
dut1.expect('Loaded app from partition at offset 0x20000', timeout=30) check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
check_sha256(sha256_bootloader, dut1.expect(re.compile(r'SHA-256 for bootloader:\s+([a-f0-9]+)'))[0]) check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
check_sha256(sha256_app, dut1.expect(re.compile(r'SHA-256 for current firmware:\s+([a-f0-9]+)'))[0])
try: try:
ip_address = dut1.expect(re.compile(r' eth ip: ([^,]+),'), timeout=30) ip_address = dut.expect(r' eth ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address)) 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')
thread1.terminate() 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 + ':8000/simple_ota.bin')) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut1.write('https://' + host_ip + ':8000/simple_ota.bin') dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut1.expect('Writing to partition subtype 16 at offset 0x120000', timeout=20) dut.expect('Writing to partition subtype 16 at offset 0x120000', timeout=20)
dut1.expect('Verifying image signature...', timeout=60) dut.expect('Verifying image signature...', timeout=60)
dut1.expect('Loaded app from partition at offset 0x120000', timeout=20) dut.expect('Loaded app from partition at offset 0x120000', timeout=20)
dut1.expect('Starting OTA example', timeout=30) dut.expect('Starting OTA example', timeout=30)
thread1.terminate() thread1.terminate()
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV12') @pytest.mark.esp32
def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_update_no_secure_boot_rsa(env, extra_data): @pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet
@pytest.mark.parametrize('config', ['on_update_no_sb_rsa',], indirect=True)
def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_update_no_secure_boot_rsa(dut: Dut) -> None:
""" """
steps: | steps: |
1. join AP 1. join AP
2. Fetch OTA image over HTTPS 2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image 3. Reboot with the new OTA image
""" """
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', dut_class=ttfw_idf.ESP32DUT, sha256_bootloader, sha256_app = calc_all_sha256(dut)
app_config_name='on_update_no_sb_rsa')
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('simple_ota_bin_size', '{}KB'.format(bin_size // 1024))
sha256_bootloader, sha256_app = calc_all_sha256(dut1)
# start test # start test
host_ip = get_my_ip() host_ip = get_my_ip()
if (get_server_status(host_ip, 8000) is False): if (get_server_status(host_ip, 8000) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1.daemon = True thread1.daemon = True
thread1.start() thread1.start()
dut1.start_app() dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
dut1.expect('Loaded app from partition at offset 0x20000', timeout=30) check_sha256(sha256_bootloader, str(dut.expect(r'SHA-256 for bootloader:\s+([a-f0-9]){64}')[0]))
check_sha256(sha256_bootloader, dut1.expect(re.compile(r'SHA-256 for bootloader:\s+([a-f0-9]+)'))[0]) check_sha256(sha256_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
check_sha256(sha256_app, dut1.expect(re.compile(r'SHA-256 for current firmware:\s+([a-f0-9]+)'))[0])
try: try:
ip_address = dut1.expect(re.compile(r' eth ip: ([^,]+),'), timeout=30) ip_address = dut.expect(r' eth ip: ([^,]+),', timeout=30)
print('Connected to AP with IP: {}'.format(ip_address)) 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')
thread1.terminate() 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 + ':8000/simple_ota.bin')) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut1.write('https://' + host_ip + ':8000/simple_ota.bin') dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut1.expect('Writing to partition subtype 16 at offset 0x120000', timeout=20) dut.expect('Writing to partition subtype 16 at offset 0x120000', timeout=20)
dut1.expect('Verifying image signature...', timeout=60) dut.expect('Verifying image signature...', timeout=60)
dut1.expect('#0 app key digest == #0 trusted key digest', timeout=10) dut.expect('#0 app key digest == #0 trusted key digest', timeout=10)
dut1.expect('Verifying with RSA-PSS...', timeout=10) dut.expect('Verifying with RSA-PSS...', timeout=10)
dut1.expect('Signature verified successfully!', timeout=10) dut.expect('Signature verified successfully!', timeout=10)
dut1.expect('Loaded app from partition at offset 0x120000', timeout=20) dut.expect('Loaded app from partition at offset 0x120000', timeout=20)
dut1.expect('Starting OTA example', timeout=30) dut.expect('Starting OTA example', timeout=30)
thread1.terminate() thread1.terminate()
@@ -382,10 +372,3 @@ if __name__ == '__main__':
start_https_server(bin_dir, '', port, start_https_server(bin_dir, '', port,
server_file=os.path.join(cert_dir, 'ca_cert.pem'), server_file=os.path.join(cert_dir, 'ca_cert.pem'),
key_file=os.path.join(cert_dir, 'ca_key.pem')) key_file=os.path.join(cert_dir, 'ca_key.pem'))
else:
test_examples_protocol_simple_ota_example()
test_examples_protocol_simple_ota_example_ethernet_with_spiram_config()
test_examples_protocol_simple_ota_example_with_flash_encryption()
test_examples_protocol_simple_ota_example_with_flash_encryption_wifi()
test_examples_protocol_simple_ota_example_with_verify_app_signature_on_update_no_secure_boot_ecdsa()
test_examples_protocol_simple_ota_example_with_verify_app_signature_on_update_no_secure_boot_rsa()

5
pytest.ini
View File

@@ -36,6 +36,11 @@ markers =
ethernet_ota: ethernet OTA runners ethernet_ota: ethernet OTA runners
flash_encryption: Flash Encryption runners flash_encryption: Flash Encryption runners
ir_transceiver: runners with a pair of IR transmitter and receiver ir_transceiver: runners with a pair of IR transmitter and receiver
wifi_ota: wifi OTA runners
flash_encryption_ota: flash encryprion ota ethernet runner
flash_encryption_wifi_ota: flash encryprion ota wifi runner
ethernet: ethernet runner
ethernet_flash_8m: ethernet runner with 8mb flash
wifi: wifi runner wifi: wifi runner
# multi-dut markers # multi-dut markers