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

style: format python files with isort and double-quote-string-fixer

Browse Source
This commit is contained in:
Fu Hanxi
2021-01-26 10:49:01 +08:00
parent dc8402ea61
commit 0146f258d7
276 changed files with 8241 additions and 8162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
examples/system/app_trace_to_host/example_test.py
View File

@@ -1,10 +1,12 @@
from __future__ import unicode_literals
import os
import re
import ttfw_idf
@ttfw_idf.idf_example_test(env_tag="test_jtag_arm")
@ttfw_idf.idf_example_test(env_tag='test_jtag_arm')
def test_examples_app_trace_to_host(env, extra_data):
rel_project_path = os.path.join('examples', 'system', 'app_trace_to_host')
dut = env.get_dut('app_trace_to_host', rel_project_path)
@@ -22,7 +24,7 @@ def test_examples_app_trace_to_host(env, extra_data):
re.compile(r'example: Collected \d+ samples in 20 ms.'),
timeout=20)
ocd.apptrace_start("file://adc.log 0 9000 5 0 0")
ocd.apptrace_start('file://adc.log 0 9000 5 0 0')
ocd.apptrace_wait_stop(tmo=30)
with ttfw_idf.CustomProcess(' '.join([os.path.join(idf_path, 'tools/esp_app_trace/logtrace_proc.py'),

4
examples/system/base_mac_address/example_test.py
View File

@@ -1,7 +1,9 @@
from __future__ import unicode_literals
from tiny_test_fw import Utility
import re
import ttfw_idf
from tiny_test_fw import Utility
@ttfw_idf.idf_example_test(env_tag='Example_GENERIC')

8
examples/system/console/example_test.py
View File

@@ -6,15 +6,15 @@ import ttfw_idf
@ttfw_idf.idf_example_test(env_tag='Example_GENERIC')
def test_examples_system_console(env, extra_data):
dut = env.get_dut('console_example', 'examples/system/console', app_config_name='history')
print("Using binary path: {}".format(dut.app.binary_path))
print('Using binary path: {}'.format(dut.app.binary_path))
dut.start_app()
dut.expect("Command history enabled")
dut.expect('Command history enabled')
env.close_dut(dut.name)
dut = env.get_dut('console_example', 'examples/system/console', app_config_name='nohistory')
print("Using binary path: {}".format(dut.app.binary_path))
print('Using binary path: {}'.format(dut.app.binary_path))
dut.start_app()
dut.expect("Command history disabled")
dut.expect('Command history disabled')
if __name__ == '__main__':

2
examples/system/deep_sleep/example_test.py
View File

@@ -1,5 +1,7 @@
from __future__ import unicode_literals
import re
import ttfw_idf

2
examples/system/efuse/example_test.py
View File

@@ -1,5 +1,7 @@
from __future__ import unicode_literals
import re
import ttfw_idf

76
examples/system/esp_event/default_event_loop/example_test.py
View File

@@ -5,80 +5,80 @@ import ttfw_idf
# Timer events
TIMER_EVENT_LIMIT = 3
TIMER_EXPIRY_HANDLING = "TIMER_EVENTS:TIMER_EVENT_EXPIRY: timer_expiry_handler, executed {} out of " + str(TIMER_EVENT_LIMIT) + " times"
TIMER_EXPIRY_HANDLING = 'TIMER_EVENTS:TIMER_EVENT_EXPIRY: timer_expiry_handler, executed {} out of ' + str(TIMER_EVENT_LIMIT) + ' times'
# Task events
TASK_ITERATION_LIMIT = 5
TASK_UNREGISTRATION_LIMIT = 3
TASK_ITERATION_POST = "TASK_EVENTS:TASK_ITERATION_EVENT: posting to default loop, {} out of " + str(TASK_ITERATION_LIMIT)
TASK_ITERATION_HANDLING = "TASK_EVENTS:TASK_ITERATION_EVENT: task_iteration_handler, executed {} times"
TASK_ITERATION_POST = 'TASK_EVENTS:TASK_ITERATION_EVENT: posting to default loop, {} out of ' + str(TASK_ITERATION_LIMIT)
TASK_ITERATION_HANDLING = 'TASK_EVENTS:TASK_ITERATION_EVENT: task_iteration_handler, executed {} times'
def _test_timer_events(dut):
dut.start_app()
print("Checking timer events posting and handling")
print('Checking timer events posting and handling')
dut.expect("setting up")
dut.expect("starting event sources")
dut.expect('setting up')
dut.expect('starting event sources')
print("Finished setup")
print('Finished setup')
dut.expect("TIMER_EVENTS:TIMER_EVENT_STARTED: posting to default loop")
print("Posted timer started event")
dut.expect("TIMER_EVENTS:TIMER_EVENT_STARTED: timer_started_handler, instance 0")
dut.expect("TIMER_EVENTS:TIMER_EVENT_STARTED: timer_started_handler, instance 1")
dut.expect("TIMER_EVENTS:TIMER_EVENT_STARTED: timer_started_handler_2")
dut.expect("TIMER_EVENTS:TIMER_EVENT_STARTED: timer_any_handler")
dut.expect("TIMER_EVENTS:TIMER_EVENT_STARTED: all_event_handler")
print("Handled timer started event")
dut.expect('TIMER_EVENTS:TIMER_EVENT_STARTED: posting to default loop')
print('Posted timer started event')
dut.expect('TIMER_EVENTS:TIMER_EVENT_STARTED: timer_started_handler, instance 0')
dut.expect('TIMER_EVENTS:TIMER_EVENT_STARTED: timer_started_handler, instance 1')
dut.expect('TIMER_EVENTS:TIMER_EVENT_STARTED: timer_started_handler_2')
dut.expect('TIMER_EVENTS:TIMER_EVENT_STARTED: timer_any_handler')
dut.expect('TIMER_EVENTS:TIMER_EVENT_STARTED: all_event_handler')
print('Handled timer started event')
for expiries in range(1, TIMER_EVENT_LIMIT + 1):
dut.expect("TIMER_EVENTS:TIMER_EVENT_EXPIRY: posting to default loop")
print("Posted timer expiry event {} out of {}".format(expiries, TIMER_EVENT_LIMIT))
dut.expect('TIMER_EVENTS:TIMER_EVENT_EXPIRY: posting to default loop')
print('Posted timer expiry event {} out of {}'.format(expiries, TIMER_EVENT_LIMIT))
if expiries >= TIMER_EVENT_LIMIT:
dut.expect("TIMER_EVENTS:TIMER_EVENT_STOPPED: posting to default loop")
print("Posted timer stopped event")
dut.expect('TIMER_EVENTS:TIMER_EVENT_STOPPED: posting to default loop')
print('Posted timer stopped event')
dut.expect(TIMER_EXPIRY_HANDLING.format(expiries))
dut.expect("TIMER_EVENTS:TIMER_EVENT_EXPIRY: timer_any_handler")
dut.expect("TIMER_EVENTS:TIMER_EVENT_EXPIRY: all_event_handler")
dut.expect('TIMER_EVENTS:TIMER_EVENT_EXPIRY: timer_any_handler')
dut.expect('TIMER_EVENTS:TIMER_EVENT_EXPIRY: all_event_handler')
print("Handled timer expiry event {} out of {}".format(expiries, TIMER_EVENT_LIMIT))
print('Handled timer expiry event {} out of {}'.format(expiries, TIMER_EVENT_LIMIT))
dut.expect("TIMER_EVENTS:TIMER_EVENT_STOPPED: timer_stopped_handler")
dut.expect("TIMER_EVENTS:TIMER_EVENT_STOPPED: deleted timer event source")
print("Handled timer stopped event")
dut.expect('TIMER_EVENTS:TIMER_EVENT_STOPPED: timer_stopped_handler')
dut.expect('TIMER_EVENTS:TIMER_EVENT_STOPPED: deleted timer event source')
print('Handled timer stopped event')
def _test_iteration_events(dut):
dut.start_app()
print("Checking iteration events posting and handling")
dut.expect("setting up")
dut.expect("starting event sources")
print("Finished setup")
print('Checking iteration events posting and handling')
dut.expect('setting up')
dut.expect('starting event sources')
print('Finished setup')
for iteration in range(1, TASK_ITERATION_LIMIT + 1):
dut.expect(TASK_ITERATION_POST.format(iteration))
print("Posted iteration {} out of {}".format(iteration, TASK_ITERATION_LIMIT))
print('Posted iteration {} out of {}'.format(iteration, TASK_ITERATION_LIMIT))
if iteration < TASK_UNREGISTRATION_LIMIT:
dut.expect(TASK_ITERATION_HANDLING.format(iteration))
dut.expect("TASK_EVENTS:TASK_ITERATION_EVENT: all_event_handler")
dut.expect('TASK_EVENTS:TASK_ITERATION_EVENT: all_event_handler')
elif iteration == TASK_UNREGISTRATION_LIMIT:
dut.expect("TASK_EVENTS:TASK_ITERATION_EVENT: unregistering task_iteration_handler")
dut.expect("TASK_EVENTS:TASK_ITERATION_EVENT: all_event_handler")
print("Unregistered handler at iteration {} out of {}".format(iteration, TASK_ITERATION_LIMIT))
dut.expect('TASK_EVENTS:TASK_ITERATION_EVENT: unregistering task_iteration_handler')
dut.expect('TASK_EVENTS:TASK_ITERATION_EVENT: all_event_handler')
print('Unregistered handler at iteration {} out of {}'.format(iteration, TASK_ITERATION_LIMIT))
else:
dut.expect("TASK_EVENTS:TASK_ITERATION_EVENT: all_event_handler")
dut.expect('TASK_EVENTS:TASK_ITERATION_EVENT: all_event_handler')
print("Handled iteration {} out of {}".format(iteration, TASK_ITERATION_LIMIT))
print('Handled iteration {} out of {}'.format(iteration, TASK_ITERATION_LIMIT))
dut.expect("TASK_EVENTS:TASK_ITERATION_EVENT: deleting task event source")
print("Deleted task event source")
dut.expect('TASK_EVENTS:TASK_ITERATION_EVENT: deleting task event source')
print('Deleted task event source')
@ttfw_idf.idf_example_test(env_tag='Example_GENERIC')

24
examples/system/esp_event/user_event_loops/example_test.py
View File

@@ -4,8 +4,8 @@ import ttfw_idf
TASK_ITERATION_LIMIT = 10
TASK_ITERATION_POSTING = "posting TASK_EVENTS:TASK_ITERATION_EVENT to {}, iteration {} out of " + str(TASK_ITERATION_LIMIT)
TASK_ITERATION_HANDLING = "handling TASK_EVENTS:TASK_ITERATION_EVENT from {}, iteration {}"
TASK_ITERATION_POSTING = 'posting TASK_EVENTS:TASK_ITERATION_EVENT to {}, iteration {} out of ' + str(TASK_ITERATION_LIMIT)
TASK_ITERATION_HANDLING = 'handling TASK_EVENTS:TASK_ITERATION_EVENT from {}, iteration {}'
@ttfw_idf.idf_example_test(env_tag='Example_GENERIC')
@@ -14,26 +14,26 @@ def test_user_event_loops_example(env, extra_data):
dut.start_app()
dut.expect("setting up")
dut.expect("starting event source")
dut.expect("starting application task")
print("Finished setup")
dut.expect('setting up')
dut.expect('starting event source')
dut.expect('starting application task')
print('Finished setup')
for iteration in range(1, TASK_ITERATION_LIMIT + 1):
loop = None
if (iteration % 2 == 0):
loop = "loop_with_task"
loop = 'loop_with_task'
else:
loop = "loop_without_task"
loop = 'loop_without_task'
dut.expect(TASK_ITERATION_POSTING.format(loop, iteration))
print("Posted iteration {} to {}".format(iteration, loop))
print('Posted iteration {} to {}'.format(iteration, loop))
dut.expect(TASK_ITERATION_HANDLING.format(loop, iteration))
print("Handled iteration {} from {}".format(iteration, loop))
print('Handled iteration {} from {}'.format(iteration, loop))
dut.expect("deleting task event source")
print("Deleted task event source")
dut.expect('deleting task event source')
print('Deleted task event source')
if __name__ == '__main__':

1
examples/system/esp_timer/example_test.py
View File

@@ -1,4 +1,5 @@
from __future__ import print_function
import re
import ttfw_idf

2
examples/system/freertos/real_time_stats/example_test.py
View File

@@ -3,7 +3,7 @@ from __future__ import print_function
import ttfw_idf
STATS_TASK_ITERS = 3
STATS_TASK_EXPECT = "Real time stats obtained"
STATS_TASK_EXPECT = 'Real time stats obtained'
@ttfw_idf.idf_example_test(env_tag='Example_GENERIC')

5
examples/system/gcov/example_test.py
View File

@@ -1,12 +1,13 @@
from __future__ import unicode_literals
import os
import time
from ttfw_idf import Utility
import ttfw_idf
from ttfw_idf import Utility
@ttfw_idf.idf_example_test(env_tag="test_jtag_arm")
@ttfw_idf.idf_example_test(env_tag='test_jtag_arm')
def test_examples_gcov(env, extra_data):
rel_project_path = os.path.join('examples', 'system', 'gcov')

2
examples/system/himem/example_test.py
View File

@@ -1,5 +1,7 @@
from __future__ import unicode_literals
import re
import ttfw_idf

1
examples/system/light_sleep/example_test.py
View File

@@ -1,4 +1,5 @@
from __future__ import print_function
import re
import time

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

@@ -1,71 +1,71 @@
import re
import os
import struct
import socket
import http.server
from threading import Thread
import ssl
from tiny_test_fw import DUT
import ttfw_idf
import os
import random
import re
import socket
import ssl
import struct
import subprocess
from threading import Thread
server_cert = "-----BEGIN CERTIFICATE-----\n" \
"MIIDXTCCAkWgAwIBAgIJAP4LF7E72HakMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV\n"\
"BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX\n"\
"aWRnaXRzIFB0eSBMdGQwHhcNMTkwNjA3MDk1OTE2WhcNMjAwNjA2MDk1OTE2WjBF\n"\
"MQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50\n"\
"ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\n"\
"CgKCAQEAlzfCyv3mIv7TlLkObxunKfCdrJ/zgdANrsx0RBtpEPhV560hWJ0fEin0\n"\
"nIOMpJSiF9E6QsPdr6Q+eogH4XnOMU9JE+iG743N1dPfGEzJvRlyct/Ck8SswKPC\n"\
"9+VXsnOdZmUw9y/xtANbURA/TspvPzz3Avv382ffffrJGh7ooOmaZSCZFlSYHLZA\n"\
"w/XlRr0sSRbLpFGY0gXjaAV8iHHiPDYLy4kZOepjV9U51xi+IGsL4w75zuMgsHyF\n"\
"3nJeGYHgtGVBrkL0ZKG5udY0wcBjysjubDJC4iSlNiq2HD3fhs7j6CZddV2v845M\n"\
"lVKNxP0kO4Uj4D8r+5USWC8JKfAwxQIDAQABo1AwTjAdBgNVHQ4EFgQU6OE7ssfY\n"\
"IIPTDThiUoofUpsD5NwwHwYDVR0jBBgwFoAU6OE7ssfYIIPTDThiUoofUpsD5Nww\n"\
"DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAXIlHS/FJWfmcinUAxyBd\n"\
"/xd5Lu8ykeru6oaUCci+Vk9lyoMMES7lQ+b/00d5x7AcTawkTil9EWpBTPTOTraA\n"\
"lzJMQhNKmSLk0iIoTtAJtSZgUSpIIozqK6lenxQQDsHbXKU6h+u9H6KZE8YcjsFl\n"\
"6vL7sw9BVotw/VxfgjQ5OSGLgoLrdVT0z5C2qOuwOgz1c7jNiJhtMdwN+cOtnJp2\n"\
"fuBgEYyE3eeuWogvkWoDcIA8r17Ixzkpq2oJsdvZcHZPIZShPKW2SHUsl98KDemu\n"\
"y0pQyExmQUbwKE4vbFb9XuWCcL9XaOHQytyszt2DeD67AipvoBwVU7/LBOvqnsmy\n"\
"hA==\n"\
"-----END CERTIFICATE-----\n"
import ttfw_idf
from tiny_test_fw import DUT
server_key = "-----BEGIN PRIVATE KEY-----\n"\
"MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCXN8LK/eYi/tOU\n"\
"uQ5vG6cp8J2sn/OB0A2uzHREG2kQ+FXnrSFYnR8SKfScg4yklKIX0TpCw92vpD56\n"\
"iAfhec4xT0kT6Ibvjc3V098YTMm9GXJy38KTxKzAo8L35Veyc51mZTD3L/G0A1tR\n"\
"ED9Oym8/PPcC+/fzZ999+skaHuig6ZplIJkWVJgctkDD9eVGvSxJFsukUZjSBeNo\n"\
"BXyIceI8NgvLiRk56mNX1TnXGL4gawvjDvnO4yCwfIXecl4ZgeC0ZUGuQvRkobm5\n"\
"1jTBwGPKyO5sMkLiJKU2KrYcPd+GzuPoJl11Xa/zjkyVUo3E/SQ7hSPgPyv7lRJY\n"\
"Lwkp8DDFAgMBAAECggEAfBhAfQE7mUByNbxgAgI5fot9eaqR1Nf+QpJ6X2H3KPwC\n"\
"02sa0HOwieFwYfj6tB1doBoNq7i89mTc+QUlIn4pHgIowHO0OGawomeKz5BEhjCZ\n"\
"4XeLYGSoODary2+kNkf2xY8JTfFEcyvGBpJEwc4S2VyYgRRx+IgnumTSH+N5mIKZ\n"\
"SXWNdZIuHEmkwod+rPRXs6/r+PH0eVW6WfpINEbr4zVAGXJx2zXQwd2cuV1GTJWh\n"\
"cPVOXLu+XJ9im9B370cYN6GqUnR3fui13urYbnWnEf3syvoH/zuZkyrVChauoFf8\n"\
"8EGb74/HhXK7Q2s8NRakx2c7OxQifCbcy03liUMmyQKBgQDFAob5B/66N4Q2cq/N\n"\
"MWPf98kYBYoLaeEOhEJhLQlKk0pIFCTmtpmUbpoEes2kCUbH7RwczpYko8tlKyoB\n"\
"6Fn6RY4zQQ64KZJI6kQVsjkYpcP/ihnOY6rbds+3yyv+4uPX7Eh9sYZwZMggE19M\n"\
"CkFHkwAjiwqhiiSlUxe20sWmowKBgQDEfx4lxuFzA1PBPeZKGVBTxYPQf+DSLCre\n"\
"ZFg3ZmrxbCjRq1O7Lra4FXWD3dmRq7NDk79JofoW50yD8wD7I0B7opdDfXD2idO8\n"\
"0dBnWUKDr2CAXyoLEINce9kJPbx4kFBQRN9PiGF7VkDQxeQ3kfS8CvcErpTKCOdy\n"\
"5wOwBTwJdwKBgDiTFTeGeDv5nVoVbS67tDao7XKchJvqd9q3WGiXikeELJyuTDqE\n"\
"zW22pTwMF+m3UEAxcxVCrhMvhkUzNAkANHaOatuFHzj7lyqhO5QPbh4J3FMR0X9X\n"\
"V8VWRSg+jA/SECP9koOl6zlzd5Tee0tW1pA7QpryXscs6IEhb3ns5R2JAoGAIkzO\n"\
"RmnhEOKTzDex611f2D+yMsMfy5BKK2f4vjLymBH5TiBKDXKqEpgsW0huoi8Gq9Uu\n"\
"nvvXXAgkIyRYF36f0vUe0nkjLuYAQAWgC2pZYgNLJR13iVbol0xHJoXQUHtgiaJ8\n"\
"GLYFzjHQPqFMpSalQe3oELko39uOC1CoJCHFySECgYBeycUnRBikCO2n8DNhY4Eg\n"\
"9Y3oxcssRt6ea5BZwgW2eAYi7/XqKkmxoSoOykUt3MJx9+EkkrL17bxFSpkj1tvL\n"\
"qvxn7egtsKjjgGNAxwXC4MwCvhveyUQQxtQb8AqGrGqo4jEEN0L15cnP38i2x1Uo\n"\
"muhfskWf4MABV0yTUaKcGg==\n"\
"-----END PRIVATE KEY-----\n"
server_cert = '-----BEGIN CERTIFICATE-----\n' \
'MIIDXTCCAkWgAwIBAgIJAP4LF7E72HakMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV\n'\
'BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX\n'\
'aWRnaXRzIFB0eSBMdGQwHhcNMTkwNjA3MDk1OTE2WhcNMjAwNjA2MDk1OTE2WjBF\n'\
'MQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50\n'\
'ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\n'\
'CgKCAQEAlzfCyv3mIv7TlLkObxunKfCdrJ/zgdANrsx0RBtpEPhV560hWJ0fEin0\n'\
'nIOMpJSiF9E6QsPdr6Q+eogH4XnOMU9JE+iG743N1dPfGEzJvRlyct/Ck8SswKPC\n'\
'9+VXsnOdZmUw9y/xtANbURA/TspvPzz3Avv382ffffrJGh7ooOmaZSCZFlSYHLZA\n'\
'w/XlRr0sSRbLpFGY0gXjaAV8iHHiPDYLy4kZOepjV9U51xi+IGsL4w75zuMgsHyF\n'\
'3nJeGYHgtGVBrkL0ZKG5udY0wcBjysjubDJC4iSlNiq2HD3fhs7j6CZddV2v845M\n'\
'lVKNxP0kO4Uj4D8r+5USWC8JKfAwxQIDAQABo1AwTjAdBgNVHQ4EFgQU6OE7ssfY\n'\
'IIPTDThiUoofUpsD5NwwHwYDVR0jBBgwFoAU6OE7ssfYIIPTDThiUoofUpsD5Nww\n'\
'DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAXIlHS/FJWfmcinUAxyBd\n'\
'/xd5Lu8ykeru6oaUCci+Vk9lyoMMES7lQ+b/00d5x7AcTawkTil9EWpBTPTOTraA\n'\
'lzJMQhNKmSLk0iIoTtAJtSZgUSpIIozqK6lenxQQDsHbXKU6h+u9H6KZE8YcjsFl\n'\
'6vL7sw9BVotw/VxfgjQ5OSGLgoLrdVT0z5C2qOuwOgz1c7jNiJhtMdwN+cOtnJp2\n'\
'fuBgEYyE3eeuWogvkWoDcIA8r17Ixzkpq2oJsdvZcHZPIZShPKW2SHUsl98KDemu\n'\
'y0pQyExmQUbwKE4vbFb9XuWCcL9XaOHQytyszt2DeD67AipvoBwVU7/LBOvqnsmy\n'\
'hA==\n'\
'-----END CERTIFICATE-----\n'
server_key = '-----BEGIN PRIVATE KEY-----\n'\
'MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCXN8LK/eYi/tOU\n'\
'uQ5vG6cp8J2sn/OB0A2uzHREG2kQ+FXnrSFYnR8SKfScg4yklKIX0TpCw92vpD56\n'\
'iAfhec4xT0kT6Ibvjc3V098YTMm9GXJy38KTxKzAo8L35Veyc51mZTD3L/G0A1tR\n'\
'ED9Oym8/PPcC+/fzZ999+skaHuig6ZplIJkWVJgctkDD9eVGvSxJFsukUZjSBeNo\n'\
'BXyIceI8NgvLiRk56mNX1TnXGL4gawvjDvnO4yCwfIXecl4ZgeC0ZUGuQvRkobm5\n'\
'1jTBwGPKyO5sMkLiJKU2KrYcPd+GzuPoJl11Xa/zjkyVUo3E/SQ7hSPgPyv7lRJY\n'\
'Lwkp8DDFAgMBAAECggEAfBhAfQE7mUByNbxgAgI5fot9eaqR1Nf+QpJ6X2H3KPwC\n'\
'02sa0HOwieFwYfj6tB1doBoNq7i89mTc+QUlIn4pHgIowHO0OGawomeKz5BEhjCZ\n'\
'4XeLYGSoODary2+kNkf2xY8JTfFEcyvGBpJEwc4S2VyYgRRx+IgnumTSH+N5mIKZ\n'\
'SXWNdZIuHEmkwod+rPRXs6/r+PH0eVW6WfpINEbr4zVAGXJx2zXQwd2cuV1GTJWh\n'\
'cPVOXLu+XJ9im9B370cYN6GqUnR3fui13urYbnWnEf3syvoH/zuZkyrVChauoFf8\n'\
'8EGb74/HhXK7Q2s8NRakx2c7OxQifCbcy03liUMmyQKBgQDFAob5B/66N4Q2cq/N\n'\
'MWPf98kYBYoLaeEOhEJhLQlKk0pIFCTmtpmUbpoEes2kCUbH7RwczpYko8tlKyoB\n'\
'6Fn6RY4zQQ64KZJI6kQVsjkYpcP/ihnOY6rbds+3yyv+4uPX7Eh9sYZwZMggE19M\n'\
'CkFHkwAjiwqhiiSlUxe20sWmowKBgQDEfx4lxuFzA1PBPeZKGVBTxYPQf+DSLCre\n'\
'ZFg3ZmrxbCjRq1O7Lra4FXWD3dmRq7NDk79JofoW50yD8wD7I0B7opdDfXD2idO8\n'\
'0dBnWUKDr2CAXyoLEINce9kJPbx4kFBQRN9PiGF7VkDQxeQ3kfS8CvcErpTKCOdy\n'\
'5wOwBTwJdwKBgDiTFTeGeDv5nVoVbS67tDao7XKchJvqd9q3WGiXikeELJyuTDqE\n'\
'zW22pTwMF+m3UEAxcxVCrhMvhkUzNAkANHaOatuFHzj7lyqhO5QPbh4J3FMR0X9X\n'\
'V8VWRSg+jA/SECP9koOl6zlzd5Tee0tW1pA7QpryXscs6IEhb3ns5R2JAoGAIkzO\n'\
'RmnhEOKTzDex611f2D+yMsMfy5BKK2f4vjLymBH5TiBKDXKqEpgsW0huoi8Gq9Uu\n'\
'nvvXXAgkIyRYF36f0vUe0nkjLuYAQAWgC2pZYgNLJR13iVbol0xHJoXQUHtgiaJ8\n'\
'GLYFzjHQPqFMpSalQe3oELko39uOC1CoJCHFySECgYBeycUnRBikCO2n8DNhY4Eg\n'\
'9Y3oxcssRt6ea5BZwgW2eAYi7/XqKkmxoSoOykUt3MJx9+EkkrL17bxFSpkj1tvL\n'\
'qvxn7egtsKjjgGNAxwXC4MwCvhveyUQQxtQb8AqGrGqo4jEEN0L15cnP38i2x1Uo\n'\
'muhfskWf4MABV0yTUaKcGg==\n'\
'-----END PRIVATE KEY-----\n'
def get_my_ip():
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect(("8.8.8.8", 80))
s1.connect(('8.8.8.8', 80))
my_ip = s1.getsockname()[0]
s1.close()
return my_ip
@@ -81,16 +81,16 @@ def get_server_status(host_ip, port):
def create_file(server_file, file_data):
with open(server_file, "w+") as file:
with open(server_file, 'w+') as file:
file.write(file_data)
def get_ca_cert(ota_image_dir):
os.chdir(ota_image_dir)
server_file = os.path.join(ota_image_dir, "server_cert.pem")
server_file = os.path.join(ota_image_dir, 'server_cert.pem')
create_file(server_file, server_cert)
key_file = os.path.join(ota_image_dir, "server_key.pem")
key_file = os.path.join(ota_image_dir, 'server_key.pem')
create_file(key_file, server_key)
return server_file, key_file
@@ -131,7 +131,7 @@ def start_https_server(ota_image_dir, server_ip, server_port):
def start_chunked_server(ota_image_dir, server_port):
server_file, key_file = get_ca_cert(ota_image_dir)
chunked_server = subprocess.Popen(["openssl", "s_server", "-WWW", "-key", key_file, "-cert", server_file, "-port", str(server_port)])
chunked_server = subprocess.Popen(['openssl', 's_server', '-WWW', '-key', key_file, '-cert', server_file, '-port', str(server_port)])
return chunked_server
@@ -141,7 +141,7 @@ def redirect_handler_factory(url):
"""
class RedirectHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self):
print("Sending resp, URL: " + url)
print('Sending resp, URL: ' + url)
self.send_response(301)
self.send_header('Location', url)
self.end_headers()
@@ -158,7 +158,7 @@ def redirect_handler_factory(url):
def start_redirect_server(ota_image_dir, server_ip, server_port, redirection_port):
os.chdir(ota_image_dir)
server_file, key_file = get_ca_cert(ota_image_dir)
redirectHandler = redirect_handler_factory("https://" + server_ip + ":" + str(redirection_port) + "/advanced_https_ota.bin")
redirectHandler = redirect_handler_factory('https://' + server_ip + ':' + str(redirection_port) + '/advanced_https_ota.bin')
httpd = http.server.HTTPServer((server_ip, server_port), redirectHandler)
@@ -168,7 +168,7 @@ def start_redirect_server(ota_image_dir, server_ip, server_port, redirection_por
httpd.serve_forever()
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_advanced_https_ota_example(env, extra_data):
"""
This is a positive test case, which downloads complete binary file multiple number of times.
@@ -178,16 +178,16 @@ def test_examples_protocol_advanced_https_ota_example(env, extra_data):
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
# Number of iterations to validate OTA
iterations = 3
server_port = 8001
# File to be downloaded. This file is generated after compilation
bin_name = "advanced_https_ota.bin"
bin_name = 'advanced_https_ota.bin'
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -196,23 +196,23 @@ def test_examples_protocol_advanced_https_ota_example(env, extra_data):
thread1.start()
dut1.start_app()
for i in range(iterations):
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
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)
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()
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_data):
"""
Working of OTA if binary file is truncated is validated in this test case.
@@ -223,25 +223,25 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_d
3. Fetch OTA image over HTTPS
4. Check working of code if bin is truncated
"""
dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001
# Original binary file generated after compilation
bin_name = "advanced_https_ota.bin"
bin_name = 'advanced_https_ota.bin'
# Truncated binary file to be generated from original binary file
truncated_bin_name = "truncated.bin"
truncated_bin_name = 'truncated.bin'
# Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
# truncated_bin_size is set to 64000 to reduce consumed by the test case
truncated_bin_size = 64000
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
f = open(binary_file, "rb+")
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+')
fo.write(f.read(truncated_bin_size))
fo.close()
f.close()
binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -249,21 +249,21 @@ def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_d
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
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)
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)
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extra_data):
"""
Working of OTA if headers of binary file are truncated is vaildated in this test case.
@@ -274,24 +274,24 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extr
3. Fetch OTA image over HTTPS
4. Check working of code if headers are not sent completely
"""
dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001
# Original binary file generated after compilation
bin_name = "advanced_https_ota.bin"
bin_name = 'advanced_https_ota.bin'
# Truncated binary file to be generated from original binary file
truncated_bin_name = "truncated_header.bin"
truncated_bin_name = 'truncated_header.bin'
# Size of truncated file to be grnerated. This value should be less than 288 bytes (Image header size)
truncated_bin_size = 180
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
f = open(binary_file, "rb+")
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+')
fo.write(f.read(truncated_bin_size))
fo.close()
f.close()
binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -299,21 +299,21 @@ def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extr
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
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)
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)
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
"""
Working of OTA if random data is added in binary file are validated in this test case.
@@ -324,23 +324,23 @@ def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
3. Fetch OTA image over HTTPS
4. Check working of code for random binary file
"""
dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001
# Random binary file to be generated
random_bin_name = "random.bin"
random_bin_name = 'random.bin'
# Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
random_bin_size = 32000
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, random_bin_name)
fo = open(binary_file, "wb+")
fo = open(binary_file, 'wb+')
# First byte of binary file is always set to zero. If first byte is generated randomly,
# in some cases it may generate 0xE9 which will result in failure of testcase.
fo.write(struct.pack("B", 0))
fo.write(struct.pack('B', 0))
for i in range(random_bin_size - 1):
fo.write(struct.pack("B", random.randrange(0,255,1)))
fo.write(struct.pack('B', random.randrange(0,255,1)))
fo.close()
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -348,21 +348,21 @@ def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
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)
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)
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_advanced_https_ota_example_chunked(env, extra_data):
"""
This is a positive test case, which downloads complete binary file multiple number of times.
@@ -372,35 +372,35 @@ def test_examples_protocol_advanced_https_ota_example_chunked(env, extra_data):
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
# File to be downloaded. This file is generated after compilation
bin_name = "advanced_https_ota.bin"
bin_name = 'advanced_https_ota.bin'
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
chunked_server = start_chunked_server(dut1.app.binary_path, 8070)
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
dut1.expect('Starting Advanced OTA example', timeout=30)
print("writing to device: {}".format("https://" + host_ip + ":8070/" + bin_name))
dut1.write("https://" + host_ip + ":8070/" + bin_name)
dut1.expect("Loaded app from partition at offset", timeout=60)
dut1.expect("Starting Advanced OTA example", timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
dut1.write('https://' + host_ip + ':8070/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting Advanced OTA example', timeout=30)
chunked_server.kill()
os.remove(os.path.join(dut1.app.binary_path, "server_cert.pem"))
os.remove(os.path.join(dut1.app.binary_path, "server_key.pem"))
os.remove(os.path.join(dut1.app.binary_path, 'server_cert.pem'))
os.remove(os.path.join(dut1.app.binary_path, 'server_key.pem'))
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_data):
"""
This is a positive test case, which starts a server and a redirection server.
@@ -411,16 +411,16 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_da
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
server_port = 8001
# Port to which the request should be redirecetd
redirection_server_port = 8081
# File to be downloaded. This file is generated after compilation
bin_name = "advanced_https_ota.bin"
bin_name = 'advanced_https_ota.bin'
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -431,24 +431,24 @@ def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_da
thread2.daemon = True
thread2.start()
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
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)
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()
@ttfw_idf.idf_example_test(env_tag="Example_8Mflash_Ethernet")
@ttfw_idf.idf_example_test(env_tag='Example_8Mflash_Ethernet')
def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_data):
"""
Working of OTA when anti_rollback is enabled and security version of new image is less than current one.
@@ -459,17 +459,17 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_d
3. Fetch OTA image over HTTPS
4. Check working of anti_rollback feature
"""
dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT, app_config_name='anti_rollback')
dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='anti_rollback')
server_port = 8001
# Original binary file generated after compilation
bin_name = "advanced_https_ota.bin"
bin_name = 'advanced_https_ota.bin'
# Modified firmware image to lower security version in its header. This is to enable negative test case
anti_rollback_bin_name = "advanced_https_ota_lower_sec_version.bin"
anti_rollback_bin_name = 'advanced_https_ota_lower_sec_version.bin'
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
file_size = os.path.getsize(binary_file)
f = open(binary_file, "rb+")
fo = open(os.path.join(dut1.app.binary_path, anti_rollback_bin_name), "wb+")
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut1.app.binary_path, anti_rollback_bin_name), 'wb+')
fo.write(f.read(file_size))
# Change security_version to 0 for negative test case
fo.seek(36)
@@ -478,7 +478,7 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_d
f.close()
binary_file = os.path.join(dut1.app.binary_path, anti_rollback_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -487,27 +487,27 @@ def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_d
thread1.start()
dut1.start_app()
# Positive Case
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" eth ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' eth ip: ([^,]+),'), timeout=30)
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)
dut1.expect('Starting Advanced OTA example', timeout=30)
# Use originally generated image with secure_version=1
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(re.compile(r" eth ip: ([^,]+),"), timeout=30)
dut1.expect("App is valid, rollback cancelled successfully", 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(re.compile(r' eth ip: ([^,]+),'), timeout=30)
dut1.expect('App is valid, rollback cancelled successfully', 30)
# Negative Case
dut1.expect("Starting Advanced OTA example", timeout=30)
dut1.expect('Starting Advanced OTA example', timeout=30)
# Use modified image with secure_version=0
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)
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)

262
examples/system/ota/native_ota_example/example_test.py
View File

@@ -1,71 +1,71 @@
import re
import os
import struct
import socket
import http.server
from threading import Thread
import ssl
from tiny_test_fw import DUT
import ttfw_idf
import os
import random
import re
import socket
import ssl
import struct
import subprocess
from threading import Thread
server_cert = "-----BEGIN CERTIFICATE-----\n" \
"MIIDXTCCAkWgAwIBAgIJAP4LF7E72HakMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV\n"\
"BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX\n"\
"aWRnaXRzIFB0eSBMdGQwHhcNMTkwNjA3MDk1OTE2WhcNMjAwNjA2MDk1OTE2WjBF\n"\
"MQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50\n"\
"ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\n"\
"CgKCAQEAlzfCyv3mIv7TlLkObxunKfCdrJ/zgdANrsx0RBtpEPhV560hWJ0fEin0\n"\
"nIOMpJSiF9E6QsPdr6Q+eogH4XnOMU9JE+iG743N1dPfGEzJvRlyct/Ck8SswKPC\n"\
"9+VXsnOdZmUw9y/xtANbURA/TspvPzz3Avv382ffffrJGh7ooOmaZSCZFlSYHLZA\n"\
"w/XlRr0sSRbLpFGY0gXjaAV8iHHiPDYLy4kZOepjV9U51xi+IGsL4w75zuMgsHyF\n"\
"3nJeGYHgtGVBrkL0ZKG5udY0wcBjysjubDJC4iSlNiq2HD3fhs7j6CZddV2v845M\n"\
"lVKNxP0kO4Uj4D8r+5USWC8JKfAwxQIDAQABo1AwTjAdBgNVHQ4EFgQU6OE7ssfY\n"\
"IIPTDThiUoofUpsD5NwwHwYDVR0jBBgwFoAU6OE7ssfYIIPTDThiUoofUpsD5Nww\n"\
"DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAXIlHS/FJWfmcinUAxyBd\n"\
"/xd5Lu8ykeru6oaUCci+Vk9lyoMMES7lQ+b/00d5x7AcTawkTil9EWpBTPTOTraA\n"\
"lzJMQhNKmSLk0iIoTtAJtSZgUSpIIozqK6lenxQQDsHbXKU6h+u9H6KZE8YcjsFl\n"\
"6vL7sw9BVotw/VxfgjQ5OSGLgoLrdVT0z5C2qOuwOgz1c7jNiJhtMdwN+cOtnJp2\n"\
"fuBgEYyE3eeuWogvkWoDcIA8r17Ixzkpq2oJsdvZcHZPIZShPKW2SHUsl98KDemu\n"\
"y0pQyExmQUbwKE4vbFb9XuWCcL9XaOHQytyszt2DeD67AipvoBwVU7/LBOvqnsmy\n"\
"hA==\n"\
"-----END CERTIFICATE-----\n"
import ttfw_idf
from tiny_test_fw import DUT
server_key = "-----BEGIN PRIVATE KEY-----\n"\
"MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCXN8LK/eYi/tOU\n"\
"uQ5vG6cp8J2sn/OB0A2uzHREG2kQ+FXnrSFYnR8SKfScg4yklKIX0TpCw92vpD56\n"\
"iAfhec4xT0kT6Ibvjc3V098YTMm9GXJy38KTxKzAo8L35Veyc51mZTD3L/G0A1tR\n"\
"ED9Oym8/PPcC+/fzZ999+skaHuig6ZplIJkWVJgctkDD9eVGvSxJFsukUZjSBeNo\n"\
"BXyIceI8NgvLiRk56mNX1TnXGL4gawvjDvnO4yCwfIXecl4ZgeC0ZUGuQvRkobm5\n"\
"1jTBwGPKyO5sMkLiJKU2KrYcPd+GzuPoJl11Xa/zjkyVUo3E/SQ7hSPgPyv7lRJY\n"\
"Lwkp8DDFAgMBAAECggEAfBhAfQE7mUByNbxgAgI5fot9eaqR1Nf+QpJ6X2H3KPwC\n"\
"02sa0HOwieFwYfj6tB1doBoNq7i89mTc+QUlIn4pHgIowHO0OGawomeKz5BEhjCZ\n"\
"4XeLYGSoODary2+kNkf2xY8JTfFEcyvGBpJEwc4S2VyYgRRx+IgnumTSH+N5mIKZ\n"\
"SXWNdZIuHEmkwod+rPRXs6/r+PH0eVW6WfpINEbr4zVAGXJx2zXQwd2cuV1GTJWh\n"\
"cPVOXLu+XJ9im9B370cYN6GqUnR3fui13urYbnWnEf3syvoH/zuZkyrVChauoFf8\n"\
"8EGb74/HhXK7Q2s8NRakx2c7OxQifCbcy03liUMmyQKBgQDFAob5B/66N4Q2cq/N\n"\
"MWPf98kYBYoLaeEOhEJhLQlKk0pIFCTmtpmUbpoEes2kCUbH7RwczpYko8tlKyoB\n"\
"6Fn6RY4zQQ64KZJI6kQVsjkYpcP/ihnOY6rbds+3yyv+4uPX7Eh9sYZwZMggE19M\n"\
"CkFHkwAjiwqhiiSlUxe20sWmowKBgQDEfx4lxuFzA1PBPeZKGVBTxYPQf+DSLCre\n"\
"ZFg3ZmrxbCjRq1O7Lra4FXWD3dmRq7NDk79JofoW50yD8wD7I0B7opdDfXD2idO8\n"\
"0dBnWUKDr2CAXyoLEINce9kJPbx4kFBQRN9PiGF7VkDQxeQ3kfS8CvcErpTKCOdy\n"\
"5wOwBTwJdwKBgDiTFTeGeDv5nVoVbS67tDao7XKchJvqd9q3WGiXikeELJyuTDqE\n"\
"zW22pTwMF+m3UEAxcxVCrhMvhkUzNAkANHaOatuFHzj7lyqhO5QPbh4J3FMR0X9X\n"\
"V8VWRSg+jA/SECP9koOl6zlzd5Tee0tW1pA7QpryXscs6IEhb3ns5R2JAoGAIkzO\n"\
"RmnhEOKTzDex611f2D+yMsMfy5BKK2f4vjLymBH5TiBKDXKqEpgsW0huoi8Gq9Uu\n"\
"nvvXXAgkIyRYF36f0vUe0nkjLuYAQAWgC2pZYgNLJR13iVbol0xHJoXQUHtgiaJ8\n"\
"GLYFzjHQPqFMpSalQe3oELko39uOC1CoJCHFySECgYBeycUnRBikCO2n8DNhY4Eg\n"\
"9Y3oxcssRt6ea5BZwgW2eAYi7/XqKkmxoSoOykUt3MJx9+EkkrL17bxFSpkj1tvL\n"\
"qvxn7egtsKjjgGNAxwXC4MwCvhveyUQQxtQb8AqGrGqo4jEEN0L15cnP38i2x1Uo\n"\
"muhfskWf4MABV0yTUaKcGg==\n"\
"-----END PRIVATE KEY-----\n"
server_cert = '-----BEGIN CERTIFICATE-----\n' \
'MIIDXTCCAkWgAwIBAgIJAP4LF7E72HakMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV\n'\
'BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX\n'\
'aWRnaXRzIFB0eSBMdGQwHhcNMTkwNjA3MDk1OTE2WhcNMjAwNjA2MDk1OTE2WjBF\n'\
'MQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50\n'\
'ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\n'\
'CgKCAQEAlzfCyv3mIv7TlLkObxunKfCdrJ/zgdANrsx0RBtpEPhV560hWJ0fEin0\n'\
'nIOMpJSiF9E6QsPdr6Q+eogH4XnOMU9JE+iG743N1dPfGEzJvRlyct/Ck8SswKPC\n'\
'9+VXsnOdZmUw9y/xtANbURA/TspvPzz3Avv382ffffrJGh7ooOmaZSCZFlSYHLZA\n'\
'w/XlRr0sSRbLpFGY0gXjaAV8iHHiPDYLy4kZOepjV9U51xi+IGsL4w75zuMgsHyF\n'\
'3nJeGYHgtGVBrkL0ZKG5udY0wcBjysjubDJC4iSlNiq2HD3fhs7j6CZddV2v845M\n'\
'lVKNxP0kO4Uj4D8r+5USWC8JKfAwxQIDAQABo1AwTjAdBgNVHQ4EFgQU6OE7ssfY\n'\
'IIPTDThiUoofUpsD5NwwHwYDVR0jBBgwFoAU6OE7ssfYIIPTDThiUoofUpsD5Nww\n'\
'DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAXIlHS/FJWfmcinUAxyBd\n'\
'/xd5Lu8ykeru6oaUCci+Vk9lyoMMES7lQ+b/00d5x7AcTawkTil9EWpBTPTOTraA\n'\
'lzJMQhNKmSLk0iIoTtAJtSZgUSpIIozqK6lenxQQDsHbXKU6h+u9H6KZE8YcjsFl\n'\
'6vL7sw9BVotw/VxfgjQ5OSGLgoLrdVT0z5C2qOuwOgz1c7jNiJhtMdwN+cOtnJp2\n'\
'fuBgEYyE3eeuWogvkWoDcIA8r17Ixzkpq2oJsdvZcHZPIZShPKW2SHUsl98KDemu\n'\
'y0pQyExmQUbwKE4vbFb9XuWCcL9XaOHQytyszt2DeD67AipvoBwVU7/LBOvqnsmy\n'\
'hA==\n'\
'-----END CERTIFICATE-----\n'
server_key = '-----BEGIN PRIVATE KEY-----\n'\
'MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCXN8LK/eYi/tOU\n'\
'uQ5vG6cp8J2sn/OB0A2uzHREG2kQ+FXnrSFYnR8SKfScg4yklKIX0TpCw92vpD56\n'\
'iAfhec4xT0kT6Ibvjc3V098YTMm9GXJy38KTxKzAo8L35Veyc51mZTD3L/G0A1tR\n'\
'ED9Oym8/PPcC+/fzZ999+skaHuig6ZplIJkWVJgctkDD9eVGvSxJFsukUZjSBeNo\n'\
'BXyIceI8NgvLiRk56mNX1TnXGL4gawvjDvnO4yCwfIXecl4ZgeC0ZUGuQvRkobm5\n'\
'1jTBwGPKyO5sMkLiJKU2KrYcPd+GzuPoJl11Xa/zjkyVUo3E/SQ7hSPgPyv7lRJY\n'\
'Lwkp8DDFAgMBAAECggEAfBhAfQE7mUByNbxgAgI5fot9eaqR1Nf+QpJ6X2H3KPwC\n'\
'02sa0HOwieFwYfj6tB1doBoNq7i89mTc+QUlIn4pHgIowHO0OGawomeKz5BEhjCZ\n'\
'4XeLYGSoODary2+kNkf2xY8JTfFEcyvGBpJEwc4S2VyYgRRx+IgnumTSH+N5mIKZ\n'\
'SXWNdZIuHEmkwod+rPRXs6/r+PH0eVW6WfpINEbr4zVAGXJx2zXQwd2cuV1GTJWh\n'\
'cPVOXLu+XJ9im9B370cYN6GqUnR3fui13urYbnWnEf3syvoH/zuZkyrVChauoFf8\n'\
'8EGb74/HhXK7Q2s8NRakx2c7OxQifCbcy03liUMmyQKBgQDFAob5B/66N4Q2cq/N\n'\
'MWPf98kYBYoLaeEOhEJhLQlKk0pIFCTmtpmUbpoEes2kCUbH7RwczpYko8tlKyoB\n'\
'6Fn6RY4zQQ64KZJI6kQVsjkYpcP/ihnOY6rbds+3yyv+4uPX7Eh9sYZwZMggE19M\n'\
'CkFHkwAjiwqhiiSlUxe20sWmowKBgQDEfx4lxuFzA1PBPeZKGVBTxYPQf+DSLCre\n'\
'ZFg3ZmrxbCjRq1O7Lra4FXWD3dmRq7NDk79JofoW50yD8wD7I0B7opdDfXD2idO8\n'\
'0dBnWUKDr2CAXyoLEINce9kJPbx4kFBQRN9PiGF7VkDQxeQ3kfS8CvcErpTKCOdy\n'\
'5wOwBTwJdwKBgDiTFTeGeDv5nVoVbS67tDao7XKchJvqd9q3WGiXikeELJyuTDqE\n'\
'zW22pTwMF+m3UEAxcxVCrhMvhkUzNAkANHaOatuFHzj7lyqhO5QPbh4J3FMR0X9X\n'\
'V8VWRSg+jA/SECP9koOl6zlzd5Tee0tW1pA7QpryXscs6IEhb3ns5R2JAoGAIkzO\n'\
'RmnhEOKTzDex611f2D+yMsMfy5BKK2f4vjLymBH5TiBKDXKqEpgsW0huoi8Gq9Uu\n'\
'nvvXXAgkIyRYF36f0vUe0nkjLuYAQAWgC2pZYgNLJR13iVbol0xHJoXQUHtgiaJ8\n'\
'GLYFzjHQPqFMpSalQe3oELko39uOC1CoJCHFySECgYBeycUnRBikCO2n8DNhY4Eg\n'\
'9Y3oxcssRt6ea5BZwgW2eAYi7/XqKkmxoSoOykUt3MJx9+EkkrL17bxFSpkj1tvL\n'\
'qvxn7egtsKjjgGNAxwXC4MwCvhveyUQQxtQb8AqGrGqo4jEEN0L15cnP38i2x1Uo\n'\
'muhfskWf4MABV0yTUaKcGg==\n'\
'-----END PRIVATE KEY-----\n'
def get_my_ip():
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect(("8.8.8.8", 80))
s1.connect(('8.8.8.8', 80))
my_ip = s1.getsockname()[0]
s1.close()
return my_ip
@@ -81,16 +81,16 @@ def get_server_status(host_ip, port):
def create_file(server_file, file_data):
with open(server_file, "w+") as file:
with open(server_file, 'w+') as file:
file.write(file_data)
def get_ca_cert(ota_image_dir):
os.chdir(ota_image_dir)
server_file = os.path.join(ota_image_dir, "server_cert.pem")
server_file = os.path.join(ota_image_dir, 'server_cert.pem')
create_file(server_file, server_cert)
key_file = os.path.join(ota_image_dir, "server_key.pem")
key_file = os.path.join(ota_image_dir, 'server_key.pem')
create_file(key_file, server_key)
return server_file, key_file
@@ -131,11 +131,11 @@ def start_https_server(ota_image_dir, server_ip, server_port):
def start_chunked_server(ota_image_dir, server_port):
server_file, key_file = get_ca_cert(ota_image_dir)
chunked_server = subprocess.Popen(["openssl", "s_server", "-WWW", "-key", key_file, "-cert", server_file, "-port", str(server_port)])
chunked_server = subprocess.Popen(['openssl', 's_server', '-WWW', '-key', key_file, '-cert', server_file, '-port', str(server_port)])
return chunked_server
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_native_ota_example(env, extra_data):
"""
This is a positive test case, which downloads complete binary file multiple number of times.
@@ -145,16 +145,16 @@ def test_examples_protocol_native_ota_example(env, extra_data):
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut("native_ota_example", "examples/system/ota/native_ota_example", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
server_port = 8002
# No. of times working of application to be validated
iterations = 3
# File to be downloaded. This file is generated after compilation
bin_name = "native_ota.bin"
bin_name = 'native_ota.bin'
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("native_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -163,23 +163,23 @@ def test_examples_protocol_native_ota_example(env, extra_data):
thread1.start()
dut1.start_app()
for i in range(iterations):
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
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)
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()
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data):
"""
Working of OTA if binary file is truncated is validated in this test case.
@@ -190,25 +190,25 @@ def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data):
3. Fetch OTA image over HTTPS
4. Check working of code if bin is truncated
"""
dut1 = env.get_dut("native_ota_example", "examples/system/ota/native_ota_example", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
server_port = 8002
# Original binary file generated after compilation
bin_name = "native_ota.bin"
bin_name = 'native_ota.bin'
# Truncated binary file to be generated from original binary file
truncated_bin_name = "truncated.bin"
truncated_bin_name = 'truncated.bin'
# Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
# truncated_bin_size is set to 64000 to reduce consumed by the test case
truncated_bin_size = 64000
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
f = open(binary_file, "rb+")
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+')
fo.write(f.read(truncated_bin_size))
fo.close()
f.close()
binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("native_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -216,21 +216,21 @@ def test_examples_protocol_native_ota_example_truncated_bin(env, extra_data):
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=60)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=60)
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)
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)
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)
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_native_ota_example_truncated_header(env, extra_data):
"""
Working of OTA if headers of binary file are truncated is vaildated in this test case.
@@ -241,24 +241,24 @@ def test_examples_protocol_native_ota_example_truncated_header(env, extra_data):
3. Fetch OTA image over HTTPS
4. Check working of code if headers are not sent completely
"""
dut1 = env.get_dut("native_ota_example", "examples/system/ota/native_ota_example", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
server_port = 8002
# Original binary file generated after compilation
bin_name = "native_ota.bin"
bin_name = 'native_ota.bin'
# Truncated binary file to be generated from original binary file
truncated_bin_name = "truncated_header.bin"
truncated_bin_name = 'truncated_header.bin'
# Size of truncated file to be grnerated. This value should be less than 288 bytes (Image header size)
truncated_bin_size = 180
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
f = open(binary_file, "rb+")
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
f = open(binary_file, 'rb+')
fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+')
fo.write(f.read(truncated_bin_size))
fo.close()
f.close()
binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("native_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -266,21 +266,21 @@ def test_examples_protocol_native_ota_example_truncated_header(env, extra_data):
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=60)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=60)
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)
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)
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)
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_native_ota_example_random(env, extra_data):
"""
Working of OTA if random data is added in binary file are validated in this test case.
@@ -291,23 +291,23 @@ def test_examples_protocol_native_ota_example_random(env, extra_data):
3. Fetch OTA image over HTTPS
4. Check working of code for random binary file
"""
dut1 = env.get_dut("native_ota_example", "examples/system/ota/native_ota_example", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
server_port = 8002
# Random binary file to be generated
random_bin_name = "random.bin"
random_bin_name = 'random.bin'
# Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
random_bin_size = 32000
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, random_bin_name)
fo = open(binary_file, "wb+")
fo = open(binary_file, 'wb+')
# First byte of binary file is always set to zero. If first byte is generated randomly,
# in some cases it may generate 0xE9 which will result in failure of testcase.
fo.write(struct.pack("B", 0))
fo.write(struct.pack('B', 0))
for i in range(random_bin_size - 1):
fo.write(struct.pack("B", random.randrange(0,255,1)))
fo.write(struct.pack('B', random.randrange(0,255,1)))
fo.close()
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("native_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
if (get_server_status(host_ip, server_port) is False):
@@ -315,21 +315,21 @@ def test_examples_protocol_native_ota_example_random(env, extra_data):
thread1.daemon = True
thread1.start()
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=60)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=60)
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)
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)
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)
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_native_ota_example_chunked(env, extra_data):
"""
This is a positive test case, which downloads complete binary file multiple number of times.
@@ -339,32 +339,32 @@ def test_examples_protocol_native_ota_example_chunked(env, extra_data):
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut("native_ota_example", "examples/system/ota/native_ota_example", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('native_ota_example', 'examples/system/ota/native_ota_example', dut_class=ttfw_idf.ESP32DUT)
# File to be downloaded. This file is generated after compilation
bin_name = "native_ota.bin"
bin_name = 'native_ota.bin'
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, bin_name)
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("native_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('native_ota_bin_size', '{}KB'.format(bin_size // 1024))
# start test
host_ip = get_my_ip()
chunked_server = start_chunked_server(dut1.app.binary_path, 8070)
dut1.start_app()
dut1.expect("Loaded app from partition at offset", timeout=30)
dut1.expect('Loaded app from partition at offset', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
print("writing to device: {}".format("https://" + host_ip + ":8070/" + bin_name))
dut1.write("https://" + host_ip + ":8070/" + bin_name)
dut1.expect("Loaded app from partition at offset", timeout=60)
dut1.expect("Starting OTA example", timeout=30)
dut1.expect('Starting OTA example', timeout=30)
print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
dut1.write('https://' + host_ip + ':8070/' + bin_name)
dut1.expect('Loaded app from partition at offset', timeout=60)
dut1.expect('Starting OTA example', timeout=30)
chunked_server.kill()
os.remove(os.path.join(dut1.app.binary_path, "server_cert.pem"))
os.remove(os.path.join(dut1.app.binary_path, "server_key.pem"))
os.remove(os.path.join(dut1.app.binary_path, 'server_cert.pem'))
os.remove(os.path.join(dut1.app.binary_path, 'server_key.pem'))
if __name__ == '__main__':

15
examples/system/ota/otatool/example_test.py
View File

@@ -1,7 +1,8 @@
from __future__ import print_function
import os
import sys
import subprocess
import sys
import ttfw_idf
@@ -12,22 +13,22 @@ def test_otatool_example(env, extra_data):
# Verify factory firmware
dut.start_app()
dut.expect("OTA Tool Example")
dut.expect("Example end")
dut.expect('OTA Tool Example')
dut.expect('Example end')
# Close connection to DUT
dut.receive_thread.exit()
dut.port_inst.close()
script_path = os.path.join(os.getenv("IDF_PATH"), "examples", "system", "ota", "otatool", "otatool_example.py")
binary_path = ""
script_path = os.path.join(os.getenv('IDF_PATH'), 'examples', 'system', 'ota', 'otatool', 'otatool_example.py')
binary_path = ''
for flash_file in dut.app.flash_files:
if "otatool.bin" in flash_file[1]:
if 'otatool.bin' in flash_file[1]:
binary_path = flash_file[1]
break
subprocess.check_call([sys.executable, script_path, "--binary", binary_path])
subprocess.check_call([sys.executable, script_path, '--binary', binary_path])
if __name__ == '__main__':

36
examples/system/ota/otatool/get_running_partition.py
View File

@@ -16,34 +16,34 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import sys
import serial
import subprocess
import re
import argparse
import os
import re
import subprocess
import sys
from subprocess import CalledProcessError
import serial
def get_running_partition(port=None):
# Monitor the serial output of target device. The firmware outputs the currently
# running partition
IDF_PATH = os.path.expandvars("$IDF_PATH")
IDF_PATH = os.path.expandvars('$IDF_PATH')
sys.path.append(os.path.join(IDF_PATH, 'components', 'esptool_py', 'esptool'))
import esptool
ESPTOOL_PY = os.path.join(IDF_PATH, "components", "esptool_py", "esptool", "esptool.py")
ESPTOOL_PY = os.path.join(IDF_PATH, 'components', 'esptool_py', 'esptool', 'esptool.py')
baud = os.environ.get("ESPTOOL_BAUD", esptool.ESPLoader.ESP_ROM_BAUD)
baud = os.environ.get('ESPTOOL_BAUD', esptool.ESPLoader.ESP_ROM_BAUD)
if not port:
error_message = "Unable to obtain default target device port.\nSerial log:\n\n"
error_message = 'Unable to obtain default target device port.\nSerial log:\n\n'
try:
# Check what esptool.py finds on what port the device is connected to
output = subprocess.check_output([sys.executable, ESPTOOL_PY, "chip_id"]) # may raise CalledProcessError
pattern = r"Serial port ([\S]+)"
output = subprocess.check_output([sys.executable, ESPTOOL_PY, 'chip_id']) # may raise CalledProcessError
pattern = r'Serial port ([\S]+)'
pattern = re.compile(pattern.encode())
port = re.search(pattern, output).group(1) # may raise AttributeError
@@ -52,7 +52,7 @@ def get_running_partition(port=None):
except AttributeError:
raise Exception(error_message + output)
serial_instance = serial.serial_for_url(port.decode("utf-8"), baud, do_not_open=True)
serial_instance = serial.serial_for_url(port.decode('utf-8'), baud, do_not_open=True)
serial_instance.dtr = False
serial_instance.rts = False
@@ -62,16 +62,16 @@ def get_running_partition(port=None):
serial_instance.rts = False
# Read until example end and find the currently running partition string
content = serial_instance.read_until(b"Example end")
pattern = re.compile(b"Running partition: ([a-z0-9_]+)")
content = serial_instance.read_until(b'Example end')
pattern = re.compile(b'Running partition: ([a-z0-9_]+)')
running = re.search(pattern, content).group(1)
return running.decode("utf-8")
return running.decode('utf-8')
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--port", default=None)
parser.add_argument('--port', default=None)
args = parser.parse_args()
try:
@@ -83,5 +83,5 @@ def main():
print(res)
if __name__ == "__main__":
if __name__ == '__main__':
main()

70
examples/system/ota/otatool/otatool_example.py
View File

@@ -16,16 +16,16 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import os
import sys
import argparse
from get_running_partition import get_running_partition
def assert_file_same(file1, file2, err):
with open(file1, "rb") as f1:
with open(file2, "rb") as f2:
with open(file1, 'rb') as f1:
with open(file2, 'rb') as f2:
f1 = f1.read()
f2 = f2.read()
@@ -41,75 +41,75 @@ def assert_file_same(file1, file2, err):
def assert_running_partition(expected, port=None):
running = get_running_partition(port)
if running != expected:
raise Exception("Running partition %s does not match expected %s" % (running, expected))
raise Exception('Running partition %s does not match expected %s' % (running, expected))
def main():
COMPONENTS_PATH = os.path.expandvars(os.path.join("$IDF_PATH", "components"))
OTATOOL_DIR = os.path.join(COMPONENTS_PATH, "app_update")
COMPONENTS_PATH = os.path.expandvars(os.path.join('$IDF_PATH', 'components'))
OTATOOL_DIR = os.path.join(COMPONENTS_PATH, 'app_update')
sys.path.append(OTATOOL_DIR)
from otatool import OtatoolTarget
parser = argparse.ArgumentParser("ESP-IDF OTA Tool Example")
parser = argparse.ArgumentParser('ESP-IDF OTA Tool Example')
parser.add_argument("--port", "-p", help="port where the device to perform operations on is connected")
parser.add_argument("--binary", "-b", help="path to built example binary", default=os.path.join("build", "otatool.bin"))
parser.add_argument('--port', '-p', help='port where the device to perform operations on is connected')
parser.add_argument('--binary', '-b', help='path to built example binary', default=os.path.join('build', 'otatool.bin'))
args = parser.parse_args()
target = OtatoolTarget(args.port)
print("Writing factory firmware to ota_0")
print('Writing factory firmware to ota_0')
target.write_ota_partition(0, args.binary)
print("Writing factory firmware to ota_1")
target.write_ota_partition("ota_1", args.binary)
print('Writing factory firmware to ota_1')
target.write_ota_partition('ota_1', args.binary)
# Verify that the contents of the two ota slots are the same as that of the factory partition
print("Checking written firmware to ota_0 and ota_1 match factory firmware")
target.read_ota_partition("ota_0", "app0.bin")
target.read_ota_partition(1, "app1.bin")
print('Checking written firmware to ota_0 and ota_1 match factory firmware')
target.read_ota_partition('ota_0', 'app0.bin')
target.read_ota_partition(1, 'app1.bin')
assert_file_same("app0.bin", args.binary, "Slot 0 app does not match factory app")
assert_file_same("app1.bin", args.binary, "Slot 1 app does not match factory app")
assert_file_same('app0.bin', args.binary, 'Slot 0 app does not match factory app')
assert_file_same('app1.bin', args.binary, 'Slot 1 app does not match factory app')
# Switch to factory app
print("Switching to factory app")
print('Switching to factory app')
target.erase_otadata()
assert_running_partition("factory")
assert_running_partition('factory')
# Switch to slot 0
print("Switching to OTA slot 0")
print('Switching to OTA slot 0')
target.switch_ota_partition(0)
assert_running_partition("ota_0")
assert_running_partition('ota_0')
# Switch to slot 1 twice in a row
print("Switching to OTA slot 1 (twice in a row)")
print('Switching to OTA slot 1 (twice in a row)')
target.switch_ota_partition(1)
assert_running_partition("ota_1")
target.switch_ota_partition("ota_1")
assert_running_partition("ota_1")
assert_running_partition('ota_1')
target.switch_ota_partition('ota_1')
assert_running_partition('ota_1')
# Switch to slot 0 twice in a row
print("Switching to OTA slot 0 (twice in a row)")
print('Switching to OTA slot 0 (twice in a row)')
target.switch_ota_partition(0)
assert_running_partition("ota_0")
target.switch_ota_partition("ota_0")
assert_running_partition("ota_0")
assert_running_partition('ota_0')
target.switch_ota_partition('ota_0')
assert_running_partition('ota_0')
# Switch to factory app
print("Switching to factory app")
print('Switching to factory app')
target.erase_otadata()
assert_running_partition("factory")
assert_running_partition('factory')
# Switch to slot 1
print("Switching to OTA slot 1")
print('Switching to OTA slot 1')
target.switch_ota_partition(1)
assert_running_partition("ota_1")
assert_running_partition('ota_1')
# Example end and cleanup
print("\nOTA tool operations executed successfully!")
clean_files = ["app0.bin", "app1.bin"]
print('\nOTA tool operations executed successfully!')
clean_files = ['app0.bin', 'app1.bin']
for clean_file in clean_files:
os.unlink(clean_file)

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

@@ -1,68 +1,68 @@
import re
import os
import socket
import http.server
from threading import Thread
import os
import re
import socket
import ssl
from threading import Thread
from tiny_test_fw import DUT
import ttfw_idf
from tiny_test_fw import DUT
server_cert = "-----BEGIN CERTIFICATE-----\n" \
"MIIDXTCCAkWgAwIBAgIJAP4LF7E72HakMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV\n"\
"BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX\n"\
"aWRnaXRzIFB0eSBMdGQwHhcNMTkwNjA3MDk1OTE2WhcNMjAwNjA2MDk1OTE2WjBF\n"\
"MQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50\n"\
"ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\n"\
"CgKCAQEAlzfCyv3mIv7TlLkObxunKfCdrJ/zgdANrsx0RBtpEPhV560hWJ0fEin0\n"\
"nIOMpJSiF9E6QsPdr6Q+eogH4XnOMU9JE+iG743N1dPfGEzJvRlyct/Ck8SswKPC\n"\
"9+VXsnOdZmUw9y/xtANbURA/TspvPzz3Avv382ffffrJGh7ooOmaZSCZFlSYHLZA\n"\
"w/XlRr0sSRbLpFGY0gXjaAV8iHHiPDYLy4kZOepjV9U51xi+IGsL4w75zuMgsHyF\n"\
"3nJeGYHgtGVBrkL0ZKG5udY0wcBjysjubDJC4iSlNiq2HD3fhs7j6CZddV2v845M\n"\
"lVKNxP0kO4Uj4D8r+5USWC8JKfAwxQIDAQABo1AwTjAdBgNVHQ4EFgQU6OE7ssfY\n"\
"IIPTDThiUoofUpsD5NwwHwYDVR0jBBgwFoAU6OE7ssfYIIPTDThiUoofUpsD5Nww\n"\
"DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAXIlHS/FJWfmcinUAxyBd\n"\
"/xd5Lu8ykeru6oaUCci+Vk9lyoMMES7lQ+b/00d5x7AcTawkTil9EWpBTPTOTraA\n"\
"lzJMQhNKmSLk0iIoTtAJtSZgUSpIIozqK6lenxQQDsHbXKU6h+u9H6KZE8YcjsFl\n"\
"6vL7sw9BVotw/VxfgjQ5OSGLgoLrdVT0z5C2qOuwOgz1c7jNiJhtMdwN+cOtnJp2\n"\
"fuBgEYyE3eeuWogvkWoDcIA8r17Ixzkpq2oJsdvZcHZPIZShPKW2SHUsl98KDemu\n"\
"y0pQyExmQUbwKE4vbFb9XuWCcL9XaOHQytyszt2DeD67AipvoBwVU7/LBOvqnsmy\n"\
"hA==\n"\
"-----END CERTIFICATE-----\n"
server_cert = '-----BEGIN CERTIFICATE-----\n' \
'MIIDXTCCAkWgAwIBAgIJAP4LF7E72HakMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV\n'\
'BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX\n'\
'aWRnaXRzIFB0eSBMdGQwHhcNMTkwNjA3MDk1OTE2WhcNMjAwNjA2MDk1OTE2WjBF\n'\
'MQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50\n'\
'ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\n'\
'CgKCAQEAlzfCyv3mIv7TlLkObxunKfCdrJ/zgdANrsx0RBtpEPhV560hWJ0fEin0\n'\
'nIOMpJSiF9E6QsPdr6Q+eogH4XnOMU9JE+iG743N1dPfGEzJvRlyct/Ck8SswKPC\n'\
'9+VXsnOdZmUw9y/xtANbURA/TspvPzz3Avv382ffffrJGh7ooOmaZSCZFlSYHLZA\n'\
'w/XlRr0sSRbLpFGY0gXjaAV8iHHiPDYLy4kZOepjV9U51xi+IGsL4w75zuMgsHyF\n'\
'3nJeGYHgtGVBrkL0ZKG5udY0wcBjysjubDJC4iSlNiq2HD3fhs7j6CZddV2v845M\n'\
'lVKNxP0kO4Uj4D8r+5USWC8JKfAwxQIDAQABo1AwTjAdBgNVHQ4EFgQU6OE7ssfY\n'\
'IIPTDThiUoofUpsD5NwwHwYDVR0jBBgwFoAU6OE7ssfYIIPTDThiUoofUpsD5Nww\n'\
'DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAXIlHS/FJWfmcinUAxyBd\n'\
'/xd5Lu8ykeru6oaUCci+Vk9lyoMMES7lQ+b/00d5x7AcTawkTil9EWpBTPTOTraA\n'\
'lzJMQhNKmSLk0iIoTtAJtSZgUSpIIozqK6lenxQQDsHbXKU6h+u9H6KZE8YcjsFl\n'\
'6vL7sw9BVotw/VxfgjQ5OSGLgoLrdVT0z5C2qOuwOgz1c7jNiJhtMdwN+cOtnJp2\n'\
'fuBgEYyE3eeuWogvkWoDcIA8r17Ixzkpq2oJsdvZcHZPIZShPKW2SHUsl98KDemu\n'\
'y0pQyExmQUbwKE4vbFb9XuWCcL9XaOHQytyszt2DeD67AipvoBwVU7/LBOvqnsmy\n'\
'hA==\n'\
'-----END CERTIFICATE-----\n'
server_key = "-----BEGIN PRIVATE KEY-----\n"\
"MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCXN8LK/eYi/tOU\n"\
"uQ5vG6cp8J2sn/OB0A2uzHREG2kQ+FXnrSFYnR8SKfScg4yklKIX0TpCw92vpD56\n"\
"iAfhec4xT0kT6Ibvjc3V098YTMm9GXJy38KTxKzAo8L35Veyc51mZTD3L/G0A1tR\n"\
"ED9Oym8/PPcC+/fzZ999+skaHuig6ZplIJkWVJgctkDD9eVGvSxJFsukUZjSBeNo\n"\
"BXyIceI8NgvLiRk56mNX1TnXGL4gawvjDvnO4yCwfIXecl4ZgeC0ZUGuQvRkobm5\n"\
"1jTBwGPKyO5sMkLiJKU2KrYcPd+GzuPoJl11Xa/zjkyVUo3E/SQ7hSPgPyv7lRJY\n"\
"Lwkp8DDFAgMBAAECggEAfBhAfQE7mUByNbxgAgI5fot9eaqR1Nf+QpJ6X2H3KPwC\n"\
"02sa0HOwieFwYfj6tB1doBoNq7i89mTc+QUlIn4pHgIowHO0OGawomeKz5BEhjCZ\n"\
"4XeLYGSoODary2+kNkf2xY8JTfFEcyvGBpJEwc4S2VyYgRRx+IgnumTSH+N5mIKZ\n"\
"SXWNdZIuHEmkwod+rPRXs6/r+PH0eVW6WfpINEbr4zVAGXJx2zXQwd2cuV1GTJWh\n"\
"cPVOXLu+XJ9im9B370cYN6GqUnR3fui13urYbnWnEf3syvoH/zuZkyrVChauoFf8\n"\
"8EGb74/HhXK7Q2s8NRakx2c7OxQifCbcy03liUMmyQKBgQDFAob5B/66N4Q2cq/N\n"\
"MWPf98kYBYoLaeEOhEJhLQlKk0pIFCTmtpmUbpoEes2kCUbH7RwczpYko8tlKyoB\n"\
"6Fn6RY4zQQ64KZJI6kQVsjkYpcP/ihnOY6rbds+3yyv+4uPX7Eh9sYZwZMggE19M\n"\
"CkFHkwAjiwqhiiSlUxe20sWmowKBgQDEfx4lxuFzA1PBPeZKGVBTxYPQf+DSLCre\n"\
"ZFg3ZmrxbCjRq1O7Lra4FXWD3dmRq7NDk79JofoW50yD8wD7I0B7opdDfXD2idO8\n"\
"0dBnWUKDr2CAXyoLEINce9kJPbx4kFBQRN9PiGF7VkDQxeQ3kfS8CvcErpTKCOdy\n"\
"5wOwBTwJdwKBgDiTFTeGeDv5nVoVbS67tDao7XKchJvqd9q3WGiXikeELJyuTDqE\n"\
"zW22pTwMF+m3UEAxcxVCrhMvhkUzNAkANHaOatuFHzj7lyqhO5QPbh4J3FMR0X9X\n"\
"V8VWRSg+jA/SECP9koOl6zlzd5Tee0tW1pA7QpryXscs6IEhb3ns5R2JAoGAIkzO\n"\
"RmnhEOKTzDex611f2D+yMsMfy5BKK2f4vjLymBH5TiBKDXKqEpgsW0huoi8Gq9Uu\n"\
"nvvXXAgkIyRYF36f0vUe0nkjLuYAQAWgC2pZYgNLJR13iVbol0xHJoXQUHtgiaJ8\n"\
"GLYFzjHQPqFMpSalQe3oELko39uOC1CoJCHFySECgYBeycUnRBikCO2n8DNhY4Eg\n"\
"9Y3oxcssRt6ea5BZwgW2eAYi7/XqKkmxoSoOykUt3MJx9+EkkrL17bxFSpkj1tvL\n"\
"qvxn7egtsKjjgGNAxwXC4MwCvhveyUQQxtQb8AqGrGqo4jEEN0L15cnP38i2x1Uo\n"\
"muhfskWf4MABV0yTUaKcGg==\n"\
"-----END PRIVATE KEY-----\n"
server_key = '-----BEGIN PRIVATE KEY-----\n'\
'MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCXN8LK/eYi/tOU\n'\
'uQ5vG6cp8J2sn/OB0A2uzHREG2kQ+FXnrSFYnR8SKfScg4yklKIX0TpCw92vpD56\n'\
'iAfhec4xT0kT6Ibvjc3V098YTMm9GXJy38KTxKzAo8L35Veyc51mZTD3L/G0A1tR\n'\
'ED9Oym8/PPcC+/fzZ999+skaHuig6ZplIJkWVJgctkDD9eVGvSxJFsukUZjSBeNo\n'\
'BXyIceI8NgvLiRk56mNX1TnXGL4gawvjDvnO4yCwfIXecl4ZgeC0ZUGuQvRkobm5\n'\
'1jTBwGPKyO5sMkLiJKU2KrYcPd+GzuPoJl11Xa/zjkyVUo3E/SQ7hSPgPyv7lRJY\n'\
'Lwkp8DDFAgMBAAECggEAfBhAfQE7mUByNbxgAgI5fot9eaqR1Nf+QpJ6X2H3KPwC\n'\
'02sa0HOwieFwYfj6tB1doBoNq7i89mTc+QUlIn4pHgIowHO0OGawomeKz5BEhjCZ\n'\
'4XeLYGSoODary2+kNkf2xY8JTfFEcyvGBpJEwc4S2VyYgRRx+IgnumTSH+N5mIKZ\n'\
'SXWNdZIuHEmkwod+rPRXs6/r+PH0eVW6WfpINEbr4zVAGXJx2zXQwd2cuV1GTJWh\n'\
'cPVOXLu+XJ9im9B370cYN6GqUnR3fui13urYbnWnEf3syvoH/zuZkyrVChauoFf8\n'\
'8EGb74/HhXK7Q2s8NRakx2c7OxQifCbcy03liUMmyQKBgQDFAob5B/66N4Q2cq/N\n'\
'MWPf98kYBYoLaeEOhEJhLQlKk0pIFCTmtpmUbpoEes2kCUbH7RwczpYko8tlKyoB\n'\
'6Fn6RY4zQQ64KZJI6kQVsjkYpcP/ihnOY6rbds+3yyv+4uPX7Eh9sYZwZMggE19M\n'\
'CkFHkwAjiwqhiiSlUxe20sWmowKBgQDEfx4lxuFzA1PBPeZKGVBTxYPQf+DSLCre\n'\
'ZFg3ZmrxbCjRq1O7Lra4FXWD3dmRq7NDk79JofoW50yD8wD7I0B7opdDfXD2idO8\n'\
'0dBnWUKDr2CAXyoLEINce9kJPbx4kFBQRN9PiGF7VkDQxeQ3kfS8CvcErpTKCOdy\n'\
'5wOwBTwJdwKBgDiTFTeGeDv5nVoVbS67tDao7XKchJvqd9q3WGiXikeELJyuTDqE\n'\
'zW22pTwMF+m3UEAxcxVCrhMvhkUzNAkANHaOatuFHzj7lyqhO5QPbh4J3FMR0X9X\n'\
'V8VWRSg+jA/SECP9koOl6zlzd5Tee0tW1pA7QpryXscs6IEhb3ns5R2JAoGAIkzO\n'\
'RmnhEOKTzDex611f2D+yMsMfy5BKK2f4vjLymBH5TiBKDXKqEpgsW0huoi8Gq9Uu\n'\
'nvvXXAgkIyRYF36f0vUe0nkjLuYAQAWgC2pZYgNLJR13iVbol0xHJoXQUHtgiaJ8\n'\
'GLYFzjHQPqFMpSalQe3oELko39uOC1CoJCHFySECgYBeycUnRBikCO2n8DNhY4Eg\n'\
'9Y3oxcssRt6ea5BZwgW2eAYi7/XqKkmxoSoOykUt3MJx9+EkkrL17bxFSpkj1tvL\n'\
'qvxn7egtsKjjgGNAxwXC4MwCvhveyUQQxtQb8AqGrGqo4jEEN0L15cnP38i2x1Uo\n'\
'muhfskWf4MABV0yTUaKcGg==\n'\
'-----END PRIVATE KEY-----\n'
def get_my_ip():
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect(("8.8.8.8", 80))
s1.connect(('8.8.8.8', 80))
my_ip = s1.getsockname()[0]
s1.close()
return my_ip
@@ -75,13 +75,13 @@ def start_https_server(ota_image_dir, server_ip, server_port):
# args = parser.parse_args()
os.chdir(ota_image_dir)
server_file = os.path.join(ota_image_dir, "server_cert.pem")
cert_file_handle = open(server_file, "w+")
server_file = os.path.join(ota_image_dir, 'server_cert.pem')
cert_file_handle = open(server_file, 'w+')
cert_file_handle.write(server_cert)
cert_file_handle.close()
key_file = os.path.join(ota_image_dir, "server_key.pem")
key_file_handle = open("server_key.pem", "w+")
key_file = os.path.join(ota_image_dir, 'server_key.pem')
key_file_handle = open('server_key.pem', 'w+')
key_file_handle.write(server_key)
key_file_handle.close()
@@ -93,7 +93,7 @@ def start_https_server(ota_image_dir, server_ip, server_port):
httpd.serve_forever()
@ttfw_idf.idf_example_test(env_tag="Example_WIFI")
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_simple_ota_example(env, extra_data):
"""
steps: |
@@ -101,33 +101,33 @@ def test_examples_protocol_simple_ota_example(env, extra_data):
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut("simple_ota_example", "examples/system/ota/simple_ota_example", dut_class=ttfw_idf.ESP32DUT)
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', dut_class=ttfw_idf.ESP32DUT)
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "simple_ota.bin")
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("simple_ota_bin_size", "{}KB".format(bin_size // 1024))
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()
dut1.start_app()
dut1.expect("Loaded app from partition at offset 0x10000", timeout=30)
dut1.expect('Loaded app from partition at offset 0x10000', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
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)
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)
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)
@ttfw_idf.idf_example_test(env_tag="Example_EthKitV1")
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1')
def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(env, extra_data):
"""
steps: |
@@ -135,33 +135,33 @@ def test_examples_protocol_simple_ota_example_ethernet_with_spiram_config(env, e
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut("simple_ota_example", "examples/system/ota/simple_ota_example", dut_class=ttfw_idf.ESP32DUT, app_config_name='spiram')
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', dut_class=ttfw_idf.ESP32DUT, app_config_name='spiram')
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "simple_ota.bin")
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("simple_ota_bin_size", "{}KB".format(bin_size // 1024))
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()
dut1.start_app()
dut1.expect("Loaded app from partition at offset 0x10000", timeout=30)
dut1.expect('Loaded app from partition at offset 0x10000', timeout=30)
try:
ip_address = dut1.expect(re.compile(r" eth ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' eth ip: ([^,]+),'), timeout=30)
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)
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)
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)
@ttfw_idf.idf_example_test(env_tag="Example_Flash_Encryption_OTA")
@ttfw_idf.idf_example_test(env_tag='Example_Flash_Encryption_OTA')
def test_examples_protocol_simple_ota_example_with_flash_encryption(env, extra_data):
"""
steps: |
@@ -169,13 +169,13 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption(env, extra_d
2. Fetch OTA image over HTTPS
3. Reboot with the new OTA image
"""
dut1 = env.get_dut("simple_ota_example", "examples/system/ota/simple_ota_example", dut_class=ttfw_idf.ESP32DUT, app_config_name='flash_enc')
dut1 = env.get_dut('simple_ota_example', 'examples/system/ota/simple_ota_example', dut_class=ttfw_idf.ESP32DUT, app_config_name='flash_enc')
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "simple_ota.bin")
binary_file = os.path.join(dut1.app.binary_path, 'simple_ota.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("simple_ota_bin_size", "{}KB".format(bin_size // 1024))
ttfw_idf.log_performance('simple_ota_bin_size', '{}KB'.format(bin_size // 1024))
# erase flash on the device
print("Erasing the flash in order to have an empty NVS key partiton")
print('Erasing the flash in order to have an empty NVS key partiton')
dut1.erase_flash()
# start test
host_ip = get_my_ip()
@@ -183,21 +183,21 @@ def test_examples_protocol_simple_ota_example_with_flash_encryption(env, extra_d
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)
dut1.expect('Loaded app from partition at offset 0x20000', timeout=30)
dut1.expect('Flash encryption mode is DEVELOPMENT (not secure)', timeout=10)
try:
ip_address = dut1.expect(re.compile(r" eth ip: ([^,]+),"), timeout=30)
print("Connected to AP with IP: {}".format(ip_address))
ip_address = dut1.expect(re.compile(r' eth ip: ([^,]+),'), timeout=30)
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)
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 0x120000", timeout=60)
dut1.expect("Flash encryption mode is DEVELOPMENT (not secure)", timeout=10)
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 0x120000', timeout=60)
dut1.expect('Flash encryption mode is DEVELOPMENT (not secure)', timeout=10)
dut1.expect('Starting OTA example', timeout=30)
if __name__ == '__main__':

1
examples/system/perfmon/example_test.py
View File

@@ -1,4 +1,5 @@
from __future__ import unicode_literals
import ttfw_idf

4
examples/system/select/example_test.py
View File

@@ -1,7 +1,9 @@
from __future__ import unicode_literals
from tiny_test_fw import Utility
import os
import ttfw_idf
from tiny_test_fw import Utility
def get_socket_msgs(i):

8
examples/system/sysview_tracing/example_test.py
View File

@@ -1,14 +1,16 @@
from __future__ import unicode_literals
from io import open
import debug_backend
import os
import re
import tempfile
import time
from io import open
import debug_backend
import ttfw_idf
@ttfw_idf.idf_example_test(env_tag="test_jtag_arm")
@ttfw_idf.idf_example_test(env_tag='test_jtag_arm')
def test_examples_sysview_tracing(env, extra_data):
rel_project_path = os.path.join('examples', 'system', 'sysview_tracing')

8
examples/system/sysview_tracing_heap_log/example_test.py
View File

@@ -1,13 +1,15 @@
from __future__ import unicode_literals
from io import open
import debug_backend
import os
import re
import tempfile
from io import open
import debug_backend
import ttfw_idf
@ttfw_idf.idf_example_test(env_tag="test_jtag_arm")
@ttfw_idf.idf_example_test(env_tag='test_jtag_arm')
def test_examples_sysview_tracing_heap_log(env, extra_data):
rel_project_path = os.path.join('examples', 'system', 'sysview_tracing_heap_log')

1
examples/system/task_watchdog/example_test.py
View File

@@ -1,4 +1,5 @@
from __future__ import unicode_literals
import ttfw_idf

4
examples/system/ulp/example_test.py
View File

@@ -1,8 +1,10 @@
from __future__ import unicode_literals
from tiny_test_fw import Utility
import re
import time
import ttfw_idf
from tiny_test_fw import Utility
@ttfw_idf.idf_example_test(env_tag='Example_GENERIC')

4
examples/system/ulp_adc/example_test.py
View File

@@ -1,7 +1,9 @@
from __future__ import unicode_literals
from tiny_test_fw import Utility
import re
import ttfw_idf
from tiny_test_fw import Utility
@ttfw_idf.idf_example_test(env_tag='Example_GENERIC')

2
examples/system/unit_test/example_test.py
View File

@@ -1,5 +1,7 @@
from __future__ import unicode_literals
import re
import ttfw_idf