example_test.py 21 KB

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