alex/esp-idf
alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2024-10-05 20:47:46 -04:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

CI: Improve common test methods

Browse Source 
also fix ota test cases
This commit is contained in:
Chen Yudong 2022-07-07 00:34:06 +08:00
parent 692fbc169c
commit 2f75733ad7
46 changed files with 700 additions and 684 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitlab/ci/rules.yml
View File

@ -66,7 +66,6 @@
- "tools/ci/python_packages/gitlab_api.py"
- "tools/ci/python_packages/tiny_test_fw/**/*"
- "tools/ci/python_packages/ttfw_idf/**/*"
- "tools/ci/python_packages/common_test_methods.py"
- "tools/unit-test-app/**/*"
@ -187,6 +186,7 @@
.patterns-example_test-related_changes-wifi: &patterns-example_test-related_changes-wifi
- "components/esp_wifi/lib"
- "components/lwip/**/*"
- "examples/protocols/**/*"
- "examples/wifi/**/*"

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

@ -13,12 +13,12 @@
expire_in: 1 week
variables:
GIT_DEPTH: 1
SUBMODULES_TO_FETCH: "none"
script:
- retry_failed git clone $KNOWN_FAILURE_CASES_REPO known_failure_cases
# get runner env config file
- retry_failed git clone $TEST_ENV_CONFIG_REPO
- python $CHECKOUT_REF_SCRIPT ci-test-runner-configs ci-test-runner-configs
- cp ./ci-test-runner-configs/${CI_RUNNER_DESCRIPTION}/EnvConfig.yml EnvConfig.yml || true
# using runner tags as markers to filter the test cases
# Runner tags are comma separated, replace the comma with " and " for markers
- job_tags=$(python tools/ci/python_packages/gitlab_api.py get_job_tags $CI_PROJECT_ID --job_id $CI_JOB_ID)
@ -86,37 +86,37 @@ example_test_pytest_esp32c3_flash_suspend:
- build_pytest_examples_esp32c3
tags: [ esp32c3, flash_suspend ]
example_test_pytest_esp32_ethernet_router:
example_test_pytest_esp32_ethernet_ota:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32-ota-related_changes
needs:
- build_pytest_examples_esp32
tags: [ esp32, ethernet_router ]
tags: [ esp32, ethernet_ota ]
example_test_pytest_esp32_wifi_ota:
example_test_pytest_esp32_wifi_high_traffic:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32
needs:
- build_pytest_examples_esp32
tags: [ esp32, wifi_high_traffic ]
example_test_pytest_esp32_flash_encryption_wifi_high_traffic:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32-ota-related_changes-include_nightly_run
needs:
- build_pytest_examples_esp32
tags: [ esp32, wifi_ota ]
tags: [ esp32, flash_encryption_wifi_high_traffic ]
example_test_pytest_esp32_flash_encryption_ota:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32-ota-related_changes
needs:
- build_pytest_examples_esp32
tags: [ esp32, flash_encryption_ota ]
example_test_pytest_esp32c3_flash_encryption_wifi_ota:
example_test_pytest_esp32c3_flash_encryption_wifi_high_traffic:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32c3-ota-related_changes-include_nightly_run
needs:
- build_pytest_examples_esp32c3
tags: [ esp32c3, flash_encryption_wifi_ota ]
tags: [ esp32c3, flash_encryption_wifi_high_traffic ]
example_test_pytest_esp32_ethernet:
extends:
@ -134,13 +134,13 @@ example_test_pytest_esp32_8mb_flash:
- build_pytest_examples_esp32
tags: [ esp32, ethernet_flash_8m ]
example_test_pytest_esp32_wifi_nearby:
example_test_pytest_esp32_wifi_ap:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32-wifi-related_changes
- .rules:test:example_test-esp32
needs:
- build_pytest_examples_esp32
tags: [ esp32, wifi_nearby ]
tags: [ esp32, wifi_ap ]
example_test_pytest_esp32_wifi_router:
extends:
@ -153,7 +153,7 @@ example_test_pytest_esp32_wifi_router:
example_test_pytest_esp32_wifi_wlan:
extends:
- .pytest_examples_dir_template
- .rules:test:example_test-esp32
- .rules:test:example_test-esp32-wifi-related_changes
needs:
- build_pytest_examples_esp32
tags: [ esp32, wifi_wlan ]
@ -394,7 +394,7 @@ test_weekend_mqtt:
- .rules:labels:weekend_test
tags:
- ESP32
- Example_EthKitV1
- ethernet_router
script:
- export MQTT_PUBLISH_TEST=1
- export TEST_PATH=$CI_PROJECT_DIR/tools/test_apps/protocols/mqtt/publish_connect_test
@ -426,12 +426,6 @@ test_weekend_mqtt:
- .example_test_template
- .rules:test:example_test-esp32s3
example_test_001B:
extends: .example_test_esp32_template
tags:
- ESP32
- Example_EthKitV1
example_test_001B_V3:
extends: .example_test_esp32_template
tags:
@ -452,12 +446,14 @@ example_test_protocols:
- wifi_router
example_test_002:
extends: .example_test_esp32_template
extends:
- .example_test_esp32_template
- .rules:test:example_test-esp32-wifi-related_changes
tags:
- ESP32
- Example_ShieldBox_Basic
example_test_ethernet:
example_test_ethernet_router:
extends: .example_test_esp32_template
tags:
- ESP32
@ -611,24 +607,12 @@ test_app_test_001:
variables:
SETUP_TOOLS: "1"
test_app_test_002:
extends: .test_app_esp32_template
tags:
- ESP32
- Example_WIFI
test_app_test_eth:
extends: .test_app_esp32_template
tags:
- ESP32
- ethernet_router
test_app_test_003:
extends: .test_app_esp32_template
tags:
- ESP32
- Example_PPP
test_app_test_004:
extends: .test_app_esp32s2_template
tags:

4
examples/common_components/protocol_examples_common/Kconfig.projbuild
View File

@ -22,7 +22,7 @@ menu "Example Connection Configuration"
default y
help
Provide wifi connect commands for esp_console.
Please use `register_wifi_connect_commands` to register them.
Please use `example_register_wifi_connect_commands` to register them.
config EXAMPLE_WIFI_SSID
depends on !EXAMPLE_WIFI_SSID_PWD_FROM_STDIN
@ -305,7 +305,7 @@ menu "Example Connection Configuration"
endif # EXAMPLE_CONNECT_ETHERNET
config EXAMPLE_CONNECT_IPV6
depends on (EXAMPLE_CONNECT_WIFI || EXAMPLE_CONNECT_ETHERNET)
depends on EXAMPLE_CONNECT_WIFI || EXAMPLE_CONNECT_ETHERNET
bool "Obtain IPv6 address"
default y
select LWIP_IPV6

2
examples/common_components/protocol_examples_common/console_cmd.c
View File

@ -58,7 +58,7 @@ static int cmd_do_wifi_disconnect(int argc, char **argv)
return 0;
}
void register_wifi_connect_commands(void)
void example_register_wifi_connect_commands(void)
{
ESP_LOGI(TAG, "Registering WiFi connect commands.");
example_wifi_start();

5
examples/common_components/protocol_examples_common/include/protocol_examples_common.h
View File

@ -81,9 +81,10 @@ esp_netif_t *get_example_netif_from_desc(const char *desc);
/**
* @brief Register wifi connect commands
*
* @note Provide wifi connect commands using esp_console.
* Provide a simple wifi_connect command in esp_console.
* This function can be used after esp_console is initialized.
*/
void register_wifi_connect_commands(void);
void example_register_wifi_connect_commands(void);
#endif
#if CONFIG_EXAMPLE_CONNECT_ETHERNET

18
examples/common_components/protocol_examples_common/wifi_connect.c
View File

@ -229,18 +229,20 @@ esp_err_t example_wifi_connect(void)
};
#if CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN
example_configure_stdin_stdout();
char buf[32+64+2] = {0};
char buf[sizeof(wifi_config.sta.ssid)+sizeof(wifi_config.sta.password)+2] = {0};
ESP_LOGI(TAG, "Please input ssid password:");
fgets(buf, sizeof(buf), stdin);
int len = strlen(buf);
buf[len-1] = '\0';
memset(wifi_config.sta.ssid, 0, 32);
char *temp = strtok(buf, " ");
strncpy((char*)wifi_config.sta.ssid, temp, 32);
memset(wifi_config.sta.password, 0, 64);
temp = strtok(NULL, " ");
buf[len-1] = '\0'; /* removes '\n' */
memset(wifi_config.sta.ssid, 0, sizeof(wifi_config.sta.ssid));
char *rest = NULL;
char *temp = strtok_r(buf, " ", &rest);
strncpy((char*)wifi_config.sta.ssid, temp, sizeof(wifi_config.sta.ssid));
memset(wifi_config.sta.password, 0, sizeof(wifi_config.sta.password));
temp = strtok_r(NULL, " ", &rest);
if (temp) {
strncpy((char*)wifi_config.sta.password, temp, 64);
strncpy((char*)wifi_config.sta.password, temp, sizeof(wifi_config.sta.password));
} else {
wifi_config.sta.threshold.authmode = WIFI_AUTH_OPEN;
}

