actions: Add board provisioning testing

Run provisioning tests on schedule.

Signed-off-by: Jorgen Kvalvaag <[email protected]>

Author: jorgenmk

.github/workflows/build-and-target-test.yml

@@ -36,7 +36,7 @@ jobs:
           PUSH=${{ github.event_name == 'push' }}
           push_memory_badges=false
           if [[ $SCHEDULED == true ]]; then
-            devices='["thingy91x","nrf9151dk","ppk_thingy91x","gnss_nrf9151dk"]'
+            devices='["thingy91x","nrf9151dk","ppk_thingy91x","gnss_nrf9151dk", "prov_thingy91x"]'
             build_all=true
             push_memory_badges=true
           elif [[ $PUSH == true ]]; then

app/prj.conf

@@ -244,3 +244,5 @@ CONFIG_NRF_PROVISIONING_RX_BUF_SZ=4096
 CONFIG_NRF_PROVISIONING_TX_BUF_SZ=4096
 CONFIG_NRF_PROVISIONING_CODEC_AT_CMD_LEN=2048
 CONFIG_NRF_PROVISIONING_CODEC_RX_SZ_START=2048
+
+CONFIG_NRF_PROVISIONING_COAP_TIMEOUT_SECONDS=6

tests/on_target/tests/conftest.py

@@ -23,9 +23,6 @@
 DEVICE_UUID = os.getenv('UUID')
 NRFCLOUD_API_KEY = os.getenv('NRFCLOUD_API_KEY')
 DUT_DEVICE_TYPE = os.getenv('DUT_DEVICE_TYPE')
-if DUT_DEVICE_TYPE == "ppk_thingy91x":
-    DUT_DEVICE_TYPE = "thingy91x"
-
 
 def get_uarts():
     base_path = "/dev/serial/by-id"

tests/on_target/tests/test_provisioning/test_provisioning.py

@@ -1,7 +1,9 @@
 import os
 import sys
 import json
-import requests.exceptions # Used for handling HTTP errors from nRF Cloud API
+import time
+import pytest
+import requests.exceptions  # Used for handling HTTP errors from nRF Cloud API
 
 # Ensure the utils directory is in the Python path
 sys.path.append(os.getcwd())
@@ -16,6 +18,7 @@
 
 # --- Helper Functions ---
 
+
 def _perform_initial_device_setup_and_factory_reset(dut_cloud, hex_file: str):
     logger.info(f"Flashing device with {hex_file} and performing factory reset.")
 
@@ -24,40 +27,57 @@ def _perform_initial_device_setup_and_factory_reset(dut_cloud, hex_file: str):
     dut_cloud.uart.flush()
     reset_device()
 
+
 def _wait_for_lte_connection(dut_cloud, timeout: int = 240):
     logger.info("Waiting for device to connect to LTE network...")
 
     log_pattern_network_connected = "network: Network connectivity established"
-    dut_cloud.uart.wait_for_str(log_pattern_network_connected, timeout=timeout)
+    dut_cloud.uart.wait_for_str(
+        log_pattern_network_connected,
+        timeout=timeout,
+        start_pos=dut_cloud.uart.get_size(),
+    )
 
     logger.info("Device connected to LTE network.")
 
+
 def _disconnect_network_and_clear_modem_credentials(dut_cloud, sec_tag: int):
     logger.info("Disconnecting network and clearing modem credentials...")
 
     log_pattern_network_disconnected = "network: Network connectivity lost"
     dut_cloud.uart.write("att_network disconnect\r\n")
-    dut_cloud.uart.wait_for_str(log_pattern_network_disconnected, timeout=20)
+    dut_cloud.uart.wait_for_str(
+        log_pattern_network_disconnected,
+        timeout=20,
+        start_pos=dut_cloud.uart.get_size(),
+    )
 
     # Clear any existing credentials for the given security tag
     dut_cloud.uart.write(f"at AT%CMNG=3,{sec_tag},0\r\n")
+    time.sleep(1)
     dut_cloud.uart.write(f"at AT%CMNG=3,{sec_tag},1\r\n")
+    time.sleep(1)
     dut_cloud.uart.write(f"at AT%CMNG=3,{sec_tag},2\r\n")
+    time.sleep(1)
 
     logger.info("Modem credentials cleared.")
 
+
 def _get_attestation_token_from_device(dut_cloud) -> str:
     logger.info("Getting attestation token from device...")
 
     dut_cloud.uart.at_cmd_write("at AT%ATTESTTOKEN\r\n")
-    token_match = dut_cloud.uart.wait_for_str_re(r'%ATTESTTOKEN: "([^"]+)"', timeout=20)
+    token_match = dut_cloud.uart.wait_for_str_re(
+        r'%ATTESTTOKEN: "([^"]+)"', timeout=20
+    )
 
     assert token_match, "No attestation token found"
     attestation_token = token_match[0]
     logger.info(f"Attestation Token: {attestation_token}")
 
     return attestation_token
 
+
 def _unclaim_device_from_nrf_cloud_if_exists(dut_cloud):
     """
     Attempts to unclaim the device from nRF Cloud.
@@ -70,42 +90,75 @@ def _unclaim_device_from_nrf_cloud_if_exists(dut_cloud):
         logger.info(f"Unclaim device status_code: {status_code}")
     except requests.exceptions.HTTPError as e:
         if e.response.status_code == 404:
-            logger.info(f"Device {dut_cloud.device_id} not found or already unclaimed (404), proceeding.")
+            logger.info(
+                f"Device {dut_cloud.device_id} not found or already unclaimed (404), proceeding."
+            )
         else:
             logger.error(f"Error unclaiming device: {e}")
-            raise # Re-raise other HTTP errors
+            raise  # Re-raise other HTTP errors
+
 
 def _connect_to_network_and_wait_for_claiming_prompt(dut_cloud):
     logger.info("Connecting to network and waiting for device to request claiming...")
 
     log_pattern_network_connected = "network: Network connectivity established"
-    log_pattern_need_claiming = "Claim the device using the device's attestation token on nrfcloud.com"
+    log_pattern_need_claiming = (
+        "Claim the device using the device's attestation token on nrfcloud.com"
+    )
 
     dut_cloud.uart.write("att_network connect\r\n")
-    dut_cloud.uart.wait_for_str(log_pattern_network_connected, timeout=240)
-    dut_cloud.uart.wait_for_str(log_pattern_need_claiming, timeout=240)
+    dut_cloud.uart.wait_for_str(
+        log_pattern_network_connected, timeout=240, start_pos=dut_cloud.uart.get_size()
+    )
+    dut_cloud.uart.wait_for_str(
+        log_pattern_need_claiming, timeout=240, start_pos=dut_cloud.uart.get_size()
+    )
 
     logger.info("Device is ready to be claimed.")
 
+
 def _claim_device_on_nrf_cloud(dut_cloud, attestation_token: str):
     logger.info("Claiming device on nRF Cloud...")
 
     dut_cloud.cloud.claim_device(attestation_token=attestation_token)
 
     logger.info("Device claimed successfully.")
 
-def _wait_for_provisioning_completion_and_cloud_connection(dut_cloud, timeout: int = 240):
-    logger.info("Waiting for provisioning to complete and device to connect to nRF Cloud...")
 
-    dut_cloud.uart.wait_for_str("cloud: Provisioning finished", timeout=timeout)
-    dut_cloud.uart.wait_for_str("cloud: Connected to Cloud", timeout=timeout)
+def _wait_for_provisioning_completion_and_cloud_connection(
+    dut_cloud, timeout: int = 240
+):
+    logger.info(
+        "Waiting for provisioning to complete and device to connect to nRF Cloud..."
+    )
+
+    dut_cloud.uart.wait_for_str(
+        [
+            "cloud: nrf_provisioning_callback: Provisioning finished",
+            "cloud: Connected to Cloud",
+        ],
+        timeout=timeout,
+        start_pos=dut_cloud.uart.get_size(),
+    )
 
     logger.info("Device provisioned and connected to nRF Cloud.")
 
-def _verify_device_location_data(dut_cloud, expected_lat: float = 61.5, expected_lon: float = 10.5, lat_tolerance: float = 2.0, lon_tolerance: float = 1.0, timeout: int = 300):
+
+def _verify_device_location_data(
+    dut_cloud,
+    expected_lat: float = 61.5,
+    expected_lon: float = 10.5,
+    lat_tolerance: float = 2.0,
+    lon_tolerance: float = 1.0,
+    timeout: int = 300,
+):
     logger.info("Verifying device location data...")
 
-    values = dut_cloud.uart.wait_for_str_re(r'location_event_handler: Got location: lat: ([\d.-]+), lon: ([\d.-]+), acc: ([\d.-]+), method:', timeout=timeout)
+    values = dut_cloud.uart.wait_for_str_re(
+        r"location_event_handler: Got location: lat: ([\d.-]+), lon: ([\d.-]+), acc: ([\d.-]+),method:",
+        timeout=timeout,
+        start_pos=dut_cloud.uart.get_size(),
+    )
     assert values, "Failed to get location data from device."
 
     lat_str, lon_str, acc_str = values
@@ -117,6 +170,7 @@ def _verify_device_location_data(dut_cloud, expected_lat: float = 61.5, expected
 
     logger.info("Device location data verified.")
 
+
 def _trigger_device_reprovisioning_with_new_credentials(dut_cloud, sec_tag: int):
     """
     Initiates the reprovisioning process on the device by:
@@ -127,26 +181,25 @@ def _trigger_device_reprovisioning_with_new_credentials(dut_cloud, sec_tag: int)
     logger.info("Starting reprovisioning process with new credentials...")
 
     # Prepare the reprovisioning command for nRF Cloud
-    command_json = json.dumps({
-        "description": "Reprovisioning with new cloud credentials",
-        "request": {
-            "cloudAccessKeyGeneration": {
-                "secTag": sec_tag
-            }
+    command_json = json.dumps(
+        {
+            "description": "Reprovisioning with new cloud credentials",
+            "request": {"cloudAccessKeyGeneration": {"secTag": sec_tag}},
         }
-    })
+    )
 
     logger.info(f"Adding reprovisioning command to nRF Cloud: {command_json}")
 
     dut_cloud.cloud.add_provisioning_command(
-        device_id=dut_cloud.device_id,
-        command=command_json
+        device_id=dut_cloud.device_id, command=command_json
     )
 
     # Update device shadow to indicate a new provisioning command is available
     logger.info("Updating device shadow to trigger reprovisioning command processing.")
 
-    dut_cloud.cloud.patch_delete_command_entry_from_shadow(device_id=dut_cloud.device_id)
+    dut_cloud.cloud.patch_delete_command_entry_from_shadow(
+        device_id=dut_cloud.device_id
+    )
     dut_cloud.cloud.patch_add_provisioning_command_to_shadow(
         device_id=dut_cloud.device_id,
         command=1,  # Command ID 1 signifies a provisioning request
@@ -157,6 +210,7 @@ def _trigger_device_reprovisioning_with_new_credentials(dut_cloud, sec_tag: int)
 
     dut_cloud.uart.write("att_button_press 1\r\n")
 
+
 def _trigger_device_reprovisioning_expecting_no_commands(dut_cloud):
     """
     Initiates a reprovisioning check on the device when no new commands are expected from nRF Cloud.
@@ -167,7 +221,9 @@ def _trigger_device_reprovisioning_expecting_no_commands(dut_cloud):
     # Update device shadow to indicate a provisioning check (even if no commands are queued)
     logger.info("Updating device shadow to trigger provisioning check.")
 
-    dut_cloud.cloud.patch_delete_command_entry_from_shadow(device_id=dut_cloud.device_id)
+    dut_cloud.cloud.patch_delete_command_entry_from_shadow(
+        device_id=dut_cloud.device_id
+    )
     dut_cloud.cloud.patch_add_provisioning_command_to_shadow(
         device_id=dut_cloud.device_id,
         command=1,  # Command ID 1 typically signifies a provisioning request
@@ -178,8 +234,10 @@ def _trigger_device_reprovisioning_expecting_no_commands(dut_cloud):
 
     dut_cloud.uart.write("att_button_press 1\r\n")
 
+
 # --- Test Phases ---
 
+
 def _run_initial_provisioning(dut_cloud, hex_file):
     logger.info("--- Starting Phase 1: Initial Device Provisioning ---")
 
@@ -195,6 +253,7 @@ def _run_initial_provisioning(dut_cloud, hex_file):
 
     logger.info("--- Phase 1: Initial Device Provisioning Completed Successfully ---")
 
+
 def _run_reprovisioning(dut_cloud):
     logger.info("--- Starting Phase 2: Reprovisioning with New Credentials ---")
 
@@ -205,7 +264,10 @@ def _run_reprovisioning(dut_cloud):
     _wait_for_provisioning_completion_and_cloud_connection(dut_cloud, timeout=300)
     _verify_device_location_data(dut_cloud)
 
-    logger.info("--- Phase 2: Reprovisioning with New Credentials Completed Successfully ---")
+    logger.info(
+        "--- Phase 2: Reprovisioning with New Credentials Completed Successfully ---"
+    )
+
 
 def _run_reprovisioning_expecting_no_commands(dut_cloud):
     logger.info("--- Starting Phase 3: Reprovisioning Expecting No Commands ---")
@@ -214,14 +276,22 @@ def _run_reprovisioning_expecting_no_commands(dut_cloud):
 
     _trigger_device_reprovisioning_expecting_no_commands(dut_cloud)
     # Verify the device correctly handles the absence of new provisioning commands
-    dut_cloud.uart.wait_for_str("cloud: No commands from the nRF Provisioning Service to process", timeout=300)
-    dut_cloud.uart.wait_for_str("cloud: Connected to Cloud", timeout=240) # Ensure it reconnects
+    dut_cloud.uart.wait_for_str(
+        [
+            "cloud: No commands from the nRF Provisioning Service to process",
+            "cloud: Connected to Cloud",
+        ],
+        timeout=300,
+        start_pos=dut_cloud.uart.get_size(),
+    )
+    logger.info(
+        "--- Phase 3: Reprovisioning Expecting No Commands Completed Successfully ---"
+    )
 
-    logger.info("--- Phase 3: Reprovisioning Expecting No Commands Completed Successfully ---")
 
 # --- Main Test ---
-
-def test_device_provisioning(dut_cloud, hex_file):
+@pytest.mark.parametrize("_", range(1, 3))
+def test_device_provisioning(_, dut_cloud, hex_file):
     """
     Tests the full device provisioning and reprovisioning lifecycle:
     1. Initial provisioning: Flashes, gets attestation token, claims on nRF Cloud, connects.
@@ -231,12 +301,6 @@ def test_device_provisioning(dut_cloud, hex_file):
        connects to nRF Cloud Provisioning Service, but finds no new commands to process and establishes
        a connection without reprovisioning.
     """
-
-    for i in range(1, 21):
-        logger.info(f"--- Starting Test Iteration: {i}/20 ---")
-        _run_initial_provisioning(dut_cloud, hex_file)
-        _run_reprovisioning(dut_cloud)
-        _run_reprovisioning_expecting_no_commands(dut_cloud)
-        logger.info(f"--- Completed Test Iteration: {i}/20 ---")
-
-    logger.info("Device provisioning and reprovisioning test completed successfully.")
+    _run_initial_provisioning(dut_cloud, hex_file)
+    _run_reprovisioning(dut_cloud)
+    _run_reprovisioning_expecting_no_commands(dut_cloud)

tests/on_target/utils/logger.py

@@ -21,7 +21,7 @@
 
 def get_logger(log_level = LOG_LEVEL):
     caller = inspect.stack()[1]
-    filename = caller.filename
+    filename = caller.filename.split(".")[0]
 
     # Global Logging Function
     logger = logging.getLogger(filename)
@@ -34,7 +34,7 @@ def get_logger(log_level = LOG_LEVEL):
         console.setLevel(log_level)
         logger.addHandler(console)
 
-        formatter = '%(asctime)s - %(filename)s - %(levelname)s - %(message)s'
+        formatter = '%(asctime)s:%(filename)s:%(levelname)s:%(message)s'
 
         formatter = ColoredFormatter(formatter, datefmt='%H:%M:%S')
         console.setFormatter(formatter)
@@ -45,7 +45,7 @@ def get_logger(log_level = LOG_LEVEL):
             file_handler = logging.FileHandler(f"{LOG_DIR}/{LOG_FILENAME}_{level}.txt")
             file_handler.setLevel(level.upper())
             logger.addHandler(file_handler)
-            formatter = '%(asctime)s - %(filename)s - %(levelname)s - %(message)s'
+            formatter = '%(asctime)s:%(filename)s:%(levelname)s:%(message)s'
             file_handler.setFormatter(logging.Formatter(formatter, datefmt='%Y-%m-%d %H:%M:%S'))
 
     logger.setLevel(logging.DEBUG)