eaw/esp-idf
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix 'address already in use' exception in OTA tests

Browse Source 
We will stop the server instance at the end of each test case. This will solve the "address already in use" exception
This commit is contained in:
Harshit Malpani
2021-11-10 13:54:21 +05:30
parent e6943e970c
commit c480ba851e
3 changed files with 68 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
examples/system/ota/advanced_https_ota/example_test.py
View File

@@ -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__':