6
examples/ethernet/iperf/iperf_test.py
View File

@ -16,7 +16,7 @@ import time
import netifaces
import ttfw_idf
from common_test_methods import get_env_config, get_my_ip_by_interface
from common_test_methods import get_env_config_variable, get_host_ip_by_interface
from idf_iperf_test_util import IperfUtility
from tiny_test_fw import TinyFW
@ -62,8 +62,8 @@ def test_ethernet_throughput_basic(env, _): # type: (Any, Any) -> None
1. test TCP tx rx and UDP tx rx throughput
2. compare with the pre-defined pass standard
"""
pc_nic = get_env_config('wifi_router').get('pc_nic', 'eth1')
pc_nic_ip = get_my_ip_by_interface(pc_nic, netifaces.AF_INET)
pc_nic = get_env_config_variable('wifi_router', 'pc_nic', default='eth1')
pc_nic_ip = get_host_ip_by_interface(pc_nic, netifaces.AF_INET)
pc_iperf_log_file = os.path.join(env.log_path, 'pc_iperf_log.md')
# 1. get DUT

10
examples/network/simple_sniffer/pytest_simple_sniffer.py
View File

@ -6,12 +6,12 @@ from pytest_embedded import Dut
@pytest.mark.esp32
@pytest.mark.wifi_nearby
@pytest.mark.wifi_ap
@pytest.mark.parametrize('config', [
'mem',
], indirect=True)
def test_examples_simple_sniffer(dut: Dut) -> None:
sniffer_num = 10
sniffer_num = 9
dut.expect('sniffer>')
dut.write('pcap --open -f simple-sniffer')
dut.expect('cmd_pcap: open file successfully')
@ -27,8 +27,8 @@ def test_examples_simple_sniffer(dut: Dut) -> None:
dut.expect(r'Minor Version: [0-9]+')
dut.expect(r'SnapLen: [0-9]+')
dut.expect(r'LinkType: [0-9]+')
# Allow "save captured packet failed" once
for i in range(0, sniffer_num - 1):
# Allow "save captured packet failed" twice
for i in range(0, sniffer_num - 2):
dut.expect(f'Packet {i}:')
dut.expect(r'Timestamp \(Seconds\): [0-9]+')
dut.expect(r'Timestamp \(Microseconds\): [0-9]+')
@ -38,7 +38,7 @@ def test_examples_simple_sniffer(dut: Dut) -> None:
dut.expect(r'Frame Subtype:\s+\w+')
dut.expect(r'Destination:\s+\w+')
dut.expect(r'Source:\s+\w+')
dut.expect(r'Pcap packet Number: \d+')
dut.expect(r'Pcap packet Number: [7-9]')
dut.write('pcap --close -f simple-sniffer')
dut.expect('cmd_pcap: free memory successfully')
dut.expect('cmd_pcap: .pcap file close done')

12
examples/protocols/esp_local_ctrl/pytest_esp_local_ctrl.py
View File

@ -9,7 +9,7 @@ import sys
import pexpect
import pytest
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
@ -47,12 +47,12 @@ def test_examples_esp_local_ctrl(dut: Dut) -> None:
idf_path = get_sdk_path()
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
dut_ip = dut.expect(r'IPv4 address: (\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})')[1].decode()
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
dut_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)')[1].decode()
dut.expect('esp_https_server: Starting server')
dut.expect('esp_https_server: Server listening on port 443')
dut.expect('control: esp_local_ctrl service started with name : my_esp_ctrl_device')

10
examples/protocols/http_server/advanced_tests/pytest_http_server_advanced.py
View File

@ -17,7 +17,7 @@ except ModuleNotFoundError:
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', 'tools', 'ci', 'python_packages'))
from idf_http_server_test import test as client
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
# When running on local machine execute the following before running this script
@ -49,11 +49,11 @@ def test_examples_protocol_http_server_advanced(dut: Dut) -> None:
# Parse IP address of STA
logging.info('Waiting to connect with AP')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
got_port = dut.expect(r"(?:[\s\S]*)Started HTTP server on port: '(\d+)'", timeout=30)[1].decode()

10
examples/protocols/http_server/persistent_sockets/pytest_http_server_persistence.py
View File

@ -19,7 +19,7 @@ except ModuleNotFoundError:
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', 'tools', 'ci', 'python_packages'))
from idf_http_server_test import adder as client
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
# When running on local machine execute the following before running this script
@ -45,11 +45,11 @@ def test_examples_protocol_http_server_persistence(dut: Dut) -> None:
# Parse IP address of STA
logging.info('Waiting to connect with AP')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
got_port = dut.expect(r"(?:[\s\S]*)Starting server on port: '(\d+)'", timeout=30)[1].decode()

16
examples/protocols/http_server/simple/pytest_http_server_simple.py
View File

@ -23,7 +23,7 @@ except ModuleNotFoundError:
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', 'tools', 'ci', 'python_packages'))
from idf_http_server_test import client
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
@ -77,10 +77,10 @@ def test_examples_protocol_http_server_simple(dut: Dut) -> None:
# Parse IP address of STA
logging.info('Waiting to connect with AP')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(' '.join([ap_ssid, ap_password]))
got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
got_port = dut.expect(r"(?:[\s\S]*)Starting server on port: '(\d+)'", timeout=30)[1].decode()
@ -151,11 +151,11 @@ def test_examples_protocol_http_server_lru_purge_enable(dut: Dut) -> None:
# Parse IP address of STA
logging.info('Waiting to connect with AP')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
got_port = dut.expect(r"(?:[\s\S]*)Starting server on port: '(\d+)'", timeout=30)[1].decode()

10
examples/protocols/http_server/ws_echo_server/pytest_ws_server_example.py
View File

@ -9,7 +9,7 @@ import logging
import os
import pytest
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
try:
@ -64,11 +64,11 @@ def test_examples_protocol_http_ws_echo_server(dut: Dut) -> None:
# Parse IP address of STA
logging.info('Waiting to connect with AP')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
got_port = dut.expect(r"Starting server on port: '(\d+)'", timeout=30)[1].decode()

92
examples/protocols/https_request/pytest_https_request.py
View File

@ -10,20 +10,11 @@ from typing import Callable
import pexpect
import pytest
from common_test_methods import get_my_ip4_by_dest_ip
from common_test_methods import get_host_ip4_by_dest_ip
from pytest_embedded import Dut
from RangeHTTPServer import RangeRequestHandler
def get_server_status(host_ip: str, port: int) -> bool:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_status = sock.connect_ex((host_ip, port))
sock.close()
if server_status == 0:
return True
return False
def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]:
"""
Returns a request handler class that handles broken pipe exception
@ -77,54 +68,53 @@ def test_examples_protocol_https_request_cli_session_tickets(dut: Dut) -> None:
binary_file = os.path.join(dut.app.binary_path, 'https_request.bin')
bin_size = os.path.getsize(binary_file)
logging.info('https_request_bin_size : {}KB'.format(bin_size // 1024))
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=60)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
# start https server
host_ip = get_my_ip4_by_dest_ip(ip_address)
server_port = 8070
server_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_cert.pem')
key_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_key.pem')
if (get_server_status(host_ip, server_port) is False):
thread1 = multiprocessing.Process(target=start_https_server, args=(server_file, key_file, host_ip, server_port))
thread1.daemon = True
thread1.start()
logging.info('The server started on {}:{}'.format(host_ip, server_port))
dut.expect('Start https_request example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port)))
dut.write('https://' + host_ip + ':' + str(server_port))
logging.info("Testing for \"https_request using saved session\"")
# Check for connection using already saved client session
thread1 = multiprocessing.Process(target=start_https_server, args=(server_file, key_file, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
logging.info('The server started on localhost:{}'.format(server_port))
try:
dut.expect('https_request to local server', timeout=30)
dut.expect(['Connection established...',
'Reading HTTP response...',
'HTTP/1.1 200 OK',
'connection closed'], expect_all=True)
except Exception:
logging.info("Failed to connect to local https server\"")
raise
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=60)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
host_ip = get_host_ip4_by_dest_ip(ip_address)
try:
dut.expect('https_request using saved client session', timeout=20)
dut.expect(['Connection established...',
'Reading HTTP response...',
'HTTP/1.1 200 OK',
'connection closed'], expect_all=True)
except Exception:
logging.info("Failed the test for \"https_request using saved client session\"")
raise
dut.expect('Start https_request example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port)))
dut.write('https://' + host_ip + ':' + str(server_port))
logging.info("Testing for \"https_request using saved session\"")
logging.info("Passed the test for \"https_request using saved client session\"")
thread1.terminate()
# Check for connection using already saved client session
try:
dut.expect('https_request to local server', timeout=30)
dut.expect(['Connection established...',
'Reading HTTP response...',
'HTTP/1.1 200 OK',
'connection closed'], expect_all=True)
except Exception:
logging.info("Failed to connect to local https server\"")
raise
try:
dut.expect('https_request using saved client session', timeout=20)
dut.expect(['Connection established...',
'Reading HTTP response...',
'HTTP/1.1 200 OK',
'connection closed'], expect_all=True)
except Exception:
logging.info("Failed the test for \"https_request using saved client session\"")
raise
logging.info("Passed the test for \"https_request using saved client session\"")
finally:
thread1.terminate()
@pytest.mark.esp32

24
examples/protocols/https_server/simple/pytest_https_server_simple.py
View File

@ -9,7 +9,7 @@ import os
import ssl
import pytest
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
server_cert_pem = '-----BEGIN CERTIFICATE-----\n'\
@ -109,11 +109,11 @@ def test_examples_protocol_https_server_simple(dut: Dut) -> None:
# start test
logging.info('Waiting to connect with AP')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
# Parse IP address and port of the server
dut.expect(r'Starting server')
got_port = int(dut.expect(r'Server listening on port (\d+)', timeout=30)[1].decode())
@ -182,11 +182,11 @@ def test_examples_protocol_https_server_simple_dynamic_buffers(dut: Dut) -> None
# start test
logging.info('Waiting to connect with AP')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
# Parse IP address and port of the server
dut.expect(r'Starting server')
got_port = int(dut.expect(r'Server listening on port (\d+)', timeout=30)[1].decode())
@ -230,9 +230,9 @@ def test_examples_protocol_https_server_simple_dynamic_buffers(dut: Dut) -> None
logging.info('Checking user callback: Obtaining client certificate...')
serial_number = dut.expect(r'serial number(.*)', timeout=5)[0]
issuer_name = dut.expect(r'issuer name(.*)', timeout=5)[0]
expiry = dut.expect(r'expires on ((.*)\d{4}\-(0?[1-9]|1[012])\-(0?[1-9]|[12][0-9]|3[01])*)', timeout=5)[1].decode()
serial_number = dut.expect(r'serial number\s*:([^\n]*)', timeout=5)[0]
issuer_name = dut.expect(r'issuer name\s*:([^\n]*)', timeout=5)[0]
expiry = dut.expect(r'expires on\s*:((.*)\d{4}\-(0?[1-9]|1[012])\-(0?[1-9]|[12][0-9]|3[01])*)', timeout=5)[1].decode()
logging.info('Serial No. : {}'.format(serial_number))
logging.info('Issuer Name : {}'.format(issuer_name))

12
examples/protocols/https_server/wss_server/pytest_https_wss_server.py
View File

@ -14,7 +14,7 @@ from typing import Any, Optional
import pytest
import websocket
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
OPCODE_TEXT = 0x1
@ -122,11 +122,11 @@ def test_examples_protocol_https_wss_server(dut: Dut) -> None:
logging.info('Waiting to connect with AP')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
# Parse IP address of STA
got_port = int(dut.expect(r'Server listening on port (\d+)', timeout=30)[1].decode())
got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
@ -146,7 +146,7 @@ def test_examples_protocol_https_wss_server(dut: Dut) -> None:
opcode, data = ws.read()
data = data.decode('UTF-8')
if data != DATA:
raise RuntimeError('Failed to receive the correct echo response')
raise RuntimeError(f'Failed to receive the correct echo response.')
logging.info('Correct echo response obtained from the wss server')
# Test for PING

22
examples/protocols/https_x509_bundle/pytest_https_x509_bundle.py
View File

@ -4,7 +4,7 @@ import logging
import os
import pytest
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
@ -12,7 +12,7 @@ from pytest_embedded import Dut
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_nearby
@pytest.mark.wifi_ap
def test_examples_protocol_https_x509_bundle(dut: Dut) -> None:
"""
steps: |
@ -26,11 +26,11 @@ def test_examples_protocol_https_x509_bundle(dut: Dut) -> None:
logging.info('https_x509_bundle_bin_size : {}KB'.format(bin_size // 1024))
# Connect to AP
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_nearby')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_ap'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)
# start test
num_URLS = int(dut.expect(r'Connecting to (\d+) URLs', timeout=30)[1].decode())
@ -42,7 +42,7 @@ def test_examples_protocol_https_x509_bundle(dut: Dut) -> None:
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_nearby
@pytest.mark.wifi_ap
@pytest.mark.parametrize('config', ['ssldyn',], indirect=True)
def test_examples_protocol_https_x509_bundle_dynamic_buffer(dut: Dut) -> None:
# test mbedtls dynamic resource
@ -52,11 +52,11 @@ def test_examples_protocol_https_x509_bundle_dynamic_buffer(dut: Dut) -> None:
logging.info('https_x509_bundle_bin_size : {}KB'.format(bin_size // 1024))
# Connect to AP
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_nearby')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_ap'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)
# start test
num_URLS = int(dut.expect(r'Connecting to (\d+) URLs', timeout=30)[1].decode())

2
examples/protocols/icmp_echo/main/echo_example_main.c
View File

@ -214,7 +214,7 @@ void app_main(void)
#endif
/* register wifi connect commands */
register_wifi_connect_commands();
example_register_wifi_connect_commands();
/* register command `ping` */
register_ping();
/* register command `quit` */

14
examples/protocols/icmp_echo/pytest_icmp_echo.py
View File

@ -4,7 +4,7 @@
import os
import pytest
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
@ -12,13 +12,15 @@ from pytest_embedded import Dut
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_nearby
@pytest.mark.wifi_ap
def test_protocols_icmp_echo(dut: Dut) -> None:
# get env config
env_config = get_env_config('wifi_nearby')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
ap_channel = env_config.get('ap_channel', 0)
sdkconfig_ssid = dut.app.sdkconfig.get('CONFIG_EXAMPLE_WIFI_SSID')
sdkconfig_pwd = dut.app.sdkconfig.get('CONFIG_EXAMPLE_WIFI_SSID')
env_name = 'wifi_ap'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid', default=sdkconfig_ssid)
ap_password = get_env_config_variable(env_name, 'ap_password', default=sdkconfig_pwd)
ap_channel = get_env_config_variable(env_name, 'ap_channel', default=0)
dut.expect('esp>')
dut.write(f'wifi_connect {ap_ssid} {ap_password} -n {ap_channel}')

8
examples/protocols/modbus/tcp/example_test.py
View File

@ -29,8 +29,8 @@ STACK_PAR_OK = 6
STACK_PAR_FAIL = 7
STACK_DESTROY = 8
pattern_dict_slave = {STACK_IPV4: (r'.*I \([0-9]+\) example_connect: - IPv4 address: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}).*'),
STACK_IPV6: (r'.*I \([0-9]+\) example_connect: - IPv6 address: (([A-Fa-f0-9]{1,4}::?){1,7}[A-Fa-f0-9]{1,4}).*'),
pattern_dict_slave = {STACK_IPV4: (r'.*I \([0-9]+\) example_common: - IPv4 address: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}).*'),
STACK_IPV6: (r'.*I \([0-9]+\) example_common: - IPv6 address: (([A-Fa-f0-9]{1,4}::?){1,7}[A-Fa-f0-9]{1,4}).*'),
STACK_INIT: (r'.*I \(([0-9]+)\) MB_TCP_SLAVE_PORT: (Protocol stack initialized).'),
STACK_CONNECT: (r'.*I\s\(([0-9]+)\) MB_TCP_SLAVE_PORT: Socket \(#[0-9]+\), accept client connection from address: '
r'([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}).*'),
@ -40,8 +40,8 @@ pattern_dict_slave = {STACK_IPV4: (r'.*I \([0-9]+\) example_connect: - IPv4 addr
STACK_PAR_FAIL: (r'.*E \(([0-9]+)\) SLAVE_TEST: Response time exceeds configured [0-9]+ [ms], ignore packet.*'),
STACK_DESTROY: (r'.*I\s\(([0-9]+)\) SLAVE_TEST: (Modbus controller destroyed).')}
pattern_dict_master = {STACK_IPV4: (r'.*I \([0-9]+\) example_connect: - IPv4 address: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}).*'),
STACK_IPV6: (r'.*I \([0-9]+\) example_connect: - IPv6 address: (([A-Fa-f0-9]{1,4}::?){1,7}[A-Fa-f0-9]{1,4}).*'),
pattern_dict_master = {STACK_IPV4: (r'.*I \([0-9]+\) example_common: - IPv4 address: ([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}).*'),
STACK_IPV6: (r'.*I \([0-9]+\) example_common: - IPv6 address: (([A-Fa-f0-9]{1,4}::?){1,7}[A-Fa-f0-9]{1,4}).*'),
STACK_INIT: (r'.*I \(([0-9]+)\) MASTER_TEST: (Modbus master stack initialized).*'),
STACK_CONNECT: (r'.*.*I\s\(([0-9]+)\) MB_TCP_MASTER_PORT: (Connected [0-9]+ slaves), start polling.*'),
STACK_START: (r'.*I \(([0-9]+)\) MASTER_TEST: (Start modbus test).*'),

4
examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py
View File

@ -58,7 +58,7 @@ def on_message(client, userdata, msg):
message_log += 'Received data:' + msg.topic + ' ' + payload + '\n'
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1')
@ttfw_idf.idf_example_test(env_tag='ethernet_router')
def test_examples_protocol_mqtt_ssl(env, extra_data):
broker_url = ''
broker_port = 0
@ -110,7 +110,7 @@ def test_examples_protocol_mqtt_ssl(env, extra_data):
raise ValueError('ENV_TEST_FAILURE: Test script cannot connect to broker: {}'.format(broker_url))
dut1.start_app()
try:
ip_address = dut1.expect(re.compile(r'IPv4 address: ([^,]+),'), timeout=30)
ip_address = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
print('ENV_TEST_FAILURE: Cannot connect to AP')

8
examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py
View File

@ -7,7 +7,7 @@ import time
from threading import Thread
import ttfw_idf
from common_test_methods import get_my_ip4_by_dest_ip
from common_test_methods import get_host_ip4_by_dest_ip
from tiny_test_fw import DUT
msgid = -1
@ -46,7 +46,7 @@ def mqqt_server_sketch(my_ip, port):
print('server closed')
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1')
@ttfw_idf.idf_example_test(env_tag='ethernet_router')
def test_examples_protocol_mqtt_qos1(env, extra_data):
global msgid
"""
@ -65,13 +65,13 @@ def test_examples_protocol_mqtt_qos1(env, extra_data):
dut1.start_app()
# waiting for getting the IP address
try:
ip_address = dut1.expect(re.compile(r'IPv4 address: ([^,]+),'), timeout=30)[0]
ip_address = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# 2. start mqtt broker sketch
host_ip = get_my_ip4_by_dest_ip(ip_address)
host_ip = get_host_ip4_by_dest_ip(ip_address)
thread1 = Thread(target=mqqt_server_sketch, args=(host_ip,1883))
thread1.start()

4
examples/protocols/mqtt/ws/mqtt_ws_example_test.py
View File

@ -39,7 +39,7 @@ def on_message(client, userdata, msg):
message_log += 'Received data:' + msg.topic + ' ' + payload + '\n'
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1')
@ttfw_idf.idf_example_test(env_tag='ethernet_router')
def test_examples_protocol_mqtt_ws(env, extra_data):
broker_url = ''
broker_port = 0
@ -83,7 +83,7 @@ def test_examples_protocol_mqtt_ws(env, extra_data):
raise ValueError('ENV_TEST_FAILURE: Test script cannot connect to broker: {}'.format(broker_url))
dut1.start_app()
try:
ip_address = dut1.expect(re.compile(r'IPv4 address: ([^,]+),'), timeout=30)
ip_address = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
print('ENV_TEST_FAILURE: Cannot connect to AP')

4
examples/protocols/mqtt/wss/mqtt_wss_example_test.py
View File

@ -40,7 +40,7 @@ def on_message(client, userdata, msg):
message_log += 'Received data:' + msg.topic + ' ' + payload + '\n'
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1')
@ttfw_idf.idf_example_test(env_tag='ethernet_router')
def test_examples_protocol_mqtt_wss(env, extra_data):
broker_url = ''
broker_port = 0
@ -87,7 +87,7 @@ def test_examples_protocol_mqtt_wss(env, extra_data):
raise ValueError('ENV_TEST_FAILURE: Test script cannot connect to broker: {}'.format(broker_url))
dut1.start_app()
try:
ip_address = dut1.expect(re.compile(r'IPv4 address: ([^,]+),'), timeout=30)
ip_address = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
print('ENV_TEST_FAILURE: Cannot connect to AP')

12
examples/protocols/sntp/pytest_sntp.py
View File

@ -6,20 +6,20 @@ import logging
from typing import Any, Tuple
import pytest
from common_test_methods import get_env_config
from common_test_methods import get_env_config_variable
from pytest_embedded import Dut
@pytest.mark.esp32
@pytest.mark.wifi_nearby
@pytest.mark.wifi_ap
def test_get_time_from_sntp_server(dut: Dut) -> None:
dut.expect('Time is not set yet. Connecting to WiFi and getting time over NTP.')
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_config = get_env_config('wifi_nearby')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut.expect('Please input ssid password:')
dut.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_ap'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
dut.expect('IPv4 address:')
dut.expect('Initializing SNTP')

16
examples/protocols/sockets/tcp_client/example_test.py
View File

@ -17,7 +17,7 @@ from threading import Event, Thread
import netifaces
import ttfw_idf
from common_test_methods import get_env_config, get_my_interface_by_dest_ip, get_my_ip_by_interface
from common_test_methods import get_env_config_variable, get_host_ip_by_interface, get_my_interface_by_dest_ip
# ----------- Config ----------
PORT = 3333
@ -97,13 +97,13 @@ def test_examples_protocol_socket_tcpclient(env, extra_data):
# start test
dut1.start_app()
if dut1.app.get_sdkconfig_config_value('CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN'):
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut1.expect('Please input ssid password:')
dut1.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut1.write(f'{ap_ssid} {ap_password}')
ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
ipv4 = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8) # expect all 8 octets from IPv6 (assumes it's printed in the long form)
ipv6 = dut1.expect(re.compile(r' IPv6 address: ({})'.format(ipv6_r)), timeout=30)[0]
print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))
@ -111,13 +111,13 @@ def test_examples_protocol_socket_tcpclient(env, extra_data):
my_interface = get_my_interface_by_dest_ip(ipv4)
# test IPv4
with TcpServer(PORT, socket.AF_INET):
server_ip = get_my_ip_by_interface(my_interface, netifaces.AF_INET)
server_ip = get_host_ip_by_interface(my_interface, netifaces.AF_INET)
print('Connect tcp client to server IP={}'.format(server_ip))
dut1.write(server_ip)
dut1.expect(re.compile(r'OK: Message from ESP32'))
# test IPv6
with TcpServer(PORT, socket.AF_INET6):
server_ip = get_my_ip_by_interface(my_interface, netifaces.AF_INET6)
server_ip = get_host_ip_by_interface(my_interface, netifaces.AF_INET6)
print('Connect tcp client to server IP={}'.format(server_ip))
dut1.write(server_ip)
dut1.expect(re.compile(r'OK: Message from ESP32'))

12
examples/protocols/sockets/tcp_server/example_test.py
View File

@ -14,7 +14,7 @@ import socket
import sys
import ttfw_idf
from common_test_methods import get_env_config, get_my_interface_by_dest_ip
from common_test_methods import get_env_config_variable, get_my_interface_by_dest_ip
# ----------- Config ----------
PORT = 3333
@ -64,13 +64,13 @@ def test_examples_protocol_socket_tcpserver(env, extra_data):
# start test
dut1.start_app()
if dut1.app.get_sdkconfig_config_value('CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN'):
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut1.expect('Please input ssid password:')
dut1.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut1.write(f'{ap_ssid} {ap_password}')
ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
ipv4 = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8) # expect all 8 octets from IPv6 (assumes it's printed in the long form)
ipv6 = dut1.expect(re.compile(r' IPv6 address: ({})'.format(ipv6_r)), timeout=30)[0]
print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))

40
examples/protocols/sockets/udp_client/example_test.py
View File

@ -17,7 +17,8 @@ from threading import Event, Thread
import netifaces
import ttfw_idf
from common_test_methods import get_env_config, get_my_interface_by_dest_ip, get_my_ip_by_interface
from common_test_methods import get_env_config_variable, get_host_ip_by_interface, get_my_interface_by_dest_ip
from tiny_test_fw.DUT import ExpectTimeout
# ----------- Config ----------
PORT = 3333
@ -60,6 +61,7 @@ class UdpServer:
while not self.shutdown.is_set():
try:
data, addr = self.socket.recvfrom(1024)
print(addr)
if not data:
return
data = data.decode()
@ -90,13 +92,13 @@ def test_examples_protocol_socket_udpclient(env, extra_data):
# start test
dut1.start_app()
if dut1.app.get_sdkconfig_config_value('CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN'):
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut1.expect('Please input ssid password:')
dut1.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut1.write(f'{ap_ssid} {ap_password}')
ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
ipv4 = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8) # expect all 8 octets from IPv6 (assumes it's printed in the long form)
ipv6 = dut1.expect(re.compile(r' IPv6 address: ({})'.format(ipv6_r)), timeout=30)[0]
print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))
@ -104,16 +106,30 @@ def test_examples_protocol_socket_udpclient(env, extra_data):
my_interface = get_my_interface_by_dest_ip(ipv4)
# test IPv4
with UdpServer(PORT, socket.AF_INET):
server_ip = get_my_ip_by_interface(my_interface, netifaces.AF_INET)
server_ip = get_host_ip_by_interface(my_interface, netifaces.AF_INET)
print('Connect udp client to server IP={}'.format(server_ip))
dut1.write(server_ip)
dut1.expect(re.compile(r'OK: Message from ESP32'))
for _ in range(3):
try:
dut1.write(server_ip)
dut1.expect(re.compile(r'OK: Message from ESP32'))
break
except ExpectTimeout:
pass
else:
raise ValueError('Failed to send/recv udp packets.')
# test IPv6
with UdpServer(PORT, socket.AF_INET6):
server_ip = get_my_ip_by_interface(my_interface, netifaces.AF_INET6)
server_ip = get_host_ip_by_interface(my_interface, netifaces.AF_INET6)
print('Connect udp client to server IP={}'.format(server_ip))
dut1.write(server_ip)
dut1.expect(re.compile(r'OK: Message from ESP32'))
for _ in range(3):
try:
dut1.write(server_ip)
dut1.expect(re.compile(r'OK: Message from ESP32'))
break
except ExpectTimeout:
pass
else:
raise ValueError('Failed to send/recv udp packets.')
if __name__ == '__main__':

7
examples/protocols/sockets/udp_client/main/udp_client.c
View File

@ -73,6 +73,13 @@ static void udp_client_task(void *pvParameters)
ESP_LOGE(TAG, "Unable to create socket: errno %d", errno);
break;
}
// Set timeout
struct timeval timeout;
timeout.tv_sec = 10;
timeout.tv_usec = 0;
setsockopt (sock, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout);
ESP_LOGI(TAG, "Socket created, sending to %s:%d", HOST_IP_ADDR, PORT);
while (1) {

12
examples/protocols/sockets/udp_server/example_test.py
View File

@ -14,7 +14,7 @@ import socket
import sys
import ttfw_idf
from common_test_methods import get_env_config, get_my_interface_by_dest_ip
from common_test_methods import get_env_config_variable, get_my_interface_by_dest_ip
# ----------- Config ----------
PORT = 3333
@ -68,13 +68,13 @@ def test_examples_protocol_socket_udpserver(env, extra_data):
# start test
dut1.start_app()
if dut1.app.get_sdkconfig_config_value('CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN'):
env_config = get_env_config('wifi_router')
ap_ssid = env_config['ap_ssid']
ap_password = env_config['ap_password']
dut1.expect('Please input ssid password:')
dut1.write(' '.join([ap_ssid, ap_password]))
env_name = 'wifi_router'
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut1.write(f'{ap_ssid} {ap_password}')
ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
ipv4 = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8) # expect all 8 octets from IPv6 (assumes it's printed in the long form)
ipv6 = dut1.expect(re.compile(r' IPv6 address: ({})'.format(ipv6_r)), timeout=30)[0]
print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))

5
examples/protocols/sockets/udp_server/main/udp_server.c
View File

@ -71,6 +71,11 @@ static void udp_server_task(void *pvParameters)
setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, &opt, sizeof(opt));
}
#endif
// Set timeout
struct timeval timeout;
timeout.tv_sec = 10;
timeout.tv_usec = 0;
setsockopt (sock, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout);
int err = bind(sock, (struct sockaddr *)&dest_addr, sizeof(dest_addr));
if (err < 0) {

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

@ -86,12 +86,14 @@ cp ca_cert.pem /path/to/ota/example/server_certs/
### Internal workflow of the OTA Example
After booting, the firmware prints "Starting OTA example" to the console and:
After booting, the firmware:
1. Connects via Ethernet or to the AP using the provided SSID and password (Wi-Fi case)
2. Connects to the HTTPS server and downloads the new image
3. Writes the image to flash, and instructs the bootloader to boot from this image after the next reset
4. Reboots
1. prints "OTA example app_main start" to the console
2. Connects via Ethernet or to the AP using the provided SSID and password (Wi-Fi case)
3. prints "Starting OTA example task" to the console
4. Connects to the HTTPS server and downloads the new image
5. Writes the image to flash, and instructs the bootloader to boot from this image after the next reset
6. Reboots
If you want to rollback to the `factory` app after the upgrade (or to the first OTA partition in case the `factory` partition does not exist), run the command `idf.py erase_otadata`. This restores the `ota_data` partition to its initial state.

1
examples/system/ota/advanced_https_ota/main/advanced_https_ota_example.c
View File

@ -176,6 +176,7 @@ ota_end:
void app_main(void)
{
ESP_LOGI(TAG, "OTA example app_main start");
// Initialize NVS.
esp_err_t err = nvs_flash_init();
if (err == ESP_ERR_NVS_NO_FREE_PAGES || err == ESP_ERR_NVS_NEW_VERSION_FOUND) {

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

@ -1,6 +1,5 @@
# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Unlicense OR CC0-1.0
from ast import Try
import http.server
import multiprocessing
import os
@ -9,11 +8,12 @@ import socket
import ssl
import struct
import subprocess
import time
from typing import Callable
import pexpect
import pytest
from common_test_methods import get_my_ip4_by_dest_ip
from common_test_methods import get_env_config_variable, get_host_ip4_by_dest_ip
from pytest_embedded import Dut
from RangeHTTPServer import RangeRequestHandler
@ -97,7 +97,7 @@ def start_redirect_server(ota_image_dir: str, server_ip: str, server_port: int,
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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.
@ -111,32 +111,34 @@ def test_examples_protocol_advanced_https_ota_example(dut: Dut) -> None:
iterations = 3
server_port = 8001
bin_name = 'advanced_https_ota.bin'
# start test
for _ in range(iterations):
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
dut.expect('Starting Advanced OTA example', timeout=30)
# Start server
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
for _ in range(iterations):
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
dut.expect('Starting Advanced OTA example', timeout=30)
host_ip = get_host_ip4_by_dest_ip(ip_address)
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1.daemon = True
thread1.start()
try:
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
finally:
thread1.terminate()
dut.expect('upgrade successful. Rebooting ...', timeout=60)
finally:
thread1.terminate()
@pytest.mark.esp32
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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.
@ -159,24 +161,22 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(dut: Dut) ->
with open(binary_file, 'rb+') as f:
with open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+') as fo:
fo.write(f.read(truncated_bin_size))
binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
dut.expect('Starting Advanced OTA example', timeout=30)
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
dut.expect('Starting Advanced OTA example', timeout=30)
host_ip = get_host_ip4_by_dest_ip(ip_address)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut.expect('Image validation failed, image is corrupted', timeout=30)
@ -192,7 +192,7 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(dut: Dut) ->
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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.
@ -215,22 +215,21 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(dut: Dut)
with open(binary_file, 'rb+') as f:
with open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+') as fo:
fo.write(f.read(truncated_bin_size))
binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
@ -247,7 +246,7 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(dut: Dut)
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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.
@ -271,21 +270,20 @@ def test_examples_protocol_advanced_https_ota_example_random(dut: Dut) -> None:
fo.write(struct.pack('B', 0))
for i in range(random_bin_size - 1):
fo.write(struct.pack('B', random.randrange(0,255,1)))
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
@ -302,7 +300,7 @@ def test_examples_protocol_advanced_https_ota_example_random(dut: Dut) -> None:
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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.
@ -328,21 +326,20 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(dut: Dut)
data[13] = 0xfe
with open(random_binary_file, 'wb+') as fo:
fo.write(bytearray(data))
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
@ -359,7 +356,7 @@ def test_examples_protocol_advanced_https_ota_example_invalid_chip_id(dut: Dut)
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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.
@ -371,23 +368,25 @@ def test_examples_protocol_advanced_https_ota_example_chunked(dut: Dut) -> None:
"""
# File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin'
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
chunked_server = start_chunked_server(dut.app.binary_path, 8070)
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
dut.write('https://' + host_ip + ':8070/' + bin_name)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect('Starting Advanced OTA example', timeout=30)
dut.expect('upgrade successful. Rebooting ...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
chunked_server.kill()
@ -396,7 +395,7 @@ def test_examples_protocol_advanced_https_ota_example_chunked(dut: Dut) -> None:
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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.
@ -416,29 +415,32 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(dut: Dut) ->
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
dut.expect('Starting Advanced OTA example', timeout=30)
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
host_ip = get_host_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1.daemon = True
thread2 = multiprocessing.Process(target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port, redirection_server_port1))
thread2.daemon = True
thread3 = multiprocessing.Process(target=start_redirect_server, args=(dut.app.binary_path, host_ip, redirection_server_port1, server_port))
thread3.daemon = True
thread1.start()
thread2.start()
thread3.start()
time.sleep(1)
try:
thread1.start()
thread2.start()
thread3.start()
print('writing to device: {}'.format('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name))
dut.write('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect('Starting Advanced OTA example', timeout=30)
dut.expect('upgrade successful. Rebooting ...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
thread1.terminate()
thread2.terminate()
@ -479,27 +481,27 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(dut: Dut) ->
fo.seek(36)
fo.write(b'\x00')
binary_file = os.path.join(dut.app.binary_path, anti_rollback_bin_name)
# start test
# Positive Case
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
# Positive Case
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
# Use originally generated image with secure_version=1
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
dut.expect(r'App is valid, rollback cancelled successfully', timeout=30)
# Negative Case
@ -520,7 +522,7 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(dut: Dut) ->
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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:
"""
@ -532,35 +534,37 @@ def test_examples_protocol_advanced_https_ota_example_partial_request(dut: Dut)
"""
server_port = 8001
# Size of partial HTTP request
request_size = 16384
request_size = int(dut.app.sdkconfig.get('EXAMPLE_HTTP_REQUEST_SIZE'))
# File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin'
binary_file = os.path.join(dut.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file)
http_requests = int((bin_size / request_size) - 1)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
print('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
raise
assert http_requests > 1
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
for _ in range(http_requests):
dut.expect('Connection closed', timeout=60)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect('Starting Advanced OTA example', timeout=30)
except:
dut.expect('upgrade successful. Rebooting ...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset', timeout=30)
dut.expect('OTA example app_main start', timeout=20)
finally:
thread1.terminate()
@ -568,7 +572,7 @@ def test_examples_protocol_advanced_https_ota_example_partial_request(dut: Dut)
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_ota
@pytest.mark.wifi_high_traffic
@pytest.mark.nightly_run
@pytest.mark.parametrize('config', ['nimble',], indirect=True)
def test_examples_protocol_advanced_https_ota_example_nimble_gatts(dut: Dut) -> None:
@ -583,29 +587,37 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(dut: Dut) ->
server_port = 8001
# File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin'
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
# Parse IP address of STA
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_name = 'wifi_high_traffic'
dut.expect('Please input ssid password:')
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
print('Started GAP advertising.')
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect('Starting Advanced OTA example', timeout=30)
except:
dut.expect('upgrade successful. Rebooting ...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
thread1.terminate()
@ -613,7 +625,7 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(dut: Dut) ->
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_ota
@pytest.mark.wifi_high_traffic
@pytest.mark.nightly_run
@pytest.mark.parametrize('config', ['bluedroid',], indirect=True)
def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(dut: Dut) -> None:
@ -628,29 +640,37 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(dut: Dut)
server_port = 8001
# File to be downloaded. This file is generated after compilation
bin_name = 'advanced_https_ota.bin'
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
# Parse IP address of STA
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_name = 'wifi_high_traffic'
dut.expect('Please input ssid password:')
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut.expect('Started advertising.', timeout=30)
print('Started GAP advertising.')
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect('Starting Advanced OTA example', timeout=30)
dut.expect('upgrade successful. Rebooting ...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
thread1.terminate()
@ -659,7 +679,7 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(dut: Dut)
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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
@ -683,23 +703,25 @@ def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(dut: D
fo.write(f.read(bin_size))
for _ in range(dummy_data_size):
fo.write(struct.pack('B', random.randrange(0,255,1)))
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
chunked_server = start_chunked_server(dut.app.binary_path, 8070)
try:
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Advanced OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8070/' + aligned_bin_name))
dut.write('https://' + host_ip + ':8070/' + aligned_bin_name)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect('Starting Advanced OTA example', timeout=30)
dut.expect('upgrade successful. Rebooting ...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
try:
os.remove(aligned_bin_name)
except OSError:

1
examples/system/ota/advanced_https_ota/sdkconfig.ci.nimble
View File

@ -11,6 +11,7 @@ CONFIG_PARTITION_TABLE_CUSTOM=y
CONFIG_PARTITION_TABLE_CUSTOM_FILENAME="partitions_example_with_ble.csv"
CONFIG_PARTITION_TABLE_FILENAME="partitions_example_with_ble.csv"
CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y
CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN"
CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y
CONFIG_EXAMPLE_SKIP_VERSION_CHECK=y

4
examples/system/ota/native_ota_example/main/native_ota_example.c
View File

@ -80,7 +80,7 @@ static void ota_example_task(void *pvParameter)
esp_ota_handle_t update_handle = 0 ;
const esp_partition_t *update_partition = NULL;
ESP_LOGI(TAG, "Starting OTA example");
ESP_LOGI(TAG, "Starting OTA example task");
const esp_partition_t *configured = esp_ota_get_boot_partition();
const esp_partition_t *running = esp_ota_get_running_partition();
@ -273,6 +273,8 @@ static bool diagnostic(void)
void app_main(void)
{
ESP_LOGI(TAG, "OTA example app_main start");
uint8_t sha_256[HASH_LEN] = { 0 };
esp_partition_t partition;

163
examples/system/ota/native_ota_example/pytest_native_ota.py
View File

@ -12,7 +12,7 @@ from typing import Callable, Tuple
import pexpect
import pytest
from common_test_methods import get_my_ip4_by_dest_ip
from common_test_methods import get_host_ip4_by_dest_ip
from pytest_embedded import Dut
server_cert = '-----BEGIN CERTIFICATE-----\n' \
@ -122,7 +122,7 @@ def start_chunked_server(ota_image_dir: str, server_port: int) -> subprocess.Pop
@pytest.mark.supported_targets
@pytest.mark.ethernet_router
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example(dut: Dut) -> None:
"""
This is a positive test case, which downloads complete binary file multiple number of times.
@ -137,29 +137,31 @@ def test_examples_protocol_native_ota_example(dut: Dut) -> None:
iterations = 3
# File to be downloaded. This file is generated after compilation
bin_name = 'native_ota.bin'
# start test
for _ in range(iterations):
dut.expect('Loaded app from partition at offset', timeout=60)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
# Start server
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
# start test
for _ in range(iterations):
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
finally:
thread1.terminate()
dut.expect('Prepare to restart system!', timeout=60)
finally:
thread1.terminate()
@pytest.mark.supported_targets
@pytest.mark.ethernet_router
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example_truncated_bin(dut: Dut) -> None:
"""
Working of OTA if binary file is truncated is validated in this test case.
@ -180,26 +182,26 @@ def test_examples_protocol_native_ota_example_truncated_bin(dut: Dut) -> None:
truncated_bin_size = 64000
# check and log bin size
binary_file = os.path.join(dut.app.binary_path, bin_name)
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+')
fo.write(f.read(truncated_bin_size))
fo.close()
f.close()
with open(binary_file, 'rb+') as fr:
bin_data = fr.read(truncated_bin_size)
binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
with open(binary_file, 'wb+') as fo:
fo.write(bin_data)
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut.expect('native_ota_example: Image validation failed, image is corrupted', timeout=20)
@ -209,7 +211,7 @@ def test_examples_protocol_native_ota_example_truncated_bin(dut: Dut) -> None:
@pytest.mark.supported_targets
@pytest.mark.ethernet_router
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example_truncated_header(dut: Dut) -> None:
"""
Working of OTA if headers of binary file are truncated is vaildated in this test case.
@ -229,26 +231,26 @@ def test_examples_protocol_native_ota_example_truncated_header(dut: Dut) -> None
truncated_bin_size = 180
# check and log bin size
binary_file = os.path.join(dut.app.binary_path, bin_name)
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut.app.binary_path, truncated_bin_name), 'wb+')
fo.write(f.read(truncated_bin_size))
fo.close()
f.close()
with open(binary_file, 'rb+') as fr:
bin_data = fr.read(truncated_bin_size)
binary_file = os.path.join(dut.app.binary_path, truncated_bin_name)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
with open(binary_file, 'wb+') as fo:
fo.write(bin_data)
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
dut.expect('native_ota_example: received package is not fit len', timeout=20)
@ -258,7 +260,7 @@ def test_examples_protocol_native_ota_example_truncated_header(dut: Dut) -> None
@pytest.mark.supported_targets
@pytest.mark.ethernet_router
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example_random(dut: Dut) -> None:
"""
Working of OTA if random data is added in binary file are validated in this test case.
@ -279,25 +281,25 @@ def test_examples_protocol_native_ota_example_random(dut: Dut) -> None:
fo = open(binary_file, 'wb+')
# First byte of binary file is always set to zero. If first byte is generated randomly,
# in some cases it may generate 0xE9 which will result in failure of testcase.
fo.write(struct.pack('B', 0))
for i in range(random_bin_size - 1):
fo.write(struct.pack('B', random.randrange(0,255,1)))
fo.close()
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
with open(binary_file, 'wb+') as fo:
fo.write(struct.pack('B', 0))
for _ in range(random_bin_size - 1):
fo.write(struct.pack('B', random.randrange(0,255,1)))
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
dut.expect('esp_ota_ops: OTA image has invalid magic byte', timeout=20)
@ -307,7 +309,7 @@ def test_examples_protocol_native_ota_example_random(dut: Dut) -> None:
@pytest.mark.supported_targets
@pytest.mark.ethernet_router
@pytest.mark.ethernet_ota
def test_examples_protocol_native_ota_example_chunked(dut: Dut) -> None:
"""
This is a positive test case, which downloads complete binary file multiple number of times.
@ -319,22 +321,25 @@ def test_examples_protocol_native_ota_example_chunked(dut: Dut) -> None:
"""
# File to be downloaded. This file is generated after compilation
bin_name = 'native_ota.bin'
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
chunked_server = start_chunked_server(dut.app.binary_path, 8070)
try:
dut.expect('Starting OTA example', timeout=30)
# start test
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
dut.write('https://' + host_ip + ':8070/' + bin_name)
dut.expect('Loaded app from partition at offset', timeout=60)
dut.expect('Starting OTA example', timeout=30)
dut.expect('Prepare to restart system!', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
os.remove(os.path.join(dut.app.binary_path, 'server_cert.pem'))
os.remove(os.path.join(dut.app.binary_path, 'server_key.pem'))
finally:

25
examples/system/ota/pre_encrypted_ota/pytest_pre_encrypted_ota.py
View File

@ -9,7 +9,7 @@ from typing import Callable
import pexpect
import pytest
from common_test_methods import get_my_ip4_by_dest_ip
from common_test_methods import get_host_ip4_by_dest_ip
from pytest_embedded import Dut
from RangeHTTPServer import RangeRequestHandler
@ -56,28 +56,29 @@ def start_https_server(ota_image_dir: str, server_ip: str, server_port: int) ->
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@pytest.mark.ethernet_ota
def test_examples_protocol_pre_encrypted_ota_example(dut: Dut) -> None:
server_port = 8001
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, server_port))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
thread1.daemon = True
thread1.start()
try:
dut.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting Pre Encrypted OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + enc_bin_name))
dut.write('https://' + host_ip + ':' + str(server_port) + '/' + enc_bin_name)
dut.expect('Magic Verified', timeout=30)
dut.expect('Reading RSA private key', timeout=30)
dut.expect('upgrade successful. Rebooting', timeout=30)
# after reboot
dut.expect('Loaded app from partition at offset', timeout=30)
finally:
thread1.terminate()

4
examples/system/ota/simple_ota_example/main/simple_ota_example.c
View File

@ -78,7 +78,7 @@ esp_err_t _http_event_handler(esp_http_client_event_t *evt)
void simple_ota_example_task(void *pvParameter)
{
ESP_LOGI(TAG, "Starting OTA example");
ESP_LOGI(TAG, "Starting OTA example task");
#ifdef CONFIG_EXAMPLE_FIRMWARE_UPGRADE_BIND_IF
esp_netif_t *netif = get_example_netif_from_desc(bind_interface_name);
if (netif == NULL) {
@ -127,6 +127,7 @@ void simple_ota_example_task(void *pvParameter)
ESP_LOGI(TAG, "Attempting to download update from %s", config.url);
esp_err_t ret = esp_https_ota(&ota_config);
if (ret == ESP_OK) {
ESP_LOGI(TAG, "OTA Succeed, Rebooting...");
esp_restart();
} else {
ESP_LOGE(TAG, "Firmware upgrade failed");
@ -165,6 +166,7 @@ static void get_sha256_of_partitions(void)
void app_main(void)
{
ESP_LOGI(TAG, "OTA example app_main start");
// Initialize NVS.
esp_err_t err = nvs_flash_init();
if (err == ESP_ERR_NVS_NO_FREE_PAGES || err == ESP_ERR_NVS_NEW_VERSION_FOUND) {

221
examples/system/ota/simple_ota_example/pytest_simple_ota.py
View File

@ -3,14 +3,13 @@
import http.server
import multiprocessing
import os
import socket
import ssl
import sys
from typing import Tuple
import pexpect
import pytest
from common_test_methods import get_my_ip4_by_dest_ip
from common_test_methods import get_env_config_variable, get_host_ip4_by_dest_ip
from pytest_embedded import Dut
server_cert = '-----BEGIN CERTIFICATE-----\n' \
@ -110,8 +109,7 @@ def calc_all_sha256(dut: Dut) -> Tuple[str, str]:
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.wifi_ota
@pytest.mark.nightly_run
@pytest.mark.wifi_high_traffic
def test_examples_protocol_simple_ota_example(dut: Dut) -> None:
"""
steps: |
@ -120,27 +118,36 @@ def test_examples_protocol_simple_ota_example(dut: Dut) -> None:
3. Reboot with the new OTA image
"""
sha256_bootloader, sha256_app = calc_all_sha256(dut)
# start test
dut.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_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', 8000))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
# start test
dut.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_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
# Parse IP address of STA
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_name = 'wifi_high_traffic'
dut.expect('Please input ssid password:')
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut.expect('Loaded app from partition at offset 0x110000', timeout=60)
dut.expect('Starting OTA example', timeout=30)
dut.expect('OTA Succeed, Rebooting...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset 0x110000', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
thread1.terminate()
@ -149,7 +156,7 @@ def test_examples_protocol_simple_ota_example(dut: Dut) -> None:
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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:
"""
@ -158,73 +165,34 @@ def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(dut: D
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
# start test
dut.expect('Loaded app from partition at offset 0x10000', timeout=30))
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', 8000))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
# start test
dut.expect('Loaded app from partition at offset 0x10000', timeout=30)
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut.expect('Loaded app from partition at offset 0x110000', timeout=60)
dut.expect('Starting OTA example', timeout=30)
dut.expect('OTA Succeed, Rebooting...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset 0x110000', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
thread1.terminate()
@pytest.mark.esp32
@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: |
1. join AP/Ethernet
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
# Erase flash
dut.serial.erase_flash()
dut.serial.flash()
# start test
dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut.expect('Loaded app from partition at offset 0x120000', timeout=60)
dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
dut.expect('Starting OTA example', timeout=30)
finally:
thread1.terminate()
@pytest.mark.esp32c3
@pytest.mark.flash_encryption_wifi_ota
@pytest.mark.flash_encryption_wifi_high_traffic
@pytest.mark.nightly_run
@pytest.mark.parametrize('config', ['flash_enc_wifi',], indirect=True)
@pytest.mark.parametrize('skip_autoflash', ['y'], indirect=True)
@ -239,26 +207,35 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(dut: Du
# Erase flash
dut.serial.erase_flash()
dut.serial.flash()
dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', 8000))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
dut.expect('Loaded app from partition at offset 0x20000', timeout=30)
dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
# Parse IP address of STA
if dut.app.sdkconfig.get('EXAMPLE_WIFI_SSID_PWD_FROM_STDIN') is True:
env_name = 'flash_encryption_wifi_high_traffic'
dut.expect('Please input ssid password:')
ap_ssid = get_env_config_variable(env_name, 'ap_ssid')
ap_password = get_env_config_variable(env_name, 'ap_password')
dut.write(f'{ap_ssid} {ap_password}')
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut.expect('Loaded app from partition at offset 0x120000', timeout=60)
dut.expect('OTA Succeed, Rebooting...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset 0x120000', timeout=30)
dut.expect('Flash encryption mode is DEVELOPMENT', timeout=10)
dut.expect('Starting OTA example', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
thread1.terminate()
@ -267,7 +244,7 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(dut: Du
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@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:
"""
@ -277,30 +254,31 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat
3. Reboot with the new OTA image
"""
sha256_bootloader, sha256_app = calc_all_sha256(dut)
# start test
dut.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_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', 8000))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
# start test
dut.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_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut.expect('Writing to partition subtype 16 at offset 0x120000', timeout=20)
dut.expect('Verifying image signature...', timeout=60)
dut.expect('OTA Succeed, Rebooting...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset 0x120000', timeout=20)
dut.expect('Starting OTA example', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
thread1.terminate()
@ -309,7 +287,7 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat
@pytest.mark.esp32c3
@pytest.mark.esp32s2
@pytest.mark.esp32s3
@pytest.mark.ethernet_router
@pytest.mark.ethernet_ota
@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:
"""
@ -319,35 +297,34 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat
3. Reboot with the new OTA image
"""
sha256_bootloader, sha256_app = calc_all_sha256(dut)
# start test
dut.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_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
try:
ip_address = dut.expect(r'IPv4 address: ([^,]+),', timeout=30)
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
thread1.terminate()
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
# Start server
host_ip = get_my_ip4_by_dest_ip(ip_address)
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, host_ip, 8000))
thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', 8000))
thread1.daemon = True
thread1.start()
try:
dut.expect('Starting OTA example', timeout=30)
# start test
dut.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_app, str(dut.expect(r'SHA-256 for current firmware:\s+([a-f0-9]){64}')[0]))
try:
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode()
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
except pexpect.exceptions.TIMEOUT:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
host_ip = get_host_ip4_by_dest_ip(ip_address)
dut.expect('Starting OTA example task', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
dut.write('https://' + host_ip + ':8000/simple_ota.bin')
dut.expect('Writing to partition subtype 16 at offset 0x120000', timeout=20)
dut.expect('Verifying image signature...', timeout=60)
dut.expect('#0 app key digest == #0 trusted key digest', timeout=10)
dut.expect('Verifying with RSA-PSS...', timeout=10)
dut.expect('Signature verified successfully!', timeout=10)
dut.expect('OTA Succeed, Rebooting...', timeout=60)
# after reboot
dut.expect('Loaded app from partition at offset 0x120000', timeout=20)
dut.expect('Starting OTA example', timeout=30)
dut.expect('OTA example app_main start', timeout=10)
finally:
thread1.terminate()

1
examples/system/ota/simple_ota_example/sdkconfig.ci
View File

@ -1,3 +1,4 @@
CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y
CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN"
CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y
CONFIG_EXAMPLE_FIRMWARE_UPGRADE_BIND_IF=y

25
examples/system/ota/simple_ota_example/sdkconfig.ci.flash_enc
View File

@ -1,25 +0,0 @@
CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN"
CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y
CONFIG_SECURE_FLASH_ENC_ENABLED=y
CONFIG_SECURE_FLASH_ENCRYPTION_MODE_DEVELOPMENT=y
CONFIG_SECURE_BOOT_ALLOW_ROM_BASIC=y
CONFIG_SECURE_BOOT_ALLOW_JTAG=y
CONFIG_SECURE_FLASH_UART_BOOTLOADER_ALLOW_ENC=y
CONFIG_SECURE_FLASH_UART_BOOTLOADER_ALLOW_DEC=y
CONFIG_SECURE_FLASH_UART_BOOTLOADER_ALLOW_CACHE=y
CONFIG_SECURE_FLASH_REQUIRE_ALREADY_ENABLED=y
CONFIG_PARTITION_TABLE_OFFSET=0x9000
CONFIG_EXAMPLE_CONNECT_ETHERNET=y
CONFIG_EXAMPLE_CONNECT_WIFI=n
CONFIG_EXAMPLE_USE_INTERNAL_ETHERNET=y
CONFIG_EXAMPLE_ETH_PHY_IP101=y
CONFIG_EXAMPLE_ETH_MDC_GPIO=23
CONFIG_EXAMPLE_ETH_MDIO_GPIO=18
CONFIG_EXAMPLE_ETH_PHY_RST_GPIO=5
CONFIG_EXAMPLE_ETH_PHY_ADDR=1
# This is required for nvs encryption (which is enabled by default with flash encryption)
CONFIG_PARTITION_TABLE_TWO_OTA_ENCRYPTED_NVS=y
CONFIG_MBEDTLS_TLS_CLIENT_ONLY=y
CONFIG_COMPILER_OPTIMIZATION_SIZE=y
CONFIG_EXAMPLE_CONNECT_IPV6=n

5
examples/system/ota/simple_ota_example/sdkconfig.ci.flash_enc_wifi
View File

@ -11,5 +11,10 @@ CONFIG_SECURE_FLASH_REQUIRE_ALREADY_ENABLED=y
CONFIG_PARTITION_TABLE_OFFSET=0x9000
CONFIG_EXAMPLE_CONNECT_ETHERNET=n
CONFIG_EXAMPLE_CONNECT_WIFI=y
CONFIG_EXAMPLE_WIFI_SSID_PWD_FROM_STDIN=y
# This is required for nvs encryption (which is enabled by default with flash encryption)
CONFIG_PARTITION_TABLE_TWO_OTA_ENCRYPTED_NVS=y
CONFIG_MBEDTLS_TLS_CLIENT_ONLY=y
CONFIG_COMPILER_OPTIMIZATION_SIZE=y
CONFIG_EXAMPLE_CONNECT_IPV6=n

12
pytest.ini
View File

@ -36,16 +36,16 @@ markers =
quad_psram: runners with quad psram
octal_psram: runners with octal psram
usb_host: usb host runners
ethernet_ota: ethernet OTA runners
flash_encryption: Flash Encryption runners
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
flash_encryption_wifi_high_traffic: Flash Encryption runners with wifi high traffic support
ethernet: ethernet runner
ethernet_flash_8m: ethernet runner with 8mb flash
ethernet_router: both the runner and dut connect to the same router
wifi_nearby: runner with a wifi AP nearby
wifi_router: runner can connect to the dut by a wifi router
ethernet_router: both the runner and dut connect to the same router through ethernet NIC
wifi_ap: a wifi AP in the environment
wifi_router: both the runner and dut connect to the same wifi router
wifi_high_traffic: wifi high traffic runners
wifi_wlan: wifi runner with a wireless NIC
deepsleep_temp_tag: temporary env for running potentially harmfull deepsleep related tests

99
tools/ci/python_packages/common_test_methods.py
View File

@ -4,6 +4,7 @@
import logging
import os
import socket
from typing import Any
import netifaces
import yaml
@ -12,43 +13,40 @@ ENV_CONFIG_FILE_SEARCH = [
os.path.join(os.environ['IDF_PATH'], 'EnvConfig.yml'),
os.path.join(os.environ['IDF_PATH'], 'ci-test-runner-configs', os.environ.get('CI_RUNNER_DESCRIPTION', ''), 'EnvConfig.yml'),
]
ENV_CONFIG_TEMPLATE = '''
$IDF_PATH/EnvConfig.yml:
```yaml
<env_name>:
key: var
key2: var2
<another_env>:
key: var
```
'''
def get_my_ip_by_interface(interface_name: str, ip_type: int = netifaces.AF_INET) -> str:
for i in netifaces.ifaddresses(interface_name)[ip_type]:
interface_name = i['addr'].replace('%{}'.format(interface_name), '')
assert isinstance(interface_name, str)
return interface_name
def get_host_ip_by_interface(interface_name: str, ip_type: int = netifaces.AF_INET) -> str:
for _addr in netifaces.ifaddresses(interface_name)[ip_type]:
host_ip = _addr['addr'].replace('%{}'.format(interface_name), '')
assert isinstance(host_ip, str)
return host_ip
return ''
def get_my_ip4_by_getway(getway: str = '') -> str:
getways = netifaces.gateways()
for gw, iface_name, _ in getways[netifaces.AF_INET]:
if gw and gw == getway:
interface = iface_name
break
else:
interface = getways['default'][netifaces.AF_INET][1]
logging.debug('Using interface: {}.'.format(interface))
address = netifaces.ifaddresses(interface)[netifaces.AF_INET]
assert isinstance(address[0]['addr'], str)
return address[0]['addr']
def get_my_ip4_by_dest_ip(dest_ip: str = '') -> str:
def get_host_ip4_by_dest_ip(dest_ip: str = '') -> str:
if not dest_ip:
dest_ip = '8.8.8.8'
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect((dest_ip, 80))
my_ip = s1.getsockname()[0]
host_ip = s1.getsockname()[0]
s1.close()
assert isinstance(my_ip, str)
return my_ip
assert isinstance(host_ip, str)
print(f'Using host ip: {host_ip}')
return host_ip
def get_my_interface_by_dest_ip(dest_ip: str = '') -> str:
my_ip = get_my_ip4_by_dest_ip(dest_ip)
my_ip = get_host_ip4_by_dest_ip(dest_ip)
interfaces = netifaces.interfaces()
for interface in interfaces:
try:
@ -62,22 +60,39 @@ def get_my_interface_by_dest_ip(dest_ip: str = '') -> str:
return ''
def get_env_config(env_key: str = '', config_file: str = '') -> dict:
if not config_file:
for _file in ENV_CONFIG_FILE_SEARCH:
if os.path.exists(_file):
config_file = _file
if not config_file:
return dict()
def get_env_config_variable(env_name: str, key: str, default: Any = None) -> Any:
"""
Get test environment related variable
with open(config_file, 'r') as f:
config = yaml.load(f.read(), Loader=yaml.SafeLoader)
assert isinstance(config, dict)
if not env_key:
return config
if env_key in config:
_config = config[env_key]
assert isinstance(_config, dict)
return _config
logging.warning('Can not get env config, key: {}'.format(env_key))
return dict()
config file format: $IDF_PATH/EnvConfig.yml
```
<env_name>:
key: var
key2: var2
<env_name2>:
key: var
```
"""
config = dict()
for _file in ENV_CONFIG_FILE_SEARCH:
try:
with open(_file, 'r') as f:
data = yaml.load(f.read(), Loader=yaml.SafeLoader)
config = data[env_name]
break
except (FileNotFoundError, KeyError):
pass
else:
pass
var = config.get(key, default)
if var is None:
logging.warning(f'Failed to get env variable {env_name}/{key}.')
logging.info(f'Env config file template: {ENV_CONFIG_TEMPLATE}')
if not os.environ.get('CI_JOB_ID'):
# Try to get variable from stdin
var = input(f'You can input the variable now:')
else:
raise ValueError(f'Env variable not found: {env_name}/{key}')
logging.debug(f'Got env variable {env_name}/{key}: {var}')
return var

6
tools/test_apps/protocols/mqtt/publish_connect_test/app_test.py
View File

@ -15,7 +15,7 @@ from threading import Event, Lock, Thread
import paho.mqtt.client as mqtt
import ttfw_idf
from common_test_methods import get_my_ip4_by_dest_ip
from common_test_methods import get_host_ip4_by_dest_ip
DEFAULT_MSG_SIZE = 16
@ -236,7 +236,7 @@ class TlsServer:
def connection_tests(dut, cases, dut_ip):
ip = get_my_ip4_by_dest_ip(dut_ip)
ip = get_host_ip4_by_dest_ip(dut_ip)
set_server_cert_cn(ip)
server_port = 2222
@ -336,7 +336,7 @@ def test_app_protocol_mqtt_publish_connect(env, extra_data):
raise
dut1.start_app()
esp_ip = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
esp_ip = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0]
print('Got IP={}'.format(esp_ip))
if not os.getenv('MQTT_SKIP_CONNECT_TEST'):