twister: refactor DUT class to dataclass for serialization support

Convert the DUT class from a traditional class to a dataclass to enable
proper serialization and support for future multi-device testing with
pytest-harness.

Key changes:
- Migrated DUT class to use `dataclass` decorator with proper type hints
- Renamed `baud` property to `serial_baud` for consistency
- Updated hardware map schema to support both `baud` (legacy) and
`serial_baud` fields for backward compatibility
- Updated tests

Signed-off-by: Grzegorz Chwierut <grzegorz.chwierut@nordicsemi.no>
This commit is contained in:
Grzegorz Chwierut
2026-01-09 17:51:58 +01:00
committed by Henrik Brix Andersen
parent 8340e8c264
commit efc36d96d3
7 changed files with 64 additions and 82 deletions

View File

@@ -765,7 +765,7 @@ class DeviceHandler(Handler):
ser_pty_master, slave = pty.openpty()
serial_device = os.ttyname(slave)
logger.debug(f"Using serial device {serial_device} @ {hardware.baud} baud")
logger.debug(f"Using serial device {serial_device} @ {hardware.serial_baud} baud")
command = self._create_command(runner, hardware)
@@ -787,7 +787,7 @@ class DeviceHandler(Handler):
ser = self._create_serial_connection(
hardware,
serial_port,
hardware.baud,
hardware.serial_baud,
flash_timeout,
serial_pty,
ser_pty_process
@@ -848,7 +848,7 @@ class DeviceHandler(Handler):
try:
if serial_pty:
ser_pty_process = self._start_serial_pty(serial_pty, ser_pty_master)
logger.debug(f"Attach serial device {serial_device} @ {hardware.baud} baud")
logger.debug(f"Attach serial device {serial_device} @ {hardware.serial_baud} baud")
ser.port = serial_device
# Apply ESP32-specific RTS/DTR reset logic

View File

@@ -3,13 +3,16 @@
#
# Copyright (c) 2022 Intel Corporation
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import logging
import os
import platform
import re
from dataclasses import asdict, dataclass, field
from multiprocessing import Lock, Value
from pathlib import Path
from typing import Any
import scl
import yaml
@@ -32,50 +35,39 @@ except ImportError:
logger = logging.getLogger('twister')
@dataclass
class DUT:
def __init__(self,
id=None,
serial=None,
serial_baud=None,
platform=None,
product=None,
serial_pty=None,
connected=False,
runner_params=None,
pre_script=None,
post_script=None,
post_flash_script=None,
script_param=None,
runner=None,
flash_timeout=60,
flash_with_test=False,
flash_before=False):
"""Device Under Test configuration."""
id: str | None = None
serial: str | None = None
serial_baud: int = 115200
platform: str | None = None
product: str | None = None
serial_pty: str | None = None
connected: bool = False
runner_params: str | None = None
pre_script: str | None = None
post_script: str | None = None
post_flash_script: str | None = None
script_param: str | None = None
runner: str | None = None
flash_timeout: int = 60
flash_with_test: bool = False
flash_before: bool = False
fixtures: list[str] = field(default_factory=list)
probe_id: str | None = None
notes: str | None = None
match: bool = False
self.serial = serial
self.baud = serial_baud or 115200
self.platform = platform
self.serial_pty = serial_pty
def __post_init__(self):
"""Initialize non-serializable objects after dataclass initialization."""
# These are not dataclass fields, so they won't be serialized by asdict()
self._counter = Value("i", 0)
self._available = Value("i", 1)
self._failures = Value("i", 0)
self.connected = connected
self.pre_script = pre_script
self.id = id
self.product = product
self.runner = runner
self.runner_params = runner_params
self.flash_before = flash_before
self.fixtures = []
self.post_flash_script = post_flash_script
self.post_script = post_script
self.pre_script = pre_script
self.script_param = script_param
self.probe_id = None
self.notes = None
self.lock = Lock()
self.match = False
self.flash_timeout = flash_timeout
self.flash_with_test = flash_with_test
# Ensure serial_baud has a default value
self.serial_baud = self.serial_baud or 115200
@property
def available(self):
@@ -115,19 +107,16 @@ class DUT:
with self._failures.get_lock():
self._failures.value += value
def to_dict(self):
d = {}
exclude = ['_available', '_counter', '_failures', 'match']
v = vars(self)
for k in v:
if k not in exclude and v[k]:
d[k] = v[k]
return d
def to_dict(self) -> dict[str, Any]:
"""Convert DUT dataclass to dictionary for YAML serialization."""
result = asdict(self)
# Remove None and False values and empty lists to keep YAML clean
return {k: v for k, v in result.items() if v}
def __repr__(self):
return f"<{self.platform} ({self.product}) on {self.serial}>"
class HardwareMap:
schema_path = os.path.join(ZEPHYR_BASE, "scripts", "schemas", "twister", "hwmap-schema.yaml")
@@ -170,7 +159,7 @@ class HardwareMap:
}
def __init__(self, env=None):
self.detected = []
self.detected: list[DUT] = []
self.duts: list[DUT] = []
self.options = env.options
@@ -295,7 +284,7 @@ class HardwareMap:
runner = dut.get('runner')
runner_params = dut.get('runner_params')
serial = dut.get('serial')
baud = dut.get('baud', None)
serial_baud = dut.get('serial_baud', None) or dut.get('baud', None)
product = dut.get('product')
fixtures = dut.get('fixtures', [])
connected = dut.get('connected') and ((serial or serial_pty) is not None)
@@ -309,7 +298,7 @@ class HardwareMap:
id=id,
serial_pty=serial_pty,
serial=serial,
serial_baud=baud,
serial_baud=serial_baud,
connected=connected,
pre_script=pre_script,
flash_before=flash_before,

