twister: Allow sharing hardware platform between variants

doc/develop/test/twister.rst

@@ -1278,6 +1278,25 @@ using an external J-Link probe.  The ``probe_id`` keyword overrides the
       runner: jlink
       serial: null
 
+Using Single Board For Multiple Variants
+++++++++++++++++++++++++++++++++++++++++
+
+  The ``platform`` attribute can be a list of names or a string
+  with names separated by spaces. This allows to run tests for
+  different platform variants on the same physical board, without
+  re-configuring the hardware map file for each variant. For example:
+
+.. code-block:: yaml
+
+    - connected: true
+      id: '001234567890'
+      platform:
+      - nrf5340dk/nrf5340/cpuapp
+      - nrf5340dk/nrf5340/cpuapp/ns
+      product: J-Link
+      runner: nrfjprog
+      serial: /dev/ttyACM1
+
 Quarantine
 ++++++++++
 

scripts/pylib/twister/twisterlib/handlers.py

@@ -21,6 +21,7 @@
 import threading
 import time
 
+from contextlib import contextmanager
 from pathlib import Path
 from queue import Queue, Empty
 from twisterlib.environment import ZEPHYR_BASE, strip_ansi_sequences
@@ -457,6 +458,17 @@ def monitor_serial(self, ser, halt_event, harness):
 
         log_out_fp.close()
 
+    @staticmethod
+    @contextmanager
+    def acquire_dut_locks(duts):
+        try:
+            for d in duts:
+                d.lock.acquire()
+            yield
+        finally:
+            for d in duts:
+                d.lock.release()
+
     def device_is_available(self, instance):
         device = instance.platform.name
         fixture = instance.testsuite.harness_config.get("fixture")
@@ -474,15 +486,16 @@ def device_is_available(self, instance):
 
         # Select an available DUT with less failures
         for d in sorted(duts_found, key=lambda _dut: _dut.failures):
-            d.lock.acquire()
-            avail = False
-            if d.available:
-                d.available = 0
-                d.counter_increment()
-                avail = True
-                logger.debug(f"Retain DUT:{d.platform}, Id:{d.id}, "
-                             f"counter:{d.counter}, failures:{d.failures}")
-            d.lock.release()
+            duts_shared_hw = [_d for _d in self.duts if _d.id == d.id]  # get all DUTs with the same id
+            with self.acquire_dut_locks(duts_shared_hw):
+                avail = False
+                if d.available:
+                    for _d in duts_shared_hw:
+                        _d.available = 0
+                    d.counter_increment()
+                    avail = True
+                    logger.debug(f"Retain DUT:{d.platform}, Id:{d.id}, "
+                                f"counter:{d.counter}, failures:{d.failures}")
             if avail:
                 return d
 
@@ -493,7 +506,10 @@ def make_dut_available(self, dut):
             dut.failures_increment()
         logger.debug(f"Release DUT:{dut.platform}, Id:{dut.id}, "
                      f"counter:{dut.counter}, failures:{dut.failures}")
-        dut.available = 1
+        duts_shared_hw = [_d for _d in self.duts if _d.id == dut.id]  # get all DUTs with the same id
+        with self.acquire_dut_locks(duts_shared_hw):
+            for _d in duts_shared_hw:
+                _d.available = 1
 
     @staticmethod
     def run_custom_script(script, timeout):

scripts/pylib/twister/twisterlib/hardwaremap.py

@@ -262,36 +262,43 @@ def load(self, map_file):
             flash_before = dut.get('flash_before')
             if flash_before is None:
                 flash_before = self.options.flash_before and (not (flash_with_test or serial_pty))
-            platform  = dut.get('platform')
+            platform = dut.get('platform')
+            if isinstance(platform, str):
+                platforms = platform.split()
+            elif isinstance(platform, list):
+                platforms = platform
+            else:
+                raise ValueError(f"Invalid platform value: {platform}")
             id = dut.get('id')
             runner = dut.get('runner')
             runner_params = dut.get('runner_params')
             serial = dut.get('serial')
             baud = dut.get('baud', None)
             product = dut.get('product')
             fixtures = dut.get('fixtures', [])
-            connected= dut.get('connected') and ((serial or serial_pty) is not None)
+            connected = dut.get('connected') and ((serial or serial_pty) is not None)
             if not connected:
                 continue
-            new_dut = DUT(platform=platform,
-                          product=product,
-                          runner=runner,
-                          runner_params=runner_params,
-                          id=id,
-                          serial_pty=serial_pty,
-                          serial=serial,
-                          serial_baud=baud,
-                          connected=connected,
-                          pre_script=pre_script,
-                          flash_before=flash_before,
-                          post_script=post_script,
-                          post_flash_script=post_flash_script,
-                          script_param=script_param,
-                          flash_timeout=flash_timeout,
-                          flash_with_test=flash_with_test)
-            new_dut.fixtures = fixtures
-            new_dut.counter = 0
-            self.duts.append(new_dut)
+            for plat in platforms:
+                new_dut = DUT(platform=plat,
+                              product=product,
+                              runner=runner,
+                              runner_params=runner_params,
+                              id=id,
+                              serial_pty=serial_pty,
+                              serial=serial,
+                              serial_baud=baud,
+                              connected=connected,
+                              pre_script=pre_script,
+                              flash_before=flash_before,
+                              post_script=post_script,
+                              post_flash_script=post_flash_script,
+                              script_param=script_param,
+                              flash_timeout=flash_timeout,
+                              flash_with_test=flash_with_test)
+                new_dut.fixtures = fixtures
+                new_dut.counter = 0
+                self.duts.append(new_dut)
 
     def scan(self, persistent=False):
         from serial.tools import list_ports

scripts/schemas/twister/hwmap-schema.yaml

@@ -16,7 +16,7 @@ sequence:
         type: str
         required: false
       "platform":
-        type: str
+        type: any
         required: true
       "probe_id":
         type: str

scripts/tests/twister/test_handlers.py

@@ -1275,6 +1275,7 @@ def mock_serial(*args, **kwargs):
     dut = DUT()
     dut.available = 0
     dut.failures = 0
+    handler.duts = [dut]
 
     hardware_baud = 14400
     flash_timeout = 60