kopia lustrzana https://github.com/espressif/esp-idf
Fix 'address already in use' exception in OTA tests
We will stop the server instance at the end of each test case. This will solve the "address already in use" exceptionpull/7966/head
rodzic
9044374032
commit
b93d63b067
|
@ -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 RangeHTTPServer import RangeRequestHandler
|
||||
|
@ -191,7 +191,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()
|
||||
|
@ -202,7 +202,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))
|
||||
|
@ -210,6 +210,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_ETH_OTA')
|
||||
|
@ -245,7 +246,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()
|
||||
|
@ -255,12 +256,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_ETH_OTA')
|
||||
|
@ -295,7 +298,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()
|
||||
|
@ -305,12 +308,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_ETH_OTA')
|
||||
|
@ -344,7 +349,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()
|
||||
|
@ -354,12 +359,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_ETH_OTA')
|
||||
|
@ -424,10 +431,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()
|
||||
|
@ -437,8 +444,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))
|
||||
|
@ -446,6 +453,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')
|
||||
|
@ -485,7 +494,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()
|
||||
|
@ -496,6 +505,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
|
||||
|
@ -511,7 +521,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()
|
||||
|
||||
|
||||
@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
|
||||
|
@ -537,7 +548,7 @@ def test_examples_protocol_advanced_https_ota_example_partial_request(env, extra
|
|||
# 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()
|
||||
|
@ -548,6 +559,7 @@ def test_examples_protocol_advanced_https_ota_example_partial_request(env, extra
|
|||
except DUT.ExpectTimeout:
|
||||
Utility.console_log('ENV_TEST_FAILURE: Cannot connect to AP')
|
||||
raise
|
||||
thread1.terminate()
|
||||
dut1.expect('Starting Advanced OTA example', timeout=30)
|
||||
|
||||
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
|
||||
|
@ -557,6 +569,7 @@ def test_examples_protocol_advanced_https_ota_example_partial_request(env, extra
|
|||
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_OTA', nightly_run=True)
|
||||
|
@ -580,7 +593,7 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(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()
|
||||
dut1.start_app()
|
||||
|
@ -590,6 +603,7 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(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.terminate()
|
||||
|
||||
dut1.expect('Starting Advanced OTA example', timeout=30)
|
||||
print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
|
||||
|
@ -600,6 +614,7 @@ def test_examples_protocol_advanced_https_ota_example_nimble_gatts(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()
|
||||
|
||||
|
||||
@ttfw_idf.idf_example_test(env_tag='Example_WIFI_OTA', nightly_run=True)
|
||||
|
@ -623,7 +638,7 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(env, extra
|
|||
# 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()
|
||||
|
@ -633,6 +648,7 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(env, extra
|
|||
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) + '/' + bin_name))
|
||||
|
@ -643,6 +659,7 @@ def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(env, extra
|
|||
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_ETH_OTA')
|
||||
|
|
|
@ -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_ETH_OTA')
|
||||
|
@ -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_ETH_OTA')
|
||||
|
@ -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_ETH_OTA')
|
||||
|
@ -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_ETH_OTA')
|
||||
|
|
|
@ -1,10 +1,10 @@
|
|||
import http.server
|
||||
import multiprocessing
|
||||
import os
|
||||
import re
|
||||
import socket
|
||||
import ssl
|
||||
import sys
|
||||
from threading import Thread
|
||||
|
||||
import ttfw_idf
|
||||
from tiny_test_fw import DUT, Utility
|
||||
|
@ -68,6 +68,15 @@ def get_my_ip():
|
|||
return my_ip
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
@ -130,9 +139,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])
|
||||
|
@ -142,12 +152,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.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')
|
||||
|
@ -165,9 +177,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:
|
||||
|
@ -175,12 +188,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.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')
|
||||
|
@ -201,9 +216,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)
|
||||
|
@ -212,6 +228,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.terminate()
|
||||
dut1.expect('Starting OTA example', timeout=30)
|
||||
|
||||
print('writing to device: {}'.format('https://' + host_ip + ':8000/simple_ota.bin'))
|
||||
|
@ -219,6 +236,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_Flash_Encryption_OTA_WiFi', target=['esp32c3'], nightly_run=True)
|
||||
|
@ -239,9 +257,10 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(env, ex
|
|||
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)
|
||||
|
@ -250,6 +269,7 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(env, ex
|
|||
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'))
|
||||
|
@ -257,6 +277,7 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption_wifi(env, ex
|
|||
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')
|
||||
|
@ -276,9 +297,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])
|
||||
|
@ -288,6 +310,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'))
|
||||
|
@ -298,6 +321,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')
|
||||
|
@ -317,9 +341,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])
|
||||
|
@ -329,6 +354,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'))
|
||||
|
@ -342,6 +368,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__':
|
||||
|
|
Ładowanie…
Reference in New Issue