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.

examples/system/ota/advanced_https_ota/main/Kconfig.projbuild

@@ -54,6 +54,13 @@ menu "Example Configuration"
             This option allows one to configure the OTA process to resume downloading the OTA image
             from where it left off in case of an error or reboot.
 
+    config EXAMPLE_OTA_BUF_SIZE
+        int "OTA buffer size"
+        default 1024
+        range 1024 65536
+        help
+            Buffer size for OTA data transfers in bytes.
+
     config EXAMPLE_USE_CERT_BUNDLE
         bool "Enable certificate bundle"
         default y

examples/system/ota/advanced_https_ota/main/advanced_https_ota_example.c

@@ -236,6 +236,7 @@ void advanced_ota_example_task(void *pvParameter)
 #ifdef CONFIG_EXAMPLE_ENABLE_PARTIAL_HTTP_DOWNLOAD
         .save_client_session = true,
 #endif
+        .buffer_size = CONFIG_EXAMPLE_OTA_BUF_SIZE,
     };
 
 #ifdef CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL_FROM_STDIN

examples/system/ota/advanced_https_ota/pytest_advanced_ota.py

@@ -273,6 +273,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:

examples/system/ota/advanced_https_ota/sdkconfig.ci.ota_resumption_flash_enc

@@ -0,0 +1,39 @@
+CONFIG_EXAMPLE_FIRMWARE_UPGRADE_URL="FROM_STDIN"
+CONFIG_EXAMPLE_SKIP_COMMON_NAME_CHECK=y
+CONFIG_EXAMPLE_SKIP_VERSION_CHECK=y
+CONFIG_EXAMPLE_OTA_RECV_TIMEOUT=15000
+CONFIG_EXAMPLE_ENABLE_OTA_RESUMPTION=y
+
+CONFIG_EXAMPLE_CONNECT_ETHERNET=y
+CONFIG_EXAMPLE_CONNECT_WIFI=n
+CONFIG_ETHERNET_INTERNAL_SUPPORT=y
+CONFIG_ETHERNET_PHY_GENERIC=y
+CONFIG_ETHERNET_MDC_GPIO=23
+CONFIG_ETHERNET_MDIO_GPIO=18
+CONFIG_ETHERNET_PHY_RST_GPIO=5
+CONFIG_ETHERNET_PHY_ADDR=1
+CONFIG_EXAMPLE_CONNECT_IPV6=y
+CONFIG_ETHERNET_RX_TASK_STACK_SIZE=3072
+CONFIG_PARTITION_TABLE_OFFSET=0xd000
+
+CONFIG_MBEDTLS_TLS_CLIENT_ONLY=y
+CONFIG_COMPILER_OPTIMIZATION_SIZE=y
+CONFIG_EXAMPLE_CONNECT_IPV6=n
+
+# Default settings for testing this example in CI.
+# This configuration is not secure, don't use it in production!
+# See Flash Encryption API Guide for more details.
+
+CONFIG_SECURE_FLASH_ENC_ENABLED=y
+CONFIG_SECURE_FLASH_ENCRYPTION_MODE_DEVELOPMENT=y
+CONFIG_SECURE_BOOT_ALLOW_ROM_BASIC=y
+CONFIG_SECURE_BOOT_ALLOW_JTAG=y
+CONFIG_SECURE_FLASH_UART_BOOTLOADER_ALLOW_ENC=y
+CONFIG_SECURE_FLASH_UART_BOOTLOADER_ALLOW_DEC=y
+CONFIG_SECURE_FLASH_UART_BOOTLOADER_ALLOW_CACHE=y
+CONFIG_SECURE_FLASH_REQUIRE_ALREADY_ENABLED=y
+CONFIG_NVS_ENCRYPTION=n  # this test combination is only for flash encryption and ota resumption use-case and hence disabling it.
+CONFIG_NVS_SEC_KEY_PROTECT_USING_FLASH_ENC=y
+
+# Custom OTA buffer size configuration
+CONFIG_EXAMPLE_OTA_BUF_SIZE=2047