example_test.py 21 KB

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