alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2025-10-30 20:51:41 +00:00
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 233dc30fb1
commit c9940a3eed
3 changed files with 78 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
examples/system/ota/simple_ota_example/example_test.py
View File

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