feat: added testcase to check ota resumption if FE is enabled

This commit also added config option to set default buffer size for OTA. This testcase mainly checks if OTA resumption fails if data written is not 16 byte aligned when Flash Encryption is enabled.
2025-11-05 17:38:51 +05:30
parent 88fe9e436d
commit ab8f49823b
4 changed files with 114 additions and 0 deletions
@@ -281,6 +281,75 @@ def test_examples_protocol_advanced_https_ota_example_ota_resumption(dut: Dut) -
        thread1.terminate()


+@pytest.mark.flash_encryption_ota
+@pytest.mark.parametrize('config', ['ota_resumption_flash_enc'], indirect=True)
+@pytest.mark.parametrize('skip_autoflash', ['y'], indirect=True)
+@idf_parametrize('target', ['esp32'], indirect=['target'])
+def test_examples_protocol_advanced_https_ota_example_ota_resumption_flash_enc(dut: Dut) -> None:
+    """
+    This is a positive test case, which stops the download midway and resumes downloading again with
+    flash encryption enabled.
+    steps: |
+      1. join AP/Ethernet
+      2. Fetch OTA image over HTTPS
+      3. Reboot with the new OTA image
+    """
+    # Number of iterations to validate OTA
+    server_port = 8001
+    bin_name = 'advanced_https_ota.bin'
+
+    # For flash encryption, we need to manually erase and flash
+    dut.serial.erase_flash()
+    dut.serial.flash()
+
+    # Erase NVS partition
+    dut.serial.erase_partition(NVS_PARTITION)
+
+    # Start server
+    thread1 = multiprocessing.Process(target=start_https_server, args=(dut.app.binary_path, '0.0.0.0', server_port))
+    thread1.daemon = True
+    thread1.start()
+    try:
+        # start test
+        dut.expect('Loaded app from partition at offset', timeout=30)
+
+        try:
+            ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=60)[1].decode()
+            print(f'Connected to AP/Ethernet with IP: {ip_address}')
+        except pexpect.exceptions.TIMEOUT:
+            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+
+        dut.expect('Starting Advanced OTA example', timeout=30)
+        host_ip = get_host_ip4_by_dest_ip(ip_address)
+
+        print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
+        dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
+        dut.expect('Starting OTA...', timeout=60)
+
+        restart_device_with_random_delay(dut, 5, 15)
+
+        # Validate that the device restarts correctly
+        dut.expect('Loaded app from partition at offset', timeout=180)
+
+        try:
+            ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=60)[1].decode()
+            print(f'Connected to AP/Ethernet with IP: {ip_address}')
+        except pexpect.exceptions.TIMEOUT:
+            raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
+
+        dut.expect('Starting Advanced OTA example', timeout=30)
+        host_ip = get_host_ip4_by_dest_ip(ip_address)
+
+        print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port) + '/' + bin_name))
+        dut.write('https://' + host_ip + ':' + str(server_port) + '/' + bin_name)
+        dut.expect('Starting OTA...', timeout=60)
+
+        dut.expect('upgrade successful. Rebooting ...', timeout=150)
+
+    finally:
+        thread1.terminate()
+
+
@pytest.mark.ethernet_ota
@idf_parametrize('target', ['esp32'], indirect=['target'])
 def test_examples_protocol_advanced_https_ota_example_truncated_bin(dut: Dut) -> None: