example_test.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441
  1. import re
  2. import os
  3. import struct
  4. import socket
  5. from threading import Thread
  6. import ssl
  7. from tiny_test_fw import DUT
  8. import ttfw_idf
  9. import random
  10. import subprocess
  11. try:
  12. import BaseHTTPServer
  13. from SimpleHTTPServer import SimpleHTTPRequestHandler
  14. except ImportError:
  15. import http.server as BaseHTTPServer
  16. from http.server import SimpleHTTPRequestHandler
  17. server_cert = "-----BEGIN CERTIFICATE-----\n" \
  18. "MIIDXTCCAkWgAwIBAgIJAP4LF7E72HakMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV\n"\
  19. "BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX\n"\
  20. "aWRnaXRzIFB0eSBMdGQwHhcNMTkwNjA3MDk1OTE2WhcNMjAwNjA2MDk1OTE2WjBF\n"\
  21. "MQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50\n"\
  22. "ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB\n"\
  23. "CgKCAQEAlzfCyv3mIv7TlLkObxunKfCdrJ/zgdANrsx0RBtpEPhV560hWJ0fEin0\n"\
  24. "nIOMpJSiF9E6QsPdr6Q+eogH4XnOMU9JE+iG743N1dPfGEzJvRlyct/Ck8SswKPC\n"\
  25. "9+VXsnOdZmUw9y/xtANbURA/TspvPzz3Avv382ffffrJGh7ooOmaZSCZFlSYHLZA\n"\
  26. "w/XlRr0sSRbLpFGY0gXjaAV8iHHiPDYLy4kZOepjV9U51xi+IGsL4w75zuMgsHyF\n"\
  27. "3nJeGYHgtGVBrkL0ZKG5udY0wcBjysjubDJC4iSlNiq2HD3fhs7j6CZddV2v845M\n"\
  28. "lVKNxP0kO4Uj4D8r+5USWC8JKfAwxQIDAQABo1AwTjAdBgNVHQ4EFgQU6OE7ssfY\n"\
  29. "IIPTDThiUoofUpsD5NwwHwYDVR0jBBgwFoAU6OE7ssfYIIPTDThiUoofUpsD5Nww\n"\
  30. "DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAXIlHS/FJWfmcinUAxyBd\n"\
  31. "/xd5Lu8ykeru6oaUCci+Vk9lyoMMES7lQ+b/00d5x7AcTawkTil9EWpBTPTOTraA\n"\
  32. "lzJMQhNKmSLk0iIoTtAJtSZgUSpIIozqK6lenxQQDsHbXKU6h+u9H6KZE8YcjsFl\n"\
  33. "6vL7sw9BVotw/VxfgjQ5OSGLgoLrdVT0z5C2qOuwOgz1c7jNiJhtMdwN+cOtnJp2\n"\
  34. "fuBgEYyE3eeuWogvkWoDcIA8r17Ixzkpq2oJsdvZcHZPIZShPKW2SHUsl98KDemu\n"\
  35. "y0pQyExmQUbwKE4vbFb9XuWCcL9XaOHQytyszt2DeD67AipvoBwVU7/LBOvqnsmy\n"\
  36. "hA==\n"\
  37. "-----END CERTIFICATE-----\n"
  38. server_key = "-----BEGIN PRIVATE KEY-----\n"\
  39. "MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCXN8LK/eYi/tOU\n"\
  40. "uQ5vG6cp8J2sn/OB0A2uzHREG2kQ+FXnrSFYnR8SKfScg4yklKIX0TpCw92vpD56\n"\
  41. "iAfhec4xT0kT6Ibvjc3V098YTMm9GXJy38KTxKzAo8L35Veyc51mZTD3L/G0A1tR\n"\
  42. "ED9Oym8/PPcC+/fzZ999+skaHuig6ZplIJkWVJgctkDD9eVGvSxJFsukUZjSBeNo\n"\
  43. "BXyIceI8NgvLiRk56mNX1TnXGL4gawvjDvnO4yCwfIXecl4ZgeC0ZUGuQvRkobm5\n"\
  44. "1jTBwGPKyO5sMkLiJKU2KrYcPd+GzuPoJl11Xa/zjkyVUo3E/SQ7hSPgPyv7lRJY\n"\
  45. "Lwkp8DDFAgMBAAECggEAfBhAfQE7mUByNbxgAgI5fot9eaqR1Nf+QpJ6X2H3KPwC\n"\
  46. "02sa0HOwieFwYfj6tB1doBoNq7i89mTc+QUlIn4pHgIowHO0OGawomeKz5BEhjCZ\n"\
  47. "4XeLYGSoODary2+kNkf2xY8JTfFEcyvGBpJEwc4S2VyYgRRx+IgnumTSH+N5mIKZ\n"\
  48. "SXWNdZIuHEmkwod+rPRXs6/r+PH0eVW6WfpINEbr4zVAGXJx2zXQwd2cuV1GTJWh\n"\
  49. "cPVOXLu+XJ9im9B370cYN6GqUnR3fui13urYbnWnEf3syvoH/zuZkyrVChauoFf8\n"\
  50. "8EGb74/HhXK7Q2s8NRakx2c7OxQifCbcy03liUMmyQKBgQDFAob5B/66N4Q2cq/N\n"\
  51. "MWPf98kYBYoLaeEOhEJhLQlKk0pIFCTmtpmUbpoEes2kCUbH7RwczpYko8tlKyoB\n"\
  52. "6Fn6RY4zQQ64KZJI6kQVsjkYpcP/ihnOY6rbds+3yyv+4uPX7Eh9sYZwZMggE19M\n"\
  53. "CkFHkwAjiwqhiiSlUxe20sWmowKBgQDEfx4lxuFzA1PBPeZKGVBTxYPQf+DSLCre\n"\
  54. "ZFg3ZmrxbCjRq1O7Lra4FXWD3dmRq7NDk79JofoW50yD8wD7I0B7opdDfXD2idO8\n"\
  55. "0dBnWUKDr2CAXyoLEINce9kJPbx4kFBQRN9PiGF7VkDQxeQ3kfS8CvcErpTKCOdy\n"\
  56. "5wOwBTwJdwKBgDiTFTeGeDv5nVoVbS67tDao7XKchJvqd9q3WGiXikeELJyuTDqE\n"\
  57. "zW22pTwMF+m3UEAxcxVCrhMvhkUzNAkANHaOatuFHzj7lyqhO5QPbh4J3FMR0X9X\n"\
  58. "V8VWRSg+jA/SECP9koOl6zlzd5Tee0tW1pA7QpryXscs6IEhb3ns5R2JAoGAIkzO\n"\
  59. "RmnhEOKTzDex611f2D+yMsMfy5BKK2f4vjLymBH5TiBKDXKqEpgsW0huoi8Gq9Uu\n"\
  60. "nvvXXAgkIyRYF36f0vUe0nkjLuYAQAWgC2pZYgNLJR13iVbol0xHJoXQUHtgiaJ8\n"\
  61. "GLYFzjHQPqFMpSalQe3oELko39uOC1CoJCHFySECgYBeycUnRBikCO2n8DNhY4Eg\n"\
  62. "9Y3oxcssRt6ea5BZwgW2eAYi7/XqKkmxoSoOykUt3MJx9+EkkrL17bxFSpkj1tvL\n"\
  63. "qvxn7egtsKjjgGNAxwXC4MwCvhveyUQQxtQb8AqGrGqo4jEEN0L15cnP38i2x1Uo\n"\
  64. "muhfskWf4MABV0yTUaKcGg==\n"\
  65. "-----END PRIVATE KEY-----\n"
  66. def get_my_ip():
  67. s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  68. s1.connect(("8.8.8.8", 80))
  69. my_ip = s1.getsockname()[0]
  70. s1.close()
  71. return my_ip
  72. def get_server_status(host_ip, port):
  73. sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  74. server_status = sock.connect_ex((host_ip, port))
  75. sock.close()
  76. if server_status == 0:
  77. return True
  78. return False
  79. def create_file(server_file, file_data):
  80. with open(server_file, "w+") as file:
  81. file.write(file_data)
  82. def get_ca_cert(ota_image_dir):
  83. os.chdir(ota_image_dir)
  84. server_file = os.path.join(ota_image_dir, "server_cert.pem")
  85. create_file(server_file, server_cert)
  86. key_file = os.path.join(ota_image_dir, "server_key.pem")
  87. create_file(key_file, server_key)
  88. return server_file, key_file
  89. def start_https_server(ota_image_dir, server_ip, server_port):
  90. server_file, key_file = get_ca_cert(ota_image_dir)
  91. httpd = BaseHTTPServer.HTTPServer((server_ip, server_port),
  92. SimpleHTTPRequestHandler)
  93. httpd.socket = ssl.wrap_socket(httpd.socket,
  94. keyfile=key_file,
  95. certfile=server_file, server_side=True)
  96. httpd.serve_forever()
  97. def start_chunked_server(ota_image_dir, server_port):
  98. server_file, key_file = get_ca_cert(ota_image_dir)
  99. chunked_server = subprocess.Popen(["openssl", "s_server", "-WWW", "-key", key_file, "-cert", server_file, "-port", str(server_port)])
  100. return chunked_server
  101. def redirect_handler_factory(url):
  102. """
  103. Returns a request handler class that redirects to supplied `url`
  104. """
  105. class RedirectHandler(SimpleHTTPRequestHandler):
  106. def do_GET(self):
  107. print("Sending resp, URL: " + url)
  108. self.send_response(301)
  109. self.send_header('Location', url)
  110. self.end_headers()
  111. return RedirectHandler
  112. def start_redirect_server(ota_image_dir, server_ip, server_port, redirection_port):
  113. os.chdir(ota_image_dir)
  114. server_file, key_file = get_ca_cert(ota_image_dir)
  115. redirectHandler = redirect_handler_factory("https://" + server_ip + ":" + str(redirection_port) + "/advanced_https_ota.bin")
  116. httpd = BaseHTTPServer.HTTPServer((server_ip, server_port),
  117. redirectHandler)
  118. httpd.socket = ssl.wrap_socket(httpd.socket,
  119. keyfile=key_file,
  120. certfile=server_file, server_side=True)
  121. httpd.serve_forever()
  122. @ttfw_idf.idf_example_test(env_tag="Example_WIFI")
  123. def test_examples_protocol_advanced_https_ota_example(env, extra_data):
  124. """
  125. This is a positive test case, which downloads complete binary file multiple number of times.
  126. Number of iterations can be specified in variable iterations.
  127. steps: |
  128. 1. join AP
  129. 2. Fetch OTA image over HTTPS
  130. 3. Reboot with the new OTA image
  131. """
  132. dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
  133. # Number of iterations to validate OTA
  134. iterations = 3
  135. server_port = 8001
  136. # File to be downloaded. This file is generated after compilation
  137. bin_name = "advanced_https_ota.bin"
  138. # check and log bin size
  139. binary_file = os.path.join(dut1.app.binary_path, bin_name)
  140. bin_size = os.path.getsize(binary_file)
  141. ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
  142. ttfw_idf.check_performance("advanced_https_ota_bin_size", bin_size // 1024)
  143. # start test
  144. host_ip = get_my_ip()
  145. if (get_server_status(host_ip, server_port) is False):
  146. thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
  147. thread1.daemon = True
  148. thread1.start()
  149. dut1.start_app()
  150. for i in range(iterations):
  151. dut1.expect("Loaded app from partition at offset", timeout=30)
  152. try:
  153. ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
  154. print("Connected to AP with IP: {}".format(ip_address))
  155. except DUT.ExpectTimeout:
  156. raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
  157. thread1.close()
  158. dut1.expect("Starting Advanced OTA example", timeout=30)
  159. print("writing to device: {}".format("https://" + host_ip + ":" + str(server_port) + "/" + bin_name))
  160. dut1.write("https://" + host_ip + ":" + str(server_port) + "/" + bin_name)
  161. dut1.expect("Loaded app from partition at offset", timeout=60)
  162. dut1.expect("Starting Advanced OTA example", timeout=30)
  163. dut1.reset()
  164. @ttfw_idf.idf_example_test(env_tag="Example_WIFI")
  165. def test_examples_protocol_advanced_https_ota_example_truncated_bin(env, extra_data):
  166. """
  167. Working of OTA if binary file is truncated is validated in this test case.
  168. Application should return with error message in this case.
  169. steps: |
  170. 1. join AP
  171. 2. Generate truncated binary file
  172. 3. Fetch OTA image over HTTPS
  173. 4. Check working of code if bin is truncated
  174. """
  175. dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
  176. server_port = 8001
  177. # Original binary file generated after compilation
  178. bin_name = "advanced_https_ota.bin"
  179. # Truncated binary file to be generated from original binary file
  180. truncated_bin_name = "truncated.bin"
  181. # Size of truncated file to be grnerated. This value can range from 288 bytes (Image header size) to size of original binary file
  182. # truncated_bin_size is set to 64000 to reduce consumed by the test case
  183. truncated_bin_size = 64000
  184. # check and log bin size
  185. binary_file = os.path.join(dut1.app.binary_path, bin_name)
  186. f = open(binary_file, "rb+")
  187. fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
  188. fo.write(f.read(truncated_bin_size))
  189. fo.close()
  190. f.close()
  191. binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
  192. bin_size = os.path.getsize(binary_file)
  193. ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
  194. ttfw_idf.check_performance("advanced_https_ota_bin_size", bin_size // 1024)
  195. # start test
  196. host_ip = get_my_ip()
  197. if (get_server_status(host_ip, server_port) is False):
  198. thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
  199. thread1.daemon = True
  200. thread1.start()
  201. dut1.start_app()
  202. dut1.expect("Loaded app from partition at offset", timeout=30)
  203. try:
  204. ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
  205. print("Connected to AP with IP: {}".format(ip_address))
  206. except DUT.ExpectTimeout:
  207. raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
  208. dut1.expect("Starting Advanced OTA example", timeout=30)
  209. print("writing to device: {}".format("https://" + host_ip + ":" + str(server_port) + "/" + truncated_bin_name))
  210. dut1.write("https://" + host_ip + ":" + str(server_port) + "/" + truncated_bin_name)
  211. dut1.expect("Image validation failed, image is corrupted", timeout=30)
  212. os.remove(binary_file)
  213. @ttfw_idf.idf_example_test(env_tag="Example_WIFI")
  214. def test_examples_protocol_advanced_https_ota_example_truncated_header(env, extra_data):
  215. """
  216. Working of OTA if headers of binary file are truncated is vaildated in this test case.
  217. Application should return with error message in this case.
  218. steps: |
  219. 1. join AP
  220. 2. Generate binary file with truncated headers
  221. 3. Fetch OTA image over HTTPS
  222. 4. Check working of code if headers are not sent completely
  223. """
  224. dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
  225. server_port = 8001
  226. # Original binary file generated after compilation
  227. bin_name = "advanced_https_ota.bin"
  228. # Truncated binary file to be generated from original binary file
  229. truncated_bin_name = "truncated_header.bin"
  230. # Size of truncated file to be grnerated. This value should be less than 288 bytes (Image header size)
  231. truncated_bin_size = 180
  232. # check and log bin size
  233. binary_file = os.path.join(dut1.app.binary_path, bin_name)
  234. f = open(binary_file, "rb+")
  235. fo = open(os.path.join(dut1.app.binary_path, truncated_bin_name), "wb+")
  236. fo.write(f.read(truncated_bin_size))
  237. fo.close()
  238. f.close()
  239. binary_file = os.path.join(dut1.app.binary_path, truncated_bin_name)
  240. bin_size = os.path.getsize(binary_file)
  241. ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
  242. ttfw_idf.check_performance("advanced_https_ota_bin_size", bin_size // 1024)
  243. # start test
  244. host_ip = get_my_ip()
  245. if (get_server_status(host_ip, server_port) is False):
  246. thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
  247. thread1.daemon = True
  248. thread1.start()
  249. dut1.start_app()
  250. dut1.expect("Loaded app from partition at offset", timeout=30)
  251. try:
  252. ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
  253. print("Connected to AP with IP: {}".format(ip_address))
  254. except DUT.ExpectTimeout:
  255. raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
  256. dut1.expect("Starting Advanced OTA example", timeout=30)
  257. print("writing to device: {}".format("https://" + host_ip + ":" + str(server_port) + "/" + truncated_bin_name))
  258. dut1.write("https://" + host_ip + ":" + str(server_port) + "/" + truncated_bin_name)
  259. dut1.expect("advanced_https_ota_example: esp_https_ota_read_img_desc failed", timeout=30)
  260. os.remove(binary_file)
  261. @ttfw_idf.idf_example_test(env_tag="Example_WIFI")
  262. def test_examples_protocol_advanced_https_ota_example_random(env, extra_data):
  263. """
  264. Working of OTA if random data is added in binary file are validated in this test case.
  265. Magic byte verification should fail in this case.
  266. steps: |
  267. 1. join AP
  268. 2. Generate random binary image
  269. 3. Fetch OTA image over HTTPS
  270. 4. Check working of code for random binary file
  271. """
  272. dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
  273. server_port = 8001
  274. # Random binary file to be generated
  275. random_bin_name = "random.bin"
  276. # Size of random binary file. 32000 is choosen, to reduce the time required to run the test-case
  277. random_bin_size = 32000
  278. # check and log bin size
  279. binary_file = os.path.join(dut1.app.binary_path, random_bin_name)
  280. fo = open(binary_file, "wb+")
  281. # First byte of binary file is always set to zero. If first byte is generated randomly,
  282. # in some cases it may generate 0xE9 which will result in failure of testcase.
  283. fo.write(struct.pack("B", 0))
  284. for i in range(random_bin_size - 1):
  285. fo.write(struct.pack("B", random.randrange(0,255,1)))
  286. fo.close()
  287. bin_size = os.path.getsize(binary_file)
  288. ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
  289. ttfw_idf.check_performance("advanced_https_ota_bin_size", bin_size // 1024)
  290. # start test
  291. host_ip = get_my_ip()
  292. if (get_server_status(host_ip, server_port) is False):
  293. thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
  294. thread1.daemon = True
  295. thread1.start()
  296. dut1.start_app()
  297. dut1.expect("Loaded app from partition at offset", timeout=30)
  298. try:
  299. ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
  300. print("Connected to AP with IP: {}".format(ip_address))
  301. except DUT.ExpectTimeout:
  302. raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
  303. dut1.expect("Starting Advanced OTA example", timeout=30)
  304. print("writing to device: {}".format("https://" + host_ip + ":" + str(server_port) + "/" + random_bin_name))
  305. dut1.write("https://" + host_ip + ":" + str(server_port) + "/" + random_bin_name)
  306. dut1.expect("esp_ota_ops: OTA image has invalid magic byte", timeout=10)
  307. os.remove(binary_file)
  308. @ttfw_idf.idf_example_test(env_tag="Example_WIFI")
  309. def test_examples_protocol_advanced_https_ota_example_chunked(env, extra_data):
  310. """
  311. This is a positive test case, which downloads complete binary file multiple number of times.
  312. Number of iterations can be specified in variable iterations.
  313. steps: |
  314. 1. join AP
  315. 2. Fetch OTA image over HTTPS
  316. 3. Reboot with the new OTA image
  317. """
  318. dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
  319. # File to be downloaded. This file is generated after compilation
  320. bin_name = "advanced_https_ota.bin"
  321. # check and log bin size
  322. binary_file = os.path.join(dut1.app.binary_path, bin_name)
  323. bin_size = os.path.getsize(binary_file)
  324. ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
  325. ttfw_idf.check_performance("advanced_https_ota_bin_size", 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 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. print("writing to device: {}".format("https://" + host_ip + ":8070/" + bin_name))
  338. dut1.write("https://" + host_ip + ":8070/" + bin_name)
  339. dut1.expect("Loaded app from partition at offset", timeout=60)
  340. dut1.expect("Starting Advanced OTA example", timeout=30)
  341. chunked_server.kill()
  342. os.remove(os.path.join(dut1.app.binary_path, "server_cert.pem"))
  343. os.remove(os.path.join(dut1.app.binary_path, "server_key.pem"))
  344. @ttfw_idf.idf_example_test(env_tag="Example_WIFI")
  345. def test_examples_protocol_advanced_https_ota_example_redirect_url(env, extra_data):
  346. """
  347. This is a positive test case, which starts a server and a redirection server.
  348. Redirection server redirects http_request to different port
  349. Number of iterations can be specified in variable iterations.
  350. steps: |
  351. 1. join AP
  352. 2. Fetch OTA image over HTTPS
  353. 3. Reboot with the new OTA image
  354. """
  355. dut1 = env.get_dut("advanced_https_ota_example", "examples/system/ota/advanced_https_ota", dut_class=ttfw_idf.ESP32DUT)
  356. server_port = 8001
  357. # Port to which the request should be redirecetd
  358. redirection_server_port = 8081
  359. # File to be downloaded. This file is generated after compilation
  360. bin_name = "advanced_https_ota.bin"
  361. # check and log bin size
  362. binary_file = os.path.join(dut1.app.binary_path, bin_name)
  363. bin_size = os.path.getsize(binary_file)
  364. ttfw_idf.log_performance("advanced_https_ota_bin_size", "{}KB".format(bin_size // 1024))
  365. ttfw_idf.check_performance("advanced_https_ota_bin_size", 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 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, server_port))
  370. thread1.daemon = True
  371. thread1.start()
  372. thread2 = Thread(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 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.close()
  383. thread2.close()
  384. dut1.expect("Starting Advanced OTA example", timeout=30)
  385. print("writing to device: {}".format("https://" + host_ip + ":" + str(redirection_server_port) + "/" + bin_name))
  386. dut1.write("https://" + host_ip + ":" + str(redirection_server_port) + "/" + bin_name)
  387. dut1.expect("Loaded app from partition at offset", timeout=60)
  388. dut1.expect("Starting Advanced OTA example", timeout=30)
  389. dut1.reset()
  390. if __name__ == '__main__':
  391. test_examples_protocol_advanced_https_ota_example()
  392. test_examples_protocol_advanced_https_ota_example_chunked()
  393. test_examples_protocol_advanced_https_ota_example_redirect_url()
  394. test_examples_protocol_advanced_https_ota_example_truncated_bin()
  395. test_examples_protocol_advanced_https_ota_example_truncated_header()
  396. test_examples_protocol_advanced_https_ota_example_random()