View File

@@ -463,7 +463,7 @@ class Pytest(Harness):
else:
command.extend([
f'--device-serial={hardware.serial}',
f'--device-serial-baud={hardware.baud}'
f'--device-serial-baud={hardware.serial_baud}'
])
for extra_serial in handler.get_more_serials_from_device(hardware):
command.append(f'--device-serial={extra_serial}')

View File

@@ -41,6 +41,9 @@ sequence:
"baud":
type: int
required: false
"serial_baud":
type: int
required: false
"post_script":
type: str
required: false

View File

@@ -37,7 +37,7 @@ def mocked_hm():
TESTDATA_1 = [
(
{},
{'baud': 115200, 'lock': mock.ANY, 'flash_timeout': 60},
{'serial_baud': 115200, 'flash_timeout': 60},
'<None (None) on None>'
),
(
@@ -63,10 +63,9 @@ TESTDATA_1 = [
}
},
{
'lock': mock.ANY,
'id': 'dummy id',
'serial': 'dummy serial',
'baud': 4400,
'serial_baud': 4400,
'platform': 'dummy platform',
'product': 'dummy product',
'serial_pty': 'dummy serial pty',
@@ -269,7 +268,7 @@ def test_hardwaremap_load():
runner: r0
flash_with_test: True
flash_timeout: 15
baud: 14400
serial_baud: 14400
fixtures:
- dummy fixture 1
- dummy fixture 2
@@ -310,7 +309,7 @@ def test_hardwaremap_load():
'runner': 'r0',
'flash_timeout': 15,
'flash_with_test': True,
'baud': 14400,
'serial_baud': 14400,
'fixtures': ['dummy fixture 1', 'dummy fixture 2'],
'connected': True,
'serial': 'dummy',
@@ -322,7 +321,7 @@ def test_hardwaremap_load():
'runner': 'r1',
'flash_timeout': 30,
'flash_with_test': False,
'baud': 115200,
'serial_baud': 115200,
'fixtures': [],
'connected': True,
'serial': None,
@@ -503,50 +502,45 @@ TESTDATA_5 = [
'',
[{
'serial': 's1',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p1',
'connected': True,
'id': 1,
'product': 'pr1',
'lock': mock.ANY,
'flash_timeout': 60
},
{
'serial': 's2',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p2',
'id': 2,
'product': 'pr2',
'lock': mock.ANY,
'flash_timeout': 60
},
{
'serial': 's3',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p3',
'connected': True,
'id': 3,
'product': 'pr3',
'lock': mock.ANY,
'flash_timeout': 60
},
{
'serial': 's4',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p4',
'id': 4,
'product': 'pr4',
'lock': mock.ANY,
'flash_timeout': 60
},
{
'serial': 's5',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p5',
'connected': True,
'id': 5,
'product': 'pr5',
'lock': mock.ANY,
'flash_timeout': 60
}]
),
@@ -603,41 +597,37 @@ TESTDATA_5 = [
},
{
'serial': 's1',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p1',
'connected': True,
'id': 1,
'product': 'pr1',
'lock': mock.ANY,
'flash_timeout': 60
},
{
'serial': 's2',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p2',
'id': 2,
'product': 'pr2',
'lock': mock.ANY,
'flash_timeout': 60
},
{
'serial': 's3',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p3',
'connected': True,
'id': 3,
'product': 'pr3',
'lock': mock.ANY,
'flash_timeout': 60
},
{
'serial': 's5',
'baud': 115200,
'serial_baud': 115200,
'platform': 'p5',
'connected': True,
'id': 5,
'product': 'pr5',
'lock': mock.ANY,
'flash_timeout': 60
}]
),

View File

@@ -552,7 +552,7 @@ def test_pytest__generate_parameters_for_hardware(tmp_path, pty_value, hardware_
hardware = mock.Mock()
hardware.serial_pty = pty_value
hardware.serial = "serial"
hardware.baud = 115200
hardware.serial_baud = 115200
hardware.runner = "runner"
hardware.runner_params = ["--runner-param1", "runner-param2"]
hardware.fixtures = ["fixture1:option1", "fixture2"]

View File

@@ -112,7 +112,7 @@ class TestHardwaremap:
def test_generate(self, capfd, out_path, manufacturer, product, serial, runner):
file_name = "test-map.yaml"
path = os.path.join(ZEPHYR_BASE, file_name)
args = ['--outdir', out_path, '--generate-hardware-map', file_name]
args = ['--outdir', out_path, '--generate-hardware-map', path]
if os.path.exists(path):
os.remove(path)
@@ -164,7 +164,7 @@ class TestHardwaremap:
def test_few_generate(self, capfd, out_path, manufacturer, product, serial, runner):
file_name = "test-map.yaml"
path = os.path.join(ZEPHYR_BASE, file_name)
args = ['--outdir', out_path, '--generate-hardware-map', file_name]
args = ['--outdir', out_path, '--generate-hardware-map', path]
if os.path.exists(path):
os.remove(path)
@@ -245,7 +245,7 @@ class TestHardwaremap:
def test_texas_exeption(self, capfd, out_path, manufacturer, product, serial, location):
file_name = "test-map.yaml"
path = os.path.join(ZEPHYR_BASE, file_name)
args = ['--outdir', out_path, '--generate-hardware-map', file_name]
args = ['--outdir', out_path, '--generate-hardware-map', path]
if os.path.exists(path):
os.remove(path)