diff --git a/examples/system/ota/advanced_https_ota/example_test.py b/examples/system/ota/advanced_https_ota/example_test.py index 8167ede97e..65430f9679 100644 --- a/examples/system/ota/advanced_https_ota/example_test.py +++ b/examples/system/ota/advanced_https_ota/example_test.py @@ -1,4 +1,5 @@ import http.server +import multiprocessing import os import random import re @@ -6,7 +7,6 @@ import socket import ssl import struct import subprocess -from threading import Thread import ttfw_idf from tiny_test_fw import DUT @@ -190,7 +190,7 @@ def test_examples_protocol_advanced_https_ota_example(env, extra_data): # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -201,7 +201,7 @@ def test_examples_protocol_advanced_https_ota_example(env, extra_data): print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - thread1.close() + thread1.terminate() dut1.expect('Starting Advanced OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)) @@ -209,6 +209,7 @@ def test_examples_protocol_advanced_https_ota_example(env, extra_data): dut1.expect('Loaded app from partition at offset', timeout=60) dut1.expect('Starting Advanced OTA example', timeout=30) dut1.reset() + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_WIFI') @@ -244,7 +245,7 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_d # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -254,12 +255,14 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_d print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting Advanced OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)) dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name) dut1.expect('Image validation failed, image is corrupted', timeout=30) os.remove(binary_file) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_WIFI') @@ -294,7 +297,7 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extr # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -304,12 +307,14 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extr print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting Advanced OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)) dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name) dut1.expect('advanced_https_ota_example: esp_https_ota_read_img_desc failed', timeout=30) os.remove(binary_file) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_WIFI') @@ -343,7 +348,7 @@ def test_examples_protocol_advanced_https_ota_example_random(env, extra_data): # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -353,12 +358,14 @@ def test_examples_protocol_advanced_https_ota_example_random(env, extra_data): print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting Advanced OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)) dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name) dut1.expect('esp_ota_ops: OTA image has invalid magic byte', timeout=10) os.remove(binary_file) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_WIFI') @@ -423,10 +430,10 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_da # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() - thread2 = Thread(target=start_redirect_server, args=(dut1.app.binary_path, host_ip, redirection_server_port, server_port)) + thread2 = multiprocessing.Process(target=start_redirect_server, args=(dut1.app.binary_path, host_ip, redirection_server_port, server_port)) thread2.daemon = True thread2.start() dut1.start_app() @@ -436,8 +443,8 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_da print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - thread1.close() - thread2.close() + thread1.terminate() + thread2.terminate() dut1.expect('Starting Advanced OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name)) @@ -445,6 +452,8 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_da dut1.expect('Loaded app from partition at offset', timeout=60) dut1.expect('Starting Advanced OTA example', timeout=30) dut1.reset() + thread1.terminate() + thread2.terminate() @ttfw_idf.idf_example_test(env_tag='Example_8Mflash_Ethernet') @@ -481,7 +490,7 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_d # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -492,6 +501,7 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_d print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting Advanced OTA example', timeout=30) # Use originally generated image with secure_version=1 @@ -507,7 +517,8 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_d print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name)) dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name) dut1.expect('New firmware security version is less than eFuse programmed, 0 < 1', timeout=30) - os.remove(anti_rollback_bin_name) + os.remove(binary_file) + thread1.terminate() if __name__ == '__main__': diff --git a/examples/system/ota/native_ota_example/example_test.py b/examples/system/ota/native_ota_example/example_test.py index 66d92baa99..9c4b21fc0d 100644 --- a/examples/system/ota/native_ota_example/example_test.py +++ b/examples/system/ota/native_ota_example/example_test.py @@ -1,4 +1,5 @@ import http.server +import multiprocessing import os import random import re @@ -6,7 +7,6 @@ import socket import ssl import struct import subprocess -from threading import Thread import ttfw_idf from tiny_test_fw import DUT @@ -157,7 +157,7 @@ def test_examples_protocol_native_ota_example(env, extra_data): # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -168,7 +168,7 @@ def test_examples_protocol_native_ota_example(env, extra_data): print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - thread1.close() + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)) @@ -176,6 +176,7 @@ def test_examples_protocol_native_ota_example(env, extra_data): dut1.expect('Loaded app from partition at offset', timeout=60) dut1.expect('Starting OTA example', timeout=30) dut1.reset() + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_WIFI') @@ -211,7 +212,7 @@ def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data): # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -221,12 +222,14 @@ def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data): print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)) dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name) dut1.expect('native_ota_example: Image validation failed, image is corrupted', timeout=20) os.remove(binary_file) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_WIFI') @@ -261,7 +264,7 @@ def test_examples_protocol_native_ota_example_truncated_header(env, extra_data): # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -271,12 +274,14 @@ def test_examples_protocol_native_ota_example_truncated_header(env, extra_data): print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)) dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name) dut1.expect('native_ota_example: received package is not fit len', timeout=20) os.remove(binary_file) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_WIFI') @@ -310,7 +315,7 @@ def test_examples_protocol_native_ota_example_random(env, extra_data): # start test host_ip = get_my_ip() if (get_server_status(host_ip, server_port) is False): - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port)) thread1.daemon = True thread1.start() dut1.start_app() @@ -320,12 +325,14 @@ def test_examples_protocol_native_ota_example_random(env, extra_data): print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)) dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name) dut1.expect('esp_ota_ops: OTA image has invalid magic byte', timeout=20) os.remove(binary_file) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_WIFI') diff --git a/examples/system/ota/simple_ota_example/example_test.py b/examples/system/ota/simple_ota_example/example_test.py index f2cb87868c..c2ef78a71b 100644 --- a/examples/system/ota/simple_ota_example/example_test.py +++ b/examples/system/ota/simple_ota_example/example_test.py @@ -1,9 +1,10 @@ import http.server +import multiprocessing import os import re import socket import ssl -from threading import Thread +import sys import ttfw_idf from tiny_test_fw import DUT, Utility @@ -67,11 +68,16 @@ def get_my_ip(): return my_ip -def start_https_server(ota_image_dir, server_ip, server_port): - # parser = argparse.ArgumentParser() - # parser.add_argument('-p', '--port', dest='port', type= int, - # help= "Server Port", default= 8000) - # args = parser.parse_args() +def get_server_status(host_ip, server_port): + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_status = sock.connect_ex((host_ip, server_port)) + sock.close() + if server_status == 0: + return True + return False + + +def start_https_server(ota_image_dir, server_ip, server_port, server_file=None, key_file=None): os.chdir(ota_image_dir) server_file = os.path.join(ota_image_dir, 'server_cert.pem') @@ -131,9 +137,10 @@ def test_examples_protocol_simple_ota_example(env, extra_data): sha256_bootloader, sha256_app = calc_all_sha256(dut1) # start test host_ip = get_my_ip() - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) - thread1.daemon = True - thread1.start() + if (get_server_status(host_ip, 8000) is False): + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) + thread1.daemon = True + thread1.start() dut1.start_app() dut1.expect('Loaded app from partition at offset 0x10000', timeout=30) check_sha256(sha256_bootloader, dut1.expect(re.compile(r'SHA-256 for bootloader:\s+([a-f0-9]+)'))[0]) @@ -143,13 +150,14 @@ def test_examples_protocol_simple_ota_example(env, extra_data): print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - thread1.close() + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin')) dut1.write('https://' + host_ip + ':8000/simple_ota.bin') dut1.expect('Loaded app from partition at offset 0x110000', timeout=60) dut1.expect('Starting OTA example', timeout=30) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_EthKitV1') @@ -167,9 +175,10 @@ def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(env, e ttfw_idf.log_performance('simple_ota_bin_size', '{}KB'.format(bin_size // 1024)) # start test host_ip = get_my_ip() - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) - thread1.daemon = True - thread1.start() + if (get_server_status(host_ip, 8000) is False): + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) + thread1.daemon = True + thread1.start() dut1.start_app() dut1.expect('Loaded app from partition at offset 0x10000', timeout=30) try: @@ -177,13 +186,14 @@ def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(env, e print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - thread1.close() + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin')) dut1.write('https://' + host_ip + ':8000/simple_ota.bin') dut1.expect('Loaded app from partition at offset 0x110000', timeout=60) dut1.expect('Starting OTA example', timeout=30) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_Flash_Encryption_OTA') @@ -204,9 +214,10 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption(env, extra_d dut1.erase_flash() # start test host_ip = get_my_ip() - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) - thread1.daemon = True - thread1.start() + if (get_server_status(host_ip, 8000) is False): + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) + thread1.daemon = True + thread1.start() dut1.start_app() dut1.expect('Loaded app from partition at offset 0x20000', timeout=30) dut1.expect('Flash encryption mode is DEVELOPMENT (not secure)', timeout=10) @@ -215,7 +226,7 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption(env, extra_d print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - thread1.close() + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin')) @@ -223,6 +234,7 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption(env, extra_d dut1.expect('Loaded app from partition at offset 0x120000', timeout=60) dut1.expect('Flash encryption mode is DEVELOPMENT (not secure)', timeout=10) dut1.expect('Starting OTA example', timeout=30) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_EthKitV1') @@ -242,9 +254,10 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat sha256_bootloader, sha256_app = calc_all_sha256(dut1) # start test host_ip = get_my_ip() - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) - thread1.daemon = True - thread1.start() + if (get_server_status(host_ip, 8000) is False): + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) + thread1.daemon = True + thread1.start() dut1.start_app() dut1.expect('Loaded app from partition at offset 0x20000', timeout=30) check_sha256(sha256_bootloader, dut1.expect(re.compile(r'SHA-256 for bootloader:\s+([a-f0-9]+)'))[0]) @@ -254,6 +267,7 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin')) @@ -264,6 +278,7 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat dut1.expect('Loaded app from partition at offset 0x120000', timeout=20) dut1.expect('Starting OTA example', timeout=30) + thread1.terminate() @ttfw_idf.idf_example_test(env_tag='Example_EthKitV12') @@ -283,9 +298,10 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat sha256_bootloader, sha256_app = calc_all_sha256(dut1) # start test host_ip = get_my_ip() - thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) - thread1.daemon = True - thread1.start() + if (get_server_status(host_ip, 8000) is False): + thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) + thread1.daemon = True + thread1.start() dut1.start_app() dut1.expect('Loaded app from partition at offset 0x20000', timeout=30) check_sha256(sha256_bootloader, dut1.expect(re.compile(r'SHA-256 for bootloader:\s+([a-f0-9]+)'))[0]) @@ -295,6 +311,7 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat print('Connected to AP with IP: {}'.format(ip_address)) except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + thread1.terminate() dut1.expect('Starting OTA example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin')) @@ -308,6 +325,7 @@ def test_examples_protocol_simple_ota_example_with_verify_app_signature_on_updat dut1.expect('Loaded app from partition at offset 0x120000', timeout=20) dut1.expect('Starting OTA example', timeout=30) + thread1.terminate() if __name__ == '__main__':