diff --git a/examples/system/ota/advanced_https_ota/example_test.py b/examples/system/ota/advanced_https_ota/example_test.py index 9189ced414..20a18344b0 100644 --- a/examples/system/ota/advanced_https_ota/example_test.py +++ b/examples/system/ota/advanced_https_ota/example_test.py @@ -1,9 +1,11 @@ -import re +import http.server +import multiprocessing import os -import struct +import random +import re import socket -from threading import Thread import ssl +import struct from tiny_test_fw import DUT import ttfw_idf @@ -169,7 +171,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() @@ -180,14 +182,15 @@ 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() - dut1.expect("Starting Advanced OTA example", timeout=30) + thread1.terminate() + dut1.expect('Starting Advanced OTA example', timeout=30) print("writing to device: {}".format("https://" + host_ip + ":" + str(server_port) + "/" + bin_name)) dut1.write("https://" + host_ip + ":" + str(server_port) + "/" + bin_name) dut1.expect("Loaded app from partition at offset", timeout=60) dut1.expect("Starting Advanced OTA example", timeout=30) dut1.reset() + thread1.terminate() @ttfw_idf.idf_example_test(env_tag="Example_WIFI") @@ -224,7 +227,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() @@ -234,12 +237,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') - dut1.expect("Starting Advanced OTA example", timeout=30) + 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") @@ -275,7 +280,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() @@ -285,12 +290,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') - dut1.expect("Starting Advanced OTA example", timeout=30) + 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") @@ -325,7 +332,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() @@ -335,12 +342,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') - dut1.expect("Starting Advanced OTA example", timeout=30) + 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") @@ -407,10 +416,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() @@ -420,15 +429,17 @@ 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() - dut1.expect("Starting Advanced OTA example", timeout=30) + 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)) dut1.write("https://" + host_ip + ":" + str(redirection_server_port) + "/" + bin_name) dut1.expect("Loaded app from partition at offset", timeout=60) dut1.expect("Starting Advanced OTA example", timeout=30) dut1.reset() + thread1.terminate() + thread2.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 b01708bb70..8fed06416c 100644 --- a/examples/system/ota/native_ota_example/example_test.py +++ b/examples/system/ota/native_ota_example/example_test.py @@ -1,8 +1,10 @@ -import re +import http.server +import multiprocessing import os +import random +import re import struct import socket -from threading import Thread import ssl from tiny_test_fw import DUT @@ -141,7 +143,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() @@ -152,14 +154,15 @@ 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() - dut1.expect("Starting OTA example", timeout=30) + thread1.terminate() + dut1.expect('Starting OTA example', timeout=30) print("writing to device: {}".format("https://" + host_ip + ":" + str(server_port) + "/" + bin_name)) dut1.write("https://" + host_ip + ":" + str(server_port) + "/" + bin_name) dut1.expect("Loaded app from partition at offset", timeout=60) dut1.expect("Starting OTA example", timeout=30) dut1.reset() + thread1.terminate() @ttfw_idf.idf_example_test(env_tag="Example_WIFI") @@ -196,7 +199,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() @@ -206,12 +209,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') - dut1.expect("Starting OTA example", timeout=30) + 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") @@ -247,7 +252,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() @@ -257,12 +262,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') - dut1.expect("Starting OTA example", timeout=30) + 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") @@ -297,7 +304,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() @@ -307,12 +314,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') - dut1.expect("Starting OTA example", timeout=30) + 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 c66efae7ac..ec69865bde 100644 --- a/examples/system/ota/simple_ota_example/example_test.py +++ b/examples/system/ota/simple_ota_example/example_test.py @@ -1,8 +1,11 @@ +import http.server +import multiprocessing +import os import re import os import socket -from threading import Thread import ssl +import sys from tiny_test_fw import DUT import ttfw_idf @@ -73,11 +76,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") @@ -115,9 +123,10 @@ def test_examples_protocol_simple_ota_example(env, extra_data): ttfw_idf.check_performance("simple_ota_bin_size", 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: @@ -125,13 +134,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() if __name__ == '__main__':