1import http.server
2import multiprocessing
3import os
4import random
5import re
6import socket
7import ssl
8import struct
9import subprocess
10
11import ttfw_idf
12from RangeHTTPServer import RangeRequestHandler
13from tiny_test_fw import DUT, Utility
14
15server_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_certs/server_cert.pem')
16key_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_certs/server_key.pem')
17
18
19def get_my_ip():
20    s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
21    s1.connect(('8.8.8.8', 80))
22    my_ip = s1.getsockname()[0]
23    s1.close()
24    return my_ip
25
26
27def get_server_status(host_ip, port):
28    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
29    server_status = sock.connect_ex((host_ip, port))
30    sock.close()
31    if server_status == 0:
32        return True
33    return False
34
35
36def https_request_handler():
37    """
38    Returns a request handler class that handles broken pipe exception
39    """
40    class RequestHandler(RangeRequestHandler):
41        def finish(self):
42            try:
43                if not self.wfile.closed:
44                    self.wfile.flush()
45                    self.wfile.close()
46            except socket.error:
47                pass
48            self.rfile.close()
49
50        def handle(self):
51            try:
52                RangeRequestHandler.handle(self)
53            except socket.error:
54                pass
55
56    return RequestHandler
57
58
59def start_https_server(ota_image_dir, server_ip, server_port):
60    os.chdir(ota_image_dir)
61    requestHandler = https_request_handler()
62    httpd = http.server.HTTPServer((server_ip, server_port), requestHandler)
63
64    httpd.socket = ssl.wrap_socket(httpd.socket,
65                                   keyfile=key_file,
66                                   certfile=server_file, server_side=True)
67    httpd.serve_forever()
68
69
70def start_chunked_server(ota_image_dir, server_port):
71    os.chdir(ota_image_dir)
72    chunked_server = subprocess.Popen(['openssl', 's_server', '-WWW', '-key', key_file, '-cert', server_file, '-port', str(server_port)])
73    return chunked_server
74
75
76def redirect_handler_factory(url):
77    """
78    Returns a request handler class that redirects to supplied `url`
79    """
80    class RedirectHandler(http.server.SimpleHTTPRequestHandler):
81        def do_GET(self):
82            print('Sending resp, URL: ' + url)
83            self.send_response(301)
84            self.send_header('Location', url)
85            self.end_headers()
86
87        def handle(self):
88            try:
89                http.server.BaseHTTPRequestHandler.handle(self)
90            except socket.error:
91                pass
92
93    return RedirectHandler
94
95
96def start_redirect_server(ota_image_dir, server_ip, server_port, redirection_port):
97    os.chdir(ota_image_dir)
98    redirectHandler = redirect_handler_factory('https://' + server_ip + ':' + str(redirection_port) + '/advanced_https_ota.bin')
99
100    httpd = http.server.HTTPServer((server_ip, server_port), redirectHandler)
101
102    httpd.socket = ssl.wrap_socket(httpd.socket,
103                                   keyfile=key_file,
104                                   certfile=server_file, server_side=True)
105    httpd.serve_forever()
106
107
108@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
109def test_examples_protocol_advanced_https_ota_example(env, extra_data):
110    """
111    This is a positive test case, which downloads complete binary file multiple number of times.
112    Number of iterations can be specified in variable iterations.
113    steps: |
114      1. join AP
115      2. Fetch OTA image over HTTPS
116      3. Reboot with the new OTA image
117    """
118    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
119    # Number of iterations to validate OTA
120    iterations = 3
121    server_port = 8001
122    # File to be downloaded. This file is generated after compilation
123    bin_name = 'advanced_https_ota.bin'
124    # check and log bin size
125    binary_file = os.path.join(dut1.app.binary_path, bin_name)
126    bin_size = os.path.getsize(binary_file)
127    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
128    # start test
129    host_ip = get_my_ip()
130    if (get_server_status(host_ip, server_port) is False):
131        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
132        thread1.daemon = True
133        thread1.start()
134    dut1.start_app()
135    for i in range(iterations):
136        dut1.expect('Loaded app from partition at offset', timeout=30)
137        try:
138            ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
139            print('Connected to AP with IP: {}'.format(ip_address))
140        except DUT.ExpectTimeout:
141            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
142            thread1.terminate()
143        dut1.expect('Starting Advanced OTA example', timeout=30)
144
145        print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
146        dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
147        dut1.expect('Loaded app from partition at offset', timeout=60)
148        dut1.expect('Starting Advanced OTA example', timeout=30)
149        dut1.reset()
150    thread1.terminate()
151
152
153@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
154def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_data):
155    """
156    Working of OTA if binary file is truncated is validated in this test case.
157    Application should return with error message in this case.
158    steps: |
159      1. join AP
160      2. Generate truncated binary file
161      3. Fetch OTA image over HTTPS
162      4. Check working of code if bin is truncated
163    """
164    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
165    server_port = 8001
166    # Original binary file generated after compilation
167    bin_name = 'advanced_https_ota.bin'
168    # Truncated binary file to be generated from original binary file
169    truncated_bin_name = 'truncated.bin'
170    # Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
171    # truncated_bin_size is set to 64000 to reduce consumed by the test case
172    truncated_bin_size = 64000
173    # check and log bin size
174    binary_file = os.path.join(dut1.app.binary_path, bin_name)
175    f = open(binary_file, 'rb+')
176    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+')
177    fo.write(f.read(truncated_bin_size))
178    fo.close()
179    f.close()
180    binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
181    bin_size = os.path.getsize(binary_file)
182    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
183    # start test
184    host_ip = get_my_ip()
185    if (get_server_status(host_ip, server_port) is False):
186        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
187        thread1.daemon = True
188        thread1.start()
189    dut1.start_app()
190    dut1.expect('Loaded app from partition at offset', timeout=30)
191    try:
192        ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
193        print('Connected to AP with IP: {}'.format(ip_address))
194    except DUT.ExpectTimeout:
195        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
196        thread1.terminate()
197    dut1.expect('Starting Advanced OTA example', timeout=30)
198
199    print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
200    dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
201    dut1.expect('Image validation failed, image is corrupted', timeout=30)
202    os.remove(binary_file)
203    thread1.terminate()
204
205
206@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
207def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extra_data):
208    """
209    Working of OTA if headers of binary file are truncated is vaildated in this test case.
210    Application should return with error message in this case.
211    steps: |
212      1. join AP
213      2. Generate binary file with truncated headers
214      3. Fetch OTA image over HTTPS
215      4. Check working of code if headers are not sent completely
216    """
217    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
218    server_port = 8001
219    # Original binary file generated after compilation
220    bin_name = 'advanced_https_ota.bin'
221    # Truncated binary file to be generated from original binary file
222    truncated_bin_name = 'truncated_header.bin'
223    # Size of truncated file to be grnerated. This value should be less than 288 bytes (Image header size)
224    truncated_bin_size = 180
225    # check and log bin size
226    binary_file = os.path.join(dut1.app.binary_path, bin_name)
227    f = open(binary_file, 'rb+')
228    fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), 'wb+')
229    fo.write(f.read(truncated_bin_size))
230    fo.close()
231    f.close()
232    binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
233    bin_size = os.path.getsize(binary_file)
234    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
235    # start test
236    host_ip = get_my_ip()
237    if (get_server_status(host_ip, server_port) is False):
238        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
239        thread1.daemon = True
240        thread1.start()
241    dut1.start_app()
242    dut1.expect('Loaded app from partition at offset', timeout=30)
243    try:
244        ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
245        print('Connected to AP with IP: {}'.format(ip_address))
246    except DUT.ExpectTimeout:
247        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
248        thread1.terminate()
249    dut1.expect('Starting Advanced OTA example', timeout=30)
250
251    print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name))
252    dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + truncated_bin_name)
253    dut1.expect('advanced_https_ota_example: esp_https_ota_read_img_desc failed', timeout=30)
254    os.remove(binary_file)
255    thread1.terminate()
256
257
258@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
259def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
260    """
261    Working of OTA if random data is added in binary file are validated in this test case.
262    Magic byte verification should fail in this case.
263    steps: |
264      1. join AP
265      2. Generate random binary image
266      3. Fetch OTA image over HTTPS
267      4. Check working of code for random binary file
268    """
269    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
270    server_port = 8001
271    # Random binary file to be generated
272    random_bin_name = 'random.bin'
273    # Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
274    random_bin_size = 32000
275    # check and log bin size
276    binary_file = os.path.join(dut1.app.binary_path, random_bin_name)
277    fo = open(binary_file, 'wb+')
278    # First byte of binary file is always set to zero. If first byte is generated randomly,
279    # in some cases it may generate 0xE9 which will result in failure of testcase.
280    fo.write(struct.pack('B', 0))
281    for i in range(random_bin_size - 1):
282        fo.write(struct.pack('B', random.randrange(0,255,1)))
283    fo.close()
284    bin_size = os.path.getsize(binary_file)
285    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
286    # start test
287    host_ip = get_my_ip()
288    if (get_server_status(host_ip, server_port) is False):
289        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
290        thread1.daemon = True
291        thread1.start()
292    dut1.start_app()
293    dut1.expect('Loaded app from partition at offset', timeout=30)
294    try:
295        ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
296        print('Connected to AP with IP: {}'.format(ip_address))
297    except DUT.ExpectTimeout:
298        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
299        thread1.terminate()
300    dut1.expect('Starting Advanced OTA example', timeout=30)
301
302    print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name))
303    dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + random_bin_name)
304    dut1.expect(re.compile(r'esp_https_ota: Mismatch chip id, expected 0, found \d'), timeout=10)
305    os.remove(binary_file)
306    thread1.terminate()
307
308
309@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
310def test_examples_protocol_advanced_https_ota_example_chunked(env, extra_data):
311    """
312    This is a positive test case, which downloads complete binary file multiple number of times.
313    Number of iterations can be specified in variable iterations.
314    steps: |
315      1. join AP
316      2. Fetch OTA image over HTTPS
317      3. Reboot with the new OTA image
318    """
319    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
320    # File to be downloaded. This file is generated after compilation
321    bin_name = 'advanced_https_ota.bin'
322    # check and log bin size
323    binary_file = os.path.join(dut1.app.binary_path, bin_name)
324    bin_size = os.path.getsize(binary_file)
325    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
326    # start test
327    host_ip = get_my_ip()
328    chunked_server = start_chunked_server(dut1.app.binary_path, 8070)
329    dut1.start_app()
330    dut1.expect('Loaded app from partition at offset', timeout=30)
331    try:
332        ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
333        print('Connected to AP with IP: {}'.format(ip_address))
334    except DUT.ExpectTimeout:
335        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
336    dut1.expect('Starting Advanced OTA example', timeout=30)
337
338    print('writing to device: {}'.format('https://' + host_ip + ':8070/' + bin_name))
339    dut1.write('https://' + host_ip + ':8070/' + bin_name)
340    dut1.expect('Loaded app from partition at offset', timeout=60)
341    dut1.expect('Starting Advanced OTA example', timeout=30)
342    chunked_server.kill()
343
344
345@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
346def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_data):
347    """
348    This is a positive test case, which starts a server and a redirection server.
349    Redirection server redirects http_request to different port
350    Number of iterations can be specified in variable iterations.
351    steps: |
352      1. join AP
353      2. Fetch OTA image over HTTPS
354      3. Reboot with the new OTA image
355    """
356    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
357    server_port = 8001
358    # Port to which the request should be redirecetd
359    redirection_server_port = 8081
360    # File to be downloaded. This file is generated after compilation
361    bin_name = 'advanced_https_ota.bin'
362    # check and log bin size
363    binary_file = os.path.join(dut1.app.binary_path, bin_name)
364    bin_size = os.path.getsize(binary_file)
365    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
366    # start test
367    host_ip = get_my_ip()
368    if (get_server_status(host_ip, server_port) is False):
369        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
370        thread1.daemon = True
371        thread1.start()
372    thread2 = multiprocessing.Process(target=start_redirect_server, args=(dut1.app.binary_path, host_ip, redirection_server_port, server_port))
373    thread2.daemon = True
374    thread2.start()
375    dut1.start_app()
376    dut1.expect('Loaded app from partition at offset', timeout=30)
377    try:
378        ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
379        print('Connected to AP with IP: {}'.format(ip_address))
380    except DUT.ExpectTimeout:
381        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
382        thread1.terminate()
383        thread2.terminate()
384    dut1.expect('Starting Advanced OTA example', timeout=30)
385
386    print('writing to device: {}'.format('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name))
387    dut1.write('https://' + host_ip + ':' + str(redirection_server_port) + '/' + bin_name)
388    dut1.expect('Loaded app from partition at offset', timeout=60)
389    dut1.expect('Starting Advanced OTA example', timeout=30)
390    dut1.reset()
391    thread1.terminate()
392    thread2.terminate()
393
394
395@ttfw_idf.idf_example_test(env_tag='Example_8Mflash_Ethernet')
396def test_examples_protocol_advanced_https_ota_example_anti_rollback(env, extra_data):
397    """
398    Working of OTA when anti_rollback is enabled and security version of new image is less than current one.
399    Application should return with error message in this case.
400    steps: |
401      1. join AP
402      2. Generate binary file with lower security version
403      3. Fetch OTA image over HTTPS
404      4. Check working of anti_rollback feature
405    """
406    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='anti_rollback')
407    Utility.console_log('Erasing the flash on the chip')
408    # erase the flash
409    dut1.erase_flash()
410    server_port = 8001
411    # Original binary file generated after compilation
412    bin_name = 'advanced_https_ota.bin'
413    # Modified firmware image to lower security version in its header. This is to enable negative test case
414    anti_rollback_bin_name = 'advanced_https_ota_lower_sec_version.bin'
415    # check and log bin size
416    binary_file = os.path.join(dut1.app.binary_path, bin_name)
417    file_size = os.path.getsize(binary_file)
418    f = open(binary_file, 'rb+')
419    fo = open(os.path.join(dut1.app.binary_path, anti_rollback_bin_name), 'wb+')
420    fo.write(f.read(file_size))
421    # Change security_version to 0 for negative test case
422    fo.seek(36)
423    fo.write(b'\x00')
424    fo.close()
425    f.close()
426    binary_file = os.path.join(dut1.app.binary_path, anti_rollback_bin_name)
427    bin_size = os.path.getsize(binary_file)
428    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
429    # start test
430    host_ip = get_my_ip()
431    if (get_server_status(host_ip, server_port) is False):
432        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
433        thread1.daemon = True
434        thread1.start()
435    dut1.start_app()
436    # Positive Case
437    dut1.expect('Loaded app from partition at offset', timeout=30)
438    try:
439        ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
440        print('Connected to AP with IP: {}'.format(ip_address))
441    except DUT.ExpectTimeout:
442        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
443        thread1.terminate()
444    dut1.expect('Starting Advanced OTA example', timeout=30)
445
446    # Use originally generated image with secure_version=1
447    print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
448    dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
449    dut1.expect('Loaded app from partition at offset', timeout=60)
450    dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
451    dut1.expect('App is valid, rollback cancelled successfully', 30)
452
453    # Negative Case
454    dut1.expect('Starting Advanced OTA example', timeout=30)
455    # Use modified image with secure_version=0
456    print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name))
457    dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + anti_rollback_bin_name)
458    dut1.expect('New firmware security version is less than eFuse programmed, 0 < 1', timeout=30)
459    os.remove(binary_file)
460    thread1.terminate()
461
462
463@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
464def test_examples_protocol_advanced_https_ota_example_partial_request(env, extra_data):
465    """
466    This is a positive test case, to test OTA workflow with Range HTTP header.
467    steps: |
468      1. join AP
469      2. Fetch OTA image over HTTPS
470      3. Reboot with the new OTA image
471    """
472    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='partial_download')
473    server_port = 8001
474    # Size of partial HTTP request
475    request_size = 16384
476    # File to be downloaded. This file is generated after compilation
477    bin_name = 'advanced_https_ota.bin'
478    # check and log bin size
479    binary_file = os.path.join(dut1.app.binary_path, bin_name)
480    bin_size = os.path.getsize(binary_file)
481    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
482    http_requests = int((bin_size / request_size) - 1)
483    # start test
484    host_ip = get_my_ip()
485    if (get_server_status(host_ip, server_port) is False):
486        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
487        thread1.daemon = True
488        thread1.start()
489    dut1.start_app()
490    dut1.expect('Loaded app from partition at offset', timeout=30)
491    try:
492        ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
493        print('Connected to AP with IP: {}'.format(ip_address))
494    except DUT.ExpectTimeout:
495        Utility.console_log('ENV_TEST_FAILURE: Cannot connect to AP')
496        raise
497        thread1.terminate()
498    dut1.expect('Starting Advanced OTA example', timeout=30)
499
500    print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
501    dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
502    for _ in range(http_requests):
503        dut1.expect('Connection closed', timeout=60)
504    dut1.expect('Loaded app from partition at offset', timeout=60)
505    dut1.expect('Starting Advanced OTA example', timeout=30)
506    dut1.reset()
507    thread1.terminate()
508
509
510@ttfw_idf.idf_example_test(env_tag='Example_WIFI_OTA', nightly_run=True)
511def test_examples_protocol_advanced_https_ota_example_nimble_gatts(env, extra_data):
512    """
513    Run an OTA image update while a BLE GATT Server is running in background. This GATT server will be using NimBLE Host stack.
514    steps: |
515      1. join AP
516      2. Run BLE advertise and then GATT server.
517      3. Fetch OTA image over HTTPS
518      4. Reboot with the new OTA image
519    """
520    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='nimble')
521    server_port = 8001
522    # File to be downloaded. This file is generated after compilation
523    bin_name = 'advanced_https_ota.bin'
524    # check and log bin size
525    binary_file = os.path.join(dut1.app.binary_path, bin_name)
526    bin_size = os.path.getsize(binary_file)
527    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
528    # start test
529    host_ip = get_my_ip()
530    if (get_server_status(host_ip, server_port) is False):
531        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
532        thread1.daemon = True
533        thread1.start()
534    dut1.start_app()
535    dut1.expect('Loaded app from partition at offset', timeout=30)
536    try:
537        ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
538        print('Connected to AP with IP: {}'.format(ip_address))
539    except DUT.ExpectTimeout:
540        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
541        thread1.terminate()
542
543    dut1.expect('Starting Advanced OTA example', timeout=30)
544    print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
545    dut1.expect('GAP procedure initiated: advertise', timeout=30)
546    print('Started GAP advertising.')
547
548    dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
549    dut1.expect('Loaded app from partition at offset', timeout=60)
550    dut1.expect('Starting Advanced OTA example', timeout=30)
551    dut1.reset()
552    thread1.terminate()
553
554
555@ttfw_idf.idf_example_test(env_tag='Example_WIFI_OTA', nightly_run=True)
556def test_examples_protocol_advanced_https_ota_example_bluedroid_gatts(env, extra_data):
557    """
558    Run an OTA image update while a BLE GATT Server is running in background. This GATT server will be using Bluedroid Host stack.
559    steps: |
560      1. join AP
561      2. Run BLE advertise and then GATT server.
562      3. Fetch OTA image over HTTPS
563      4. Reboot with the new OTA image
564    """
565    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT, app_config_name='bluedroid')
566    server_port = 8001
567    # File to be downloaded. This file is generated after compilation
568    bin_name = 'advanced_https_ota.bin'
569    # check and log bin size
570    binary_file = os.path.join(dut1.app.binary_path, bin_name)
571    bin_size = os.path.getsize(binary_file)
572    ttfw_idf.log_performance('advanced_https_ota_bin_size', '{}KB'.format(bin_size // 1024))
573    # start test
574    host_ip = get_my_ip()
575    if (get_server_status(host_ip, server_port) is False):
576        thread1 = multiprocessing.Process(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
577        thread1.daemon = True
578        thread1.start()
579    dut1.start_app()
580    dut1.expect('Loaded app from partition at offset', timeout=30)
581    try:
582        ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)
583        print('Connected to AP with IP: {}'.format(ip_address))
584    except DUT.ExpectTimeout:
585        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
586        thread1.terminate()
587
588    dut1.expect('Starting Advanced OTA example', timeout=30)
589    print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
590    dut1.expect('Started advertising.', timeout=30)
591    print('Started GAP advertising.')
592
593    dut1.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
594    dut1.expect('Loaded app from partition at offset', timeout=60)
595    dut1.expect('Starting Advanced OTA example', timeout=30)
596    dut1.reset()
597    thread1.terminate()
598
599
600@ttfw_idf.idf_example_test(env_tag='EXAMPLE_ETH_OTA')
601def test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin(env, extra_data):
602    """
603    This is a test case for esp_http_client_read with binary size multiple of 289 bytes
604    steps: |
605      1. join AP
606      2. Fetch OTA image over HTTPS
607      3. Reboot with the new OTA image
608    """
609    dut1 = env.get_dut('advanced_https_ota_example', 'examples/system/ota/advanced_https_ota', dut_class=ttfw_idf.ESP32DUT)
610    # Original binary file generated after compilation
611    bin_name = 'advanced_https_ota.bin'
612    # Binary file aligned to DEFAULT_OTA_BUF_SIZE(289 bytes) boundary
613    aligned_bin_name = 'aligned.bin'
614    # check and log bin size
615    binary_file = os.path.join(dut1.app.binary_path, bin_name)
616    # Original binary size
617    bin_size = os.path.getsize(binary_file)
618    # Dummy data required to align binary size to 289 bytes boundary
619    dummy_data_size = 289 - (bin_size % 289)
620    f = open(binary_file, 'rb+')
621    fo = open(os.path.join(dut1.app.binary_path, aligned_bin_name), 'wb+')
622    fo.write(f.read(bin_size))
623    for _ in range(dummy_data_size):
624        fo.write(struct.pack('B', random.randrange(0,255,1)))
625    fo.close()
626    f.close()
627    # start test
628    host_ip = get_my_ip()
629    chunked_server = start_chunked_server(dut1.app.binary_path, 8070)
630    dut1.start_app()
631    dut1.expect('Loaded app from partition at offset', timeout=30)
632    try:
633        ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=30)
634        print('Connected to AP with IP: {}'.format(ip_address))
635    except DUT.ExpectTimeout:
636        raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
637    dut1.expect('Starting Advanced OTA example', timeout=30)
638
639    print('writing to device: {}'.format('https://' + host_ip + ':8070/' + aligned_bin_name))
640    dut1.write('https://' + host_ip + ':8070/' + aligned_bin_name)
641    dut1.expect('Loaded app from partition at offset', timeout=60)
642    dut1.expect('Starting Advanced OTA example', timeout=30)
643    chunked_server.kill()
644    os.remove(aligned_bin_name)
645
646
647if __name__ == '__main__':
648    test_examples_protocol_advanced_https_ota_example()
649    test_examples_protocol_advanced_https_ota_example_chunked()
650    test_examples_protocol_advanced_https_ota_example_redirect_url()
651    test_examples_protocol_advanced_https_ota_example_truncated_bin()
652    test_examples_protocol_advanced_https_ota_example_truncated_header()
653    test_examples_protocol_advanced_https_ota_example_random()
654    test_examples_protocol_advanced_https_ota_example_anti_rollback()
655    test_examples_protocol_advanced_https_ota_example_partial_request()
656    test_examples_protocol_advanced_https_ota_example_nimble_gatts()
657    test_examples_protocol_advanced_https_ota_example_bluedroid_gatts()
658    test_examples_protocol_advanced_https_ota_example_openssl_aligned_bin()
659