[RFC v1 03/10] roadtest: add framework
From: Vincent Whitchurch
Date: Fri Mar 11 2022 - 11:25:47 EST
Add the bulk of the roadtest framework. Apart from one init shell
script, this is written in Python and includes three closely-related
parts:
- The test runner which is invoked from the command line by the user
and which starts the backend and sends the test jobs and results
to/from UML.
- Test support code which is used by the actual driver tests run inside
UML and which interact with the backend via a file-based asynchronous
communication method.
- The backend which is run by the Python interpreter embedded in the C
backend. This part runs the hardware models and is controlled by the
tests and the driver (via virtio in the C backend).
Some unit tests for the framework itself are included and these will be
automatically run whenever the driver tests are run.
Signed-off-by: Vincent Whitchurch <vincent.whitchurch@xxxxxxxx>
---
tools/testing/roadtest/init.sh | 19 ++
tools/testing/roadtest/roadtest/__init__.py | 2 +
.../roadtest/roadtest/backend/__init__.py | 0
.../roadtest/roadtest/backend/backend.py | 32 ++
.../testing/roadtest/roadtest/backend/gpio.py | 111 +++++++
.../testing/roadtest/roadtest/backend/i2c.py | 123 ++++++++
.../testing/roadtest/roadtest/backend/main.py | 13 +
.../testing/roadtest/roadtest/backend/mock.py | 20 ++
.../roadtest/roadtest/backend/test_gpio.py | 98 ++++++
.../roadtest/roadtest/backend/test_i2c.py | 84 +++++
.../testing/roadtest/roadtest/cmd/__init__.py | 0
tools/testing/roadtest/roadtest/cmd/main.py | 146 +++++++++
tools/testing/roadtest/roadtest/cmd/remote.py | 48 +++
.../roadtest/roadtest/core/__init__.py | 0
.../testing/roadtest/roadtest/core/control.py | 52 ++++
.../roadtest/roadtest/core/devicetree.py | 155 ++++++++++
.../roadtest/roadtest/core/hardware.py | 94 ++++++
tools/testing/roadtest/roadtest/core/log.py | 42 +++
.../testing/roadtest/roadtest/core/modules.py | 38 +++
.../testing/roadtest/roadtest/core/opslog.py | 35 +++
tools/testing/roadtest/roadtest/core/proxy.py | 48 +++
tools/testing/roadtest/roadtest/core/suite.py | 286 ++++++++++++++++++
tools/testing/roadtest/roadtest/core/sysfs.py | 77 +++++
.../roadtest/roadtest/core/test_control.py | 35 +++
.../roadtest/roadtest/core/test_devicetree.py | 31 ++
.../roadtest/roadtest/core/test_hardware.py | 41 +++
.../roadtest/roadtest/core/test_log.py | 54 ++++
.../roadtest/roadtest/core/test_opslog.py | 27 ++
.../roadtest/roadtest/tests/__init__.py | 0
29 files changed, 1711 insertions(+)
create mode 100755 tools/testing/roadtest/init.sh
create mode 100644 tools/testing/roadtest/roadtest/__init__.py
create mode 100644 tools/testing/roadtest/roadtest/backend/__init__.py
create mode 100644 tools/testing/roadtest/roadtest/backend/backend.py
create mode 100644 tools/testing/roadtest/roadtest/backend/gpio.py
create mode 100644 tools/testing/roadtest/roadtest/backend/i2c.py
create mode 100644 tools/testing/roadtest/roadtest/backend/main.py
create mode 100644 tools/testing/roadtest/roadtest/backend/mock.py
create mode 100644 tools/testing/roadtest/roadtest/backend/test_gpio.py
create mode 100644 tools/testing/roadtest/roadtest/backend/test_i2c.py
create mode 100644 tools/testing/roadtest/roadtest/cmd/__init__.py
create mode 100644 tools/testing/roadtest/roadtest/cmd/main.py
create mode 100644 tools/testing/roadtest/roadtest/cmd/remote.py
create mode 100644 tools/testing/roadtest/roadtest/core/__init__.py
create mode 100644 tools/testing/roadtest/roadtest/core/control.py
create mode 100644 tools/testing/roadtest/roadtest/core/devicetree.py
create mode 100644 tools/testing/roadtest/roadtest/core/hardware.py
create mode 100644 tools/testing/roadtest/roadtest/core/log.py
create mode 100644 tools/testing/roadtest/roadtest/core/modules.py
create mode 100644 tools/testing/roadtest/roadtest/core/opslog.py
create mode 100644 tools/testing/roadtest/roadtest/core/proxy.py
create mode 100644 tools/testing/roadtest/roadtest/core/suite.py
create mode 100644 tools/testing/roadtest/roadtest/core/sysfs.py
create mode 100644 tools/testing/roadtest/roadtest/core/test_control.py
create mode 100644 tools/testing/roadtest/roadtest/core/test_devicetree.py
create mode 100644 tools/testing/roadtest/roadtest/core/test_hardware.py
create mode 100644 tools/testing/roadtest/roadtest/core/test_log.py
create mode 100644 tools/testing/roadtest/roadtest/core/test_opslog.py
create mode 100644 tools/testing/roadtest/roadtest/tests/__init__.py
diff --git a/tools/testing/roadtest/init.sh b/tools/testing/roadtest/init.sh
new file mode 100755
index 000000000000..c5fb28478aa3
--- /dev/null
+++ b/tools/testing/roadtest/init.sh
@@ -0,0 +1,19 @@
+#!/bin/sh
+# SPDX-License-Identifier: GPL-2.0-only
+
+mount -t proc proc /proc
+echo 8 > /proc/sys/kernel/printk
+mount -t sysfs nodev /sys
+mount -t debugfs nodev /sys/kernel/debug
+
+echo 0 > /sys/bus/i2c/drivers_autoprobe
+echo 0 > /sys/bus/platform/drivers_autoprobe
+
+python3 -m roadtest.cmd.remote
+status=$?
+[ "${ROADTEST_SHELL}" = "1" ] || {
+ # rsync doesn't handle these zero-sized files correctly.
+ cp -ra --no-preserve=ownership /sys/kernel/debug/gcov ${ROADTEST_WORK_DIR}/gcov
+ echo o > /proc/sysrq-trigger
+}
+exec setsid sh -c 'exec bash </dev/tty0 >/dev/tty0 2>&1'
diff --git a/tools/testing/roadtest/roadtest/__init__.py b/tools/testing/roadtest/roadtest/__init__.py
new file mode 100644
index 000000000000..dac3ce6976e5
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/__init__.py
@@ -0,0 +1,2 @@
+ENV_WORK_DIR = "ROADTEST_WORK_DIR"
+ENV_BUILD_DIR = "ROADTEST_BUILD_DIR"
diff --git a/tools/testing/roadtest/roadtest/backend/__init__.py b/tools/testing/roadtest/roadtest/backend/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tools/testing/roadtest/roadtest/backend/backend.py b/tools/testing/roadtest/roadtest/backend/backend.py
new file mode 100644
index 000000000000..bfd19fc363c2
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/backend/backend.py
@@ -0,0 +1,32 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import logging
+import os
+from pathlib import Path
+
+from roadtest import ENV_WORK_DIR
+from roadtest.core.control import ControlReader
+
+from . import gpio, i2c, mock
+
+logger = logging.getLogger(__name__)
+
+try:
+ import cbackend # type: ignore[import]
+except ModuleNotFoundError:
+ # In unit tests
+ cbackend = None
+
+
+class Backend:
+ def __init__(self) -> None:
+ work = Path(os.environ[ENV_WORK_DIR])
+ self.control = ControlReader(work_dir=work)
+ self.c = cbackend
+ self.i2c = i2c.I2CBackend(self)
+ self.gpio = gpio.GpioBackend(self)
+ self.mock = mock.MockBackend(work)
+
+ def process_control(self) -> None:
+ self.control.process({"backend": self})
diff --git a/tools/testing/roadtest/roadtest/backend/gpio.py b/tools/testing/roadtest/roadtest/backend/gpio.py
new file mode 100644
index 000000000000..2eaf52b31c72
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/backend/gpio.py
@@ -0,0 +1,111 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import logging
+import typing
+from typing import Optional
+
+if typing.TYPE_CHECKING:
+ # Avoid circular imports
+ from .backend import Backend
+
+logger = logging.getLogger(__name__)
+
+
+class Gpio:
+ IRQ_TYPE_NONE = 0x00
+ IRQ_TYPE_EDGE_RISING = 0x01
+ IRQ_TYPE_EDGE_FALLING = 0x02
+ IRQ_TYPE_EDGE_BOTH = 0x03
+ IRQ_TYPE_LEVEL_HIGH = 0x04
+ IRQ_TYPE_LEVEL_LOW = 0x08
+
+ def __init__(self, backend: "Backend", pin: int):
+ self.backend = backend
+ self.pin = pin
+ self.state = False
+ self.irq_type = Gpio.IRQ_TYPE_NONE
+ self.masked = True
+ self.edge_irq_latched = False
+
+ def _level_irq_active(self) -> bool:
+ if self.irq_type == Gpio.IRQ_TYPE_LEVEL_HIGH:
+ return self.state
+ elif self.irq_type == Gpio.IRQ_TYPE_LEVEL_LOW:
+ return not self.state
+
+ return False
+
+ def _latch_edge_irq(self, old: bool, new: bool) -> bool:
+ if old != new:
+ logger.debug(f"{self}: latch_edge_irq {self.irq_type} {old} -> {new}")
+
+ if self.irq_type == Gpio.IRQ_TYPE_EDGE_RISING:
+ return not old and new
+ elif self.irq_type == Gpio.IRQ_TYPE_EDGE_FALLING:
+ return old and not new
+ elif self.irq_type == Gpio.IRQ_TYPE_EDGE_BOTH:
+ return old != new
+
+ return False
+
+ def _check_irq(self) -> None:
+ if self.irq_type == Gpio.IRQ_TYPE_NONE or self.masked:
+ return
+ if not self.edge_irq_latched and not self._level_irq_active():
+ return
+
+ self.masked = True
+ self.edge_irq_latched = False
+
+ logger.debug(f"{self}: trigger irq")
+ self.backend.c.trigger_gpio_irq(self.pin)
+
+ def set_irq_type(self, irq_type: int) -> None:
+ logger.debug(f"{self}: set_irq_type {irq_type}")
+ if irq_type == Gpio.IRQ_TYPE_NONE:
+ self.masked = True
+
+ self.irq_type = irq_type
+ self.edge_irq_latched = False
+ self._check_irq()
+
+ def unmask(self) -> None:
+ logger.debug(f"{self}: unmask")
+ self.masked = False
+ self._check_irq()
+
+ def set(self, val: int) -> None:
+ old = self.state
+ new = bool(val)
+
+ if old != new:
+ logger.debug(f"{self}: gpio set {old} -> {new}")
+
+ self.state = new
+ if self._latch_edge_irq(old, new):
+ logger.debug(f"{self}: latching edge")
+ self.edge_irq_latched = True
+
+ self._check_irq()
+
+ def __str__(self) -> str:
+ return f"Gpio({self.pin})"
+
+
+class GpioBackend:
+ def __init__(self, backend: "Backend") -> None:
+ self.backend = backend
+ self.gpios = [Gpio(backend, pin) for pin in range(64)]
+
+ def set(self, pin: Optional[int], val: bool) -> None:
+ if pin is None:
+ return
+
+ self.gpios[pin].set(val)
+
+ def set_irq_type(self, pin: int, irq_type: int) -> None:
+ self.gpios[pin].set_irq_type(irq_type)
+
+ def unmask(self, pin: int) -> None:
+ self.gpios[pin].unmask()
diff --git a/tools/testing/roadtest/roadtest/backend/i2c.py b/tools/testing/roadtest/roadtest/backend/i2c.py
new file mode 100644
index 000000000000..b877c2b76851
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/backend/i2c.py
@@ -0,0 +1,123 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import abc
+import importlib
+import logging
+import typing
+from typing import Any, Literal, Optional
+
+if typing.TYPE_CHECKING:
+ # Avoid circular imports
+ from .backend import Backend
+
+logger = logging.getLogger(__name__)
+
+
+class I2CBackend:
+ def __init__(self, backend: "Backend") -> None:
+ self.model: Optional[I2CModel] = None
+ self.backend = backend
+
+ def load_model(self, modname: str, clsname: str, *args: Any, **kwargs: Any) -> None:
+ mod = importlib.import_module(modname)
+ cls = getattr(mod, clsname)
+ self.model = cls(*args, **kwargs, backend=self.backend)
+
+ def unload_model(self) -> None:
+ self.model = None
+
+ def read(self, length: int) -> bytes:
+ if not self.model:
+ raise Exception("No I2C model loaded")
+
+ return self.model.read(length)
+
+ def write(self, data: bytes) -> None:
+ if not self.model:
+ raise Exception("No I2C model loaded")
+
+ self.model.write(data)
+
+ def __getattr__(self, name: str) -> Any:
+ return getattr(self.model, name)
+
+
+class I2CModel(abc.ABC):
+ def __init__(self, backend: "Backend") -> None:
+ self.backend = backend
+
+ @abc.abstractmethod
+ def read(self, length: int) -> bytes:
+ return bytes(length)
+
+ @abc.abstractmethod
+ def write(self, data: bytes) -> None:
+ pass
+
+
+class SMBusModel(I2CModel):
+ def __init__(
+ self,
+ regbytes: int,
+ byteorder: Literal["little", "big"] = "little",
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.reg_addr = 0x0
+ self.regbytes = regbytes
+ self.byteorder = byteorder
+
+ @abc.abstractmethod
+ def reg_read(self, addr: int) -> int:
+ return 0
+
+ @abc.abstractmethod
+ def reg_write(self, addr: int, val: int) -> None:
+ pass
+
+ def val_to_bytes(self, val: int) -> bytes:
+ return val.to_bytes(self.regbytes, self.byteorder)
+
+ def bytes_to_val(self, data: bytes) -> int:
+ return int.from_bytes(data, self.byteorder)
+
+ def read(self, length: int) -> bytes:
+ data = bytearray()
+ for idx in range(0, length, self.regbytes):
+ addr = self.reg_addr + idx
+ val = self.reg_read(addr)
+ logger.debug(f"SMBus read {addr=:#02x} {val=:#02x}")
+ data += self.val_to_bytes(val)
+ return bytes(data)
+
+ def write(self, data: bytes) -> None:
+ self.reg_addr = data[0]
+
+ if len(data) > 1:
+ length = len(data) - 1
+ data = data[1:]
+ assert length % self.regbytes == 0
+ for idx in range(0, length, self.regbytes):
+ val = self.bytes_to_val(data[idx : (idx + self.regbytes)])
+ addr = self.reg_addr + idx
+ self.backend.mock.reg_write(addr, val)
+ self.reg_write(addr, val)
+ logger.debug(f"SMBus write {addr=:#02x} {val=:#02x}")
+ elif len(data) == 1:
+ pass
+
+
+class SimpleSMBusModel(SMBusModel):
+ def __init__(self, regs: dict[int, int], **kwargs: Any) -> None:
+ super().__init__(**kwargs)
+ self.regs = regs
+
+ def reg_read(self, addr: int) -> int:
+ val = self.regs[addr]
+ return val
+
+ def reg_write(self, addr: int, val: int) -> None:
+ assert addr in self.regs
+ self.regs[addr] = val
diff --git a/tools/testing/roadtest/roadtest/backend/main.py b/tools/testing/roadtest/roadtest/backend/main.py
new file mode 100644
index 000000000000..25be86ded9ea
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/backend/main.py
@@ -0,0 +1,13 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import logging
+
+import roadtest.backend.backend
+
+logging.basicConfig(
+ format="%(asctime)s - %(levelname)s - %(name)s: %(message)s", level=logging.DEBUG
+)
+
+backend = roadtest.backend.backend.Backend()
+backend.process_control()
diff --git a/tools/testing/roadtest/roadtest/backend/mock.py b/tools/testing/roadtest/roadtest/backend/mock.py
new file mode 100644
index 000000000000..8ce33a6bc0f1
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/backend/mock.py
@@ -0,0 +1,20 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import functools
+from pathlib import Path
+from typing import Any, Callable
+
+from roadtest.core.opslog import OpsLogWriter
+
+
+class MockBackend:
+ def __init__(self, work: Path) -> None:
+ self.opslog = OpsLogWriter(work)
+
+ @functools.cache
+ def __getattr__(self, name: str) -> Callable:
+ def func(*args: Any, **kwargs: Any) -> None:
+ self.opslog.write(f"mock.{name}(*{str(args)}, **{str(kwargs)})")
+
+ return func
diff --git a/tools/testing/roadtest/roadtest/backend/test_gpio.py b/tools/testing/roadtest/roadtest/backend/test_gpio.py
new file mode 100644
index 000000000000..feffe4fb9625
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/backend/test_gpio.py
@@ -0,0 +1,98 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import unittest
+from unittest.mock import MagicMock
+
+from .gpio import Gpio
+
+
+class TestGpio(unittest.TestCase):
+ def test_irq_low(self) -> None:
+ m = MagicMock()
+ gpio = Gpio(backend=m, pin=1)
+
+ gpio.set_irq_type(Gpio.IRQ_TYPE_LEVEL_LOW)
+ m.c.trigger_gpio_irq.assert_not_called()
+
+ gpio.unmask()
+ m.c.trigger_gpio_irq.assert_called_once_with(1)
+ m.c.trigger_gpio_irq.reset_mock()
+
+ gpio.set(True)
+ gpio.unmask()
+ m.c.trigger_gpio_irq.assert_not_called()
+
+ def test_irq_high(self) -> None:
+ m = MagicMock()
+ gpio = Gpio(backend=m, pin=2)
+
+ gpio.set_irq_type(Gpio.IRQ_TYPE_LEVEL_HIGH)
+ gpio.unmask()
+
+ m.c.trigger_gpio_irq.assert_not_called()
+
+ gpio.set(True)
+ m.c.trigger_gpio_irq.assert_called_once_with(2)
+ m.c.trigger_gpio_irq.reset_mock()
+
+ gpio.set(False)
+ gpio.unmask()
+ m.c.trigger_gpio_irq.assert_not_called()
+
+ def test_irq_rising(self) -> None:
+ m = MagicMock()
+ gpio = Gpio(backend=m, pin=63)
+
+ gpio.set_irq_type(Gpio.IRQ_TYPE_EDGE_RISING)
+ gpio.set(False)
+ gpio.set(True)
+
+ m.c.trigger_gpio_irq.assert_not_called()
+ gpio.unmask()
+ m.c.trigger_gpio_irq.assert_called_once_with(63)
+ m.c.trigger_gpio_irq.reset_mock()
+
+ gpio.set(False)
+ gpio.set(True)
+
+ gpio.unmask()
+ m.c.trigger_gpio_irq.assert_called_once()
+
+ def test_irq_falling(self) -> None:
+ m = MagicMock()
+ gpio = Gpio(backend=m, pin=0)
+
+ gpio.set_irq_type(Gpio.IRQ_TYPE_EDGE_FALLING)
+ gpio.unmask()
+ gpio.set(False)
+ gpio.set(True)
+ m.c.trigger_gpio_irq.assert_not_called()
+
+ gpio.set(False)
+ m.c.trigger_gpio_irq.assert_called_once_with(0)
+ m.c.trigger_gpio_irq.reset_mock()
+
+ gpio.set(True)
+ gpio.set(False)
+ gpio.set(True)
+ gpio.unmask()
+ m.c.trigger_gpio_irq.assert_called_once()
+
+ def test_irq_both(self) -> None:
+ m = MagicMock()
+ gpio = Gpio(backend=m, pin=32)
+
+ gpio.set_irq_type(Gpio.IRQ_TYPE_EDGE_BOTH)
+ gpio.unmask()
+ gpio.set(False)
+ gpio.set(True)
+ m.c.trigger_gpio_irq.assert_called_once_with(32)
+
+ gpio.set(False)
+ m.c.trigger_gpio_irq.assert_called_once_with(32)
+ m.c.trigger_gpio_irq.reset_mock()
+
+ gpio.set(True)
+ gpio.unmask()
+ m.c.trigger_gpio_irq.assert_called_once_with(32)
diff --git a/tools/testing/roadtest/roadtest/backend/test_i2c.py b/tools/testing/roadtest/roadtest/backend/test_i2c.py
new file mode 100644
index 000000000000..eda4e1a4b80f
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/backend/test_i2c.py
@@ -0,0 +1,84 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import unittest
+from typing import Any
+from unittest.mock import MagicMock
+
+from .i2c import SimpleSMBusModel, SMBusModel
+
+
+class DummyModel(SMBusModel):
+ def __init__(self, *args: Any, **kwargs: Any) -> None:
+ super().__init__(*args, **kwargs)
+ self.regs: dict[int, int] = {}
+
+ def reg_read(self, addr: int) -> int:
+ return self.regs[addr]
+
+ def reg_write(self, addr: int, val: int) -> None:
+ self.regs[addr] = val
+
+
+class TestSMBusModel(unittest.TestCase):
+ def test_1(self) -> None:
+ m = DummyModel(regbytes=1, backend=MagicMock())
+
+ m.write(bytes([0x12, 0x34]))
+ m.write(bytes([0x13, 0xAB, 0xCD]))
+
+ self.assertEqual(m.regs[0x12], 0x34)
+ self.assertEqual(m.regs[0x13], 0xAB)
+ self.assertEqual(m.regs[0x14], 0xCD)
+
+ m.write(bytes([0x12]))
+ self.assertEqual(m.read(1), bytes([0x34]))
+
+ m.write(bytes([0x12]))
+ self.assertEqual(m.read(3), bytes([0x34, 0xAB, 0xCD]))
+
+ def test_2big(self) -> None:
+ m = DummyModel(regbytes=2, byteorder="big", backend=MagicMock())
+
+ m.write(bytes([0x12, 0x34, 0x56, 0xAB, 0xCD]))
+ self.assertEqual(m.regs[0x12], 0x3456)
+ self.assertEqual(m.regs[0x14], 0xABCD)
+
+ m.write(bytes([0x12]))
+ self.assertEqual(m.read(2), bytes([0x34, 0x56]))
+
+ m.write(bytes([0x14]))
+ self.assertEqual(m.read(2), bytes([0xAB, 0xCD]))
+
+ m.write(bytes([0x12]))
+ self.assertEqual(m.read(4), bytes([0x34, 0x56, 0xAB, 0xCD]))
+
+ def test_2little(self) -> None:
+ m = DummyModel(regbytes=2, byteorder="little", backend=MagicMock())
+
+ m.write(bytes([0x12, 0x34, 0x56, 0xAB, 0xCD]))
+ self.assertEqual(m.regs[0x12], 0x5634)
+ self.assertEqual(m.regs[0x14], 0xCDAB)
+
+ m.write(bytes([0x12]))
+ self.assertEqual(m.read(2), bytes([0x34, 0x56]))
+
+
+class TestSimpleSMBusModel(unittest.TestCase):
+ def test_simple(self) -> None:
+ m = SimpleSMBusModel(
+ regs={0x01: 0x12, 0x02: 0x34},
+ regbytes=1,
+ backend=MagicMock(),
+ )
+ self.assertEqual(m.reg_read(0x01), 0x12)
+ self.assertEqual(m.reg_read(0x02), 0x34)
+
+ m.reg_write(0x01, 0x56)
+ self.assertEqual(m.reg_read(0x01), 0x56)
+ self.assertEqual(m.reg_read(0x02), 0x34)
+
+ with self.assertRaises(Exception):
+ m.reg_write(0x03, 0x00)
+ with self.assertRaises(Exception):
+ m.reg_read(0x03)
diff --git a/tools/testing/roadtest/roadtest/cmd/__init__.py b/tools/testing/roadtest/roadtest/cmd/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tools/testing/roadtest/roadtest/cmd/main.py b/tools/testing/roadtest/roadtest/cmd/main.py
new file mode 100644
index 000000000000..634c27fe795c
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/cmd/main.py
@@ -0,0 +1,146 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import argparse
+import fnmatch
+import sys
+import unittest
+from typing import Optional
+from unittest.suite import TestSuite
+
+assert sys.version_info >= (3, 9), "Python version is too old"
+
+from roadtest.core.suite import UMLSuite, UMLTestCase
+
+
+def make_umlsuite(args: argparse.Namespace) -> UMLSuite:
+ return UMLSuite(
+ timeout=args.timeout,
+ workdir=args.work_dir,
+ builddir=args.build_dir,
+ ksrcdir=args.ksrc_dir,
+ uml_args_pre=args.uml_prepend,
+ uml_args_post=args.uml_append,
+ shell=args.shell,
+ )
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "--timeout",
+ type=int,
+ default=60,
+ help="Timeout (in seconds) for each UML run, 0 to disable",
+ )
+ parser.add_argument("--work-dir", type=str, help="Work directory for UML runs")
+ parser.add_argument("--build-dir", type=str, required=True)
+ parser.add_argument("--ksrc-dir", type=str, required=True)
+ parser.add_argument(
+ "--uml-prepend",
+ nargs="*",
+ default=[],
+ help="Extra arguments to prepend to the UML command (example: gdbserver :1234)",
+ )
+ parser.add_argument(
+ "--uml-append",
+ nargs="*",
+ default=[],
+ help="Extra arguments to append to the UML command (example: trace_event=i2c:* tp_printk)",
+ )
+ parser.add_argument(
+ "--filter",
+ nargs="+",
+ default=[],
+ )
+ parser.add_argument("--shell", action="store_true")
+ parser.add_argument("test", nargs="?", default="roadtest")
+ args = parser.parse_args()
+
+ if args.shell:
+ args.timeout = 0
+
+ if not any(p.startswith("con=") for p in args.uml_append):
+ print(
+ "Error: --shell used but no con= UML argument specified",
+ file=sys.stderr,
+ )
+ sys.exit(1)
+
+ test = args.test
+ test = test.replace("/", ".")
+ test = test.removesuffix(".py")
+ test = test.removesuffix(".")
+
+ loader = unittest.defaultTestLoader
+ suitegroups = loader.discover(test)
+
+ args.filter = [f"*{f}*" for f in args.filter]
+
+ # Backend tests and the like don't need to be run inside UML.
+ localsuite = None
+
+ # For simplicity, we currently run all target tests in one UML instance
+ # since python in UML is slow to start up. This can be revisited if we
+ # want to run several UML instances in parallel.
+ deftargetsuite = None
+ targetsuites = []
+
+ for suites in suitegroups:
+ # unittest can in arbitrarily nest and mix TestCases
+ # and TestSuites, but we expect a fixed hierarchy.
+ assert isinstance(suites, unittest.TestSuite)
+
+ for suite in suites:
+ # assert not isinstance(suite, unittest.TestCase)
+
+ # If the import of a test fails, then suite is a
+ # unittest.loader._FailedTest instead of a suite
+ if not isinstance(suite, unittest.TestSuite):
+ suite = [suite] # type: ignore[assignment]
+
+ # Suite at this level contains one TestCase for each
+ # test method in a particular test class.
+ #
+ # All the test functions for one particular test class
+ # can only be run either in UML or locally, not mixed.
+ destsuite: Optional[TestSuite] = None
+
+ for t in suite: # type: ignore[union-attr]
+ # We don't support suites nested at this level.
+ assert isinstance(t, unittest.TestCase)
+
+ id = t.id()
+ if args.filter and not any(fnmatch.fnmatch(id, f) for f in args.filter):
+ continue
+
+ if isinstance(t, UMLTestCase):
+ if t.run_separately:
+ if not destsuite:
+ destsuite = make_umlsuite(args)
+ targetsuites.append(destsuite)
+ else:
+ if not deftargetsuite:
+ deftargetsuite = make_umlsuite(args)
+ targetsuites.append(deftargetsuite)
+
+ destsuite = deftargetsuite
+ else:
+ if not localsuite:
+ localsuite = TestSuite()
+ destsuite = localsuite
+
+ if destsuite:
+ destsuite.addTest(t)
+
+ tests = unittest.TestSuite()
+ if localsuite:
+ tests.addTest(localsuite)
+ tests.addTests(targetsuites)
+
+ result = unittest.TextTestRunner(verbosity=2).run(tests)
+ sys.exit(not result.wasSuccessful())
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tools/testing/roadtest/roadtest/cmd/remote.py b/tools/testing/roadtest/roadtest/cmd/remote.py
new file mode 100644
index 000000000000..29c3c6d35c65
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/cmd/remote.py
@@ -0,0 +1,48 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import importlib
+import json
+import os
+from pathlib import Path
+from typing import cast
+from unittest import TestSuite, TextTestRunner
+
+from roadtest import ENV_WORK_DIR
+from roadtest.core import proxy
+
+
+def main() -> None:
+ workdir = Path(os.environ[ENV_WORK_DIR])
+ with open(workdir / "tests.json") as f:
+ testinfos = json.load(f)
+
+ suite = TestSuite()
+ for info in testinfos:
+ id = info["id"]
+ *modparts, clsname, method = id.split(".")
+
+ fullname = ".".join(modparts)
+ mod = importlib.import_module(fullname)
+
+ cls = getattr(mod, clsname)
+ test = cls(methodName=method)
+
+ values = info["values"]
+ if values:
+ test.dts.values = values
+
+ suite.addTest(test)
+
+ runner = TextTestRunner(
+ verbosity=0, buffer=False, resultclass=proxy.ProxyTextTestResult
+ )
+ result = cast(proxy.ProxyTextTestResult, runner.run(suite))
+
+ proxyresult = result.to_proxy()
+ with open(workdir / "results.json", "w") as f:
+ json.dump(proxyresult, f)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tools/testing/roadtest/roadtest/core/__init__.py b/tools/testing/roadtest/roadtest/core/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tools/testing/roadtest/roadtest/core/control.py b/tools/testing/roadtest/roadtest/core/control.py
new file mode 100644
index 000000000000..cd74861099b9
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/control.py
@@ -0,0 +1,52 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import logging
+import os
+from pathlib import Path
+from typing import Optional
+
+from roadtest import ENV_WORK_DIR
+
+CONTROL_FILE = "control.txt"
+
+logger = logging.getLogger(__name__)
+
+
+class ControlReader:
+ def __init__(self, work_dir: Optional[Path] = None) -> None:
+ if not work_dir:
+ work_dir = Path(os.environ[ENV_WORK_DIR])
+
+ path = work_dir / CONTROL_FILE
+ path.unlink(missing_ok=True)
+ path.write_text("")
+
+ self.file = path.open("r")
+
+ def process(self, vars: dict) -> None:
+ for line in self.file.readlines():
+ cmd = line.rstrip()
+
+ if cmd.startswith("# "):
+ logger.info(line[2:].rstrip())
+ continue
+
+ logger.debug(cmd)
+ eval(cmd, vars)
+
+
+class ControlWriter:
+ def __init__(self, work_dir: Optional[Path] = None) -> None:
+ if not work_dir:
+ work_dir = Path(os.environ[ENV_WORK_DIR])
+ self.file = (work_dir / CONTROL_FILE).open("a", buffering=1)
+
+ def write_cmd(self, line: str) -> None:
+ self.file.write(line + "\n")
+
+ def write_log(self, line: str) -> None:
+ self.file.write(f"# {line}\n")
+
+ def close(self) -> None:
+ self.file.close()
diff --git a/tools/testing/roadtest/roadtest/core/devicetree.py b/tools/testing/roadtest/roadtest/core/devicetree.py
new file mode 100644
index 000000000000..40876738fb39
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/devicetree.py
@@ -0,0 +1,155 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import enum
+import subprocess
+from pathlib import Path
+from typing import Any, Optional
+
+HEADER = """
+/dts-v1/;
+
+/ {
+ #address-cells = <2>;
+ #size-cells = <2>;
+
+ virtio@0 {
+ compatible = "virtio,uml";
+ socket-path = "WORK/gpio.sock";
+ virtio-device-id = <0x29>;
+
+ gpio: gpio {
+ compatible = "virtio,device29";
+
+ gpio-controller;
+ #gpio-cells = <2>;
+
+ interrupt-controller;
+ #interrupt-cells = <2>;
+ };
+ };
+
+ virtio@1 {
+ compatible = "virtio,uml";
+ socket-path = "WORK/i2c.sock";
+ virtio-device-id = <0x22>;
+
+ i2c: i2c {
+ compatible = "virtio,device22";
+
+ #address-cells = <1>;
+ #size-cells = <0>;
+ };
+ };
+
+ // See Hardware.kick()
+ leds {
+ compatible = "gpio-leds";
+ led0 {
+ gpios = <&gpio 0 0>;
+ };
+ };
+};
+"""
+
+
+class DtVar(enum.Enum):
+ I2C_ADDR = 0
+ GPIO_PIN = 1
+
+
+class DtFragment:
+ def __init__(self, src: str, variables: Optional[dict[str, DtVar]] = None) -> None:
+ self.src = src
+ if not variables:
+ variables = {}
+ self.variables = variables
+ self.values: dict[str, int] = {}
+
+ def apply(self, values: dict[str, Any]) -> str:
+ src = self.src
+
+ for var in self.variables.keys():
+ typ = self.variables[var]
+ val = values[var]
+
+ if typ == DtVar.I2C_ADDR:
+ str = f"{val:02x}"
+ elif typ == DtVar.GPIO_PIN:
+ str = f"{val:d}"
+
+ src = src.replace(f"${var}$", str)
+
+ self.values = values
+ return src
+
+ def __getitem__(self, key: str) -> Any:
+ return self.values[key]
+
+
+class Devicetree:
+ def __init__(self, workdir: Path, ksrcdir: Path) -> None:
+ self.workdir: Path = workdir
+ self.ksrcdir: Path = ksrcdir
+ self.next_i2c_addr: int = 0x1
+ # 0 is used for gpio-leds for Hardware.kick()
+ self.next_gpio_pin: int = 1
+ self.src: str = ""
+
+ def assemble(self, fragments: list[DtFragment]) -> None:
+ parts = []
+ for fragment in fragments:
+ if fragment.values:
+ # Multiple test functions from the same class will use
+ # the same class instance
+ continue
+
+ values = {}
+
+ for var, type in fragment.variables.items():
+ if type == DtVar.I2C_ADDR:
+ values[var] = self.next_i2c_addr
+ self.next_i2c_addr += 1
+ elif type == DtVar.GPIO_PIN:
+ values[var] = self.next_gpio_pin
+ self.next_gpio_pin += 1
+
+ parts.append(fragment.apply(values))
+
+ self.src = "\n".join(parts)
+
+ def compile(self, dtb: str) -> None:
+ dts = self.workdir / "test.dts"
+
+ try:
+ subprocess.run(
+ [
+ "gcc",
+ "-E",
+ "-nostdinc",
+ f"-I{self.ksrcdir}/scripts/dtc/include-prefixes",
+ "-undef",
+ "-D__DTS__",
+ "-x",
+ "assembler-with-cpp",
+ "-o",
+ dts,
+ "-",
+ ],
+ input=self.src,
+ text=True,
+ check=True,
+ capture_output=True,
+ )
+
+ full = HEADER.replace("WORK", str(self.workdir)) + dts.read_text()
+ dts.write_text(full)
+
+ subprocess.run(
+ ["dtc", "-I", "dts", "-O", "dtb", dts, "-o", self.workdir / dtb],
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+ except subprocess.CalledProcessError as e:
+ raise Exception(f"{e.stderr}")
diff --git a/tools/testing/roadtest/roadtest/core/hardware.py b/tools/testing/roadtest/roadtest/core/hardware.py
new file mode 100644
index 000000000000..ae81a531d2a2
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/hardware.py
@@ -0,0 +1,94 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import contextlib
+import functools
+import os
+from pathlib import Path
+from typing import Any, Callable, Optional, Type, cast
+from unittest import TestCase
+from unittest.mock import MagicMock, call
+
+from roadtest import ENV_WORK_DIR
+
+from .control import ControlWriter
+from .opslog import OpsLogReader
+from .sysfs import write_int
+
+
+class HwMock(MagicMock):
+ def assert_reg_write_once(self, test: TestCase, reg: int, value: int) -> None:
+ test.assertEqual(
+ [c for c in self.mock_calls if c.args[0] == reg],
+ [call.reg_write(reg, value)],
+ )
+
+ def assert_last_reg_write(self, test: TestCase, reg: int, value: int) -> None:
+ test.assertEqual(
+ [c for c in self.mock_calls if c.args[0] == reg][-1:],
+ [call.reg_write(reg, value)],
+ )
+
+ def get_last_reg_write(self, reg: int) -> int:
+ return cast(int, [c for c in self.mock_calls if c.args[0] == reg][-1].args[1])
+
+
+class Hardware(contextlib.AbstractContextManager):
+ def __init__(self, bus: str, work: Optional[Path] = None) -> None:
+ if not work:
+ work = Path(os.environ[ENV_WORK_DIR])
+
+ self.bus = bus
+ self.mock = HwMock()
+ self.control = ControlWriter(work)
+ self.opslog = OpsLogReader(work)
+ self.loaded_model = False
+
+ # Ignore old entries
+ self.opslog.read_next()
+
+ def _call(self, method: str, *args: Any, **kwargs: Any) -> None:
+ self.control.write_cmd(
+ f"backend.{self.bus}.{method}(*{str(args)}, **{str(kwargs)})"
+ )
+
+ def kick(self) -> None:
+ # Control writes are only applied when the backend gets something
+ # to process, usually because the driver tried to access the device.
+ # But in some cases, such as when the driver is waiting for a
+ # sequence of interrupts, the test code needs the control write to take
+ # effect immediately. For this, we just need to kick the backend
+ # into processing its control queue.
+ #
+ # We (ab)use gpio-leds for this. devicetree.py sets up the device.
+ write_int(Path("/sys/class/leds/led0/brightness"), 0)
+
+ def load_model(self, cls: Type[Any], *args: Any, **kwargs: Any) -> "Hardware":
+ self._call("load_model", cls.__module__, cls.__name__, *args, **kwargs)
+ self.loaded_model = True
+ return self
+
+ def __enter__(self) -> "Hardware":
+ return self
+
+ def __exit__(self, *_: Any) -> None:
+ self.close()
+
+ @functools.cache
+ def __getattr__(self, name: str) -> Callable:
+ def func(*args: Any, **kwargs: Any) -> None:
+ self._call(name, *args, **kwargs)
+
+ return func
+
+ def close(self) -> None:
+ if self.loaded_model:
+ self._call("unload_model")
+ self.control.close()
+
+ def update_mock(self) -> HwMock:
+ opslog = self.opslog.read_next()
+ for line in opslog:
+ eval(line, {"mock": self.mock})
+
+ return self.mock
diff --git a/tools/testing/roadtest/roadtest/core/log.py b/tools/testing/roadtest/roadtest/core/log.py
new file mode 100644
index 000000000000..7d73e40eb2d8
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/log.py
@@ -0,0 +1,42 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+from pathlib import Path
+
+
+class LogParser:
+ DNF_MESSAGE = "<Test did not finish cleanly>"
+
+ def __init__(self, file: Path):
+ try:
+ raw = file.read_text()
+ lines = raw.splitlines()
+ except FileNotFoundError:
+ lines = []
+ raw = ""
+
+ self.raw = raw
+ self.lines = lines
+
+ def has_any(self) -> bool:
+ return "START<" in self.raw
+
+ def get_testcase_log(self, id: str) -> list[str]:
+ startmarker = f"START<{id}>"
+ stopmarker = f"STOP<{id}>"
+
+ try:
+ startpos = next(
+ i for i, line in enumerate(self.lines) if startmarker in line
+ )
+ except StopIteration:
+ return []
+
+ try:
+ stoppos = next(
+ i for i, line in enumerate(self.lines[startpos:]) if stopmarker in line
+ )
+ except StopIteration:
+ return self.lines[startpos + 1 :] + [LogParser.DNF_MESSAGE]
+
+ return self.lines[startpos + 1 : startpos + stoppos]
diff --git a/tools/testing/roadtest/roadtest/core/modules.py b/tools/testing/roadtest/roadtest/core/modules.py
new file mode 100644
index 000000000000..5bd2d92a322b
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/modules.py
@@ -0,0 +1,38 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import os
+import subprocess
+from pathlib import Path
+from typing import Any
+
+from roadtest import ENV_BUILD_DIR
+
+
+def modprobe(modname: str, remove: bool = False) -> None:
+ moddir = Path(os.environ[ENV_BUILD_DIR]) / "modules"
+ args = []
+ if remove:
+ args.append("--remove")
+ args += [f"--dirname={moddir}", modname]
+ subprocess.check_output(["/sbin/modprobe"] + args)
+
+
+def insmod(modname: str) -> None:
+ modprobe(modname)
+
+
+def rmmod(modname: str) -> None:
+ subprocess.check_output(["/sbin/rmmod", modname])
+
+
+class Module:
+ def __init__(self, name: str) -> None:
+ self.name = name
+
+ def __enter__(self) -> "Module":
+ modprobe(self.name)
+ return self
+
+ def __exit__(self, *_: Any) -> None:
+ rmmod(self.name)
diff --git a/tools/testing/roadtest/roadtest/core/opslog.py b/tools/testing/roadtest/roadtest/core/opslog.py
new file mode 100644
index 000000000000..83bb4f525d03
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/opslog.py
@@ -0,0 +1,35 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import os
+from pathlib import Path
+
+OPSLOG_FILE = "opslog.txt"
+
+
+class OpsLogWriter:
+ def __init__(self, work: Path) -> None:
+ path = work / OPSLOG_FILE
+ path.unlink(missing_ok=True)
+ self.file = open(path, "a", buffering=1)
+
+ def write(self, line: str) -> None:
+ self.file.write(line + "\n")
+
+
+class OpsLogReader:
+ def __init__(self, work: Path) -> None:
+ self.path = work / OPSLOG_FILE
+ self.opslogpos = 0
+
+ def read_next(self) -> list[str]:
+ # There is a problem in hostfs (see Hostfs Caveats) which means
+ # that reads from UML on a file which is extended on the host don't see
+ # the new data unless we open and close the file, so we can't open once
+ # and use readlines().
+ with open(self.path, "r") as f:
+ os.lseek(f.fileno(), self.opslogpos, os.SEEK_SET)
+ opslog = [line.rstrip() for line in f.readlines()]
+ self.opslogpos = os.lseek(f.fileno(), 0, os.SEEK_CUR)
+
+ return opslog
diff --git a/tools/testing/roadtest/roadtest/core/proxy.py b/tools/testing/roadtest/roadtest/core/proxy.py
new file mode 100644
index 000000000000..36089e21d7d5
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/proxy.py
@@ -0,0 +1,48 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+from typing import Any
+from unittest import TestCase, TextTestResult
+
+from . import control
+
+
+class ProxyTextTestResult(TextTestResult):
+ def __init__(self, stream: Any, descriptions: Any, verbosity: Any) -> None:
+ super().__init__(stream, descriptions, verbosity)
+ self.successes: list[tuple[TestCase, str]] = []
+
+ # Print via kmsg to avoid getting cut off by other kernel prints.
+ self.kmsg = open("/dev/kmsg", "w", buffering=1)
+ self.control = control.ControlWriter()
+
+ def addSuccess(self, test: TestCase) -> None:
+ super().addSuccess(test)
+ self.successes.append((test, ""))
+
+ def _log(self, test: TestCase, action: str) -> None:
+ line = f"{action}<{test.id()}>"
+ self.kmsg.write(line + "\n")
+ self.control.write_log(line)
+
+ def startTest(self, test: TestCase) -> None:
+ self._log(test, "START")
+ super().startTest(test)
+
+ def stopTest(self, test: TestCase) -> None:
+ super().stopTest(test)
+ self._log(test, "STOP")
+
+ def _replace_id(self, reslist: list[tuple[TestCase, str]]) -> list[tuple[str, str]]:
+ return [(case.id(), tb) for case, tb in reslist]
+
+ def to_proxy(self) -> dict[str, Any]:
+ return {
+ "testsRun": self.testsRun,
+ "wasSuccessful": self.wasSuccessful(),
+ "successes": self._replace_id(self.successes),
+ "errors": self._replace_id(self.errors),
+ "failures": self._replace_id(self.failures),
+ "skipped": self._replace_id(self.skipped),
+ "unexpectedSuccesses": [t.id() for t in self.unexpectedSuccesses],
+ }
diff --git a/tools/testing/roadtest/roadtest/core/suite.py b/tools/testing/roadtest/roadtest/core/suite.py
new file mode 100644
index 000000000000..e99a60b4faba
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/suite.py
@@ -0,0 +1,286 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import json
+import os
+import shlex
+import signal
+import subprocess
+import textwrap
+import unittest
+from pathlib import Path
+from typing import Any, ClassVar, Optional, Tuple, cast
+from unittest import TestResult
+
+from roadtest import ENV_BUILD_DIR, ENV_WORK_DIR
+
+from . import devicetree
+from .log import LogParser
+
+
+class UMLTestCase(unittest.TestCase):
+ run_separately: ClassVar[bool] = False
+ dts: ClassVar[Optional[devicetree.DtFragment]] = None
+
+
+class UMLSuite(unittest.TestSuite):
+ def __init__(
+ self,
+ timeout: int,
+ workdir: str,
+ builddir: str,
+ ksrcdir: str,
+ uml_args_pre: list[str],
+ uml_args_post: list[str],
+ shell: bool,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+
+ self.timeout = timeout
+ self.workdir = Path(workdir).resolve()
+ self.builddir = Path(builddir)
+ self.ksrcdir = Path(ksrcdir)
+ self.uml_args_pre = uml_args_pre
+ self.uml_args_post = uml_args_post
+ self.shell = shell
+
+ self.backendlog = self.workdir / "backend.txt"
+ self.umllog = self.workdir / "uml.txt"
+
+ # Used from the roadtest.cmd.remote running inside UML
+ self.testfile = self.workdir / "tests.json"
+ self.resultfile = self.workdir / "results.json"
+
+ def run(
+ self, result: unittest.TestResult, debug: bool = False
+ ) -> unittest.TestResult:
+ pwd = os.getcwd()
+
+ os.makedirs(self.workdir, exist_ok=True)
+ workdir = self.workdir
+
+ tests = cast(list[UMLTestCase], list(self))
+
+ os.environ[ENV_WORK_DIR] = str(workdir)
+ os.environ[ENV_BUILD_DIR] = str(self.builddir)
+
+ dt = devicetree.Devicetree(workdir=workdir, ksrcdir=self.ksrcdir)
+ dt.assemble([test.dts for test in tests if test.dts])
+ dt.compile("test.dtb")
+
+ testinfos = []
+ ids = []
+ for t in tests:
+ id = t.id()
+ # This fixup is needed when discover is done starting from "roadtest"
+ if not id.startswith("roadtest."):
+ id = f"roadtest.{id}"
+ ids.append(id)
+
+ testinfos.append({"id": id, "values": t.dts.values if t.dts else {}})
+
+ with self.testfile.open("w") as f:
+ json.dump(testinfos, f)
+
+ uml_args = [
+ str(self.builddir / "vmlinux"),
+ f"PYTHONPATH={pwd}",
+ f"{ENV_WORK_DIR}={workdir}",
+ f"{ENV_BUILD_DIR}={self.builddir}",
+ # Should be enough for anybody?
+ "mem=64M",
+ "dtb=test.dtb",
+ "rootfstype=hostfs",
+ "rw",
+ f"init={pwd}/init.sh",
+ f"uml_dir={workdir}",
+ "umid=uml",
+ # ProxyTextTestResult writes to /dev/kmsg
+ "printk.devkmsg=on",
+ "slub_debug",
+ # For ease of debugging
+ "no_hash_pointers",
+ ]
+
+ if self.shell:
+ # See init.sh
+ uml_args += ["ROADTEST_SHELL=1"]
+ else:
+ # Set by slub_debug
+ TAINT_BAD_PAGE = 1 << 5
+ uml_args += [
+ # init.sh increases the loglevel after bootup.
+ "quiet",
+ "panic_on_warn=1",
+ f"panic_on_taint={TAINT_BAD_PAGE}",
+ "oops=panic",
+ # Speeds up delays, but as a consequence also causes
+ # 100% CPU consumption at an idle shell prompt.
+ "time-travel",
+ ]
+
+ main_script = (Path(__file__).parent / "../backend/main.py").resolve()
+
+ args = (
+ [
+ str(self.builddir / "roadtest-backend"),
+ # The socket locations are also present in the devicetree.
+ f"--gpio-socket={workdir}/gpio.sock",
+ f"--i2c-socket={workdir}/i2c.sock",
+ f"--main-script={main_script}",
+ "--",
+ ]
+ + self.uml_args_pre
+ + uml_args
+ + self.uml_args_post
+ )
+
+ print(
+ "Running backend/UML with: {}".format(
+ " ".join([shlex.quote(a) for a in args])
+ )
+ )
+
+ # Truncate instead of deleting so that tail -f can be used to monitor
+ # the log across runs.
+ self.backendlog.write_text("")
+ self.umllog.write_text("")
+ self.resultfile.unlink(missing_ok=True)
+
+ umlpidfile = workdir / "uml/pid"
+ umlpidfile.unlink(missing_ok=True)
+
+ newenv = dict(os.environ, PYTHONPATH=pwd)
+
+ try:
+ process = None
+ with self.backendlog.open("w") as f:
+ process = subprocess.Popen(
+ args,
+ env=newenv,
+ stdin=subprocess.PIPE,
+ stdout=f,
+ stderr=subprocess.STDOUT,
+ text=True,
+ preexec_fn=os.setsid,
+ )
+ process.wait(self.timeout if self.timeout else None)
+ except subprocess.TimeoutExpired:
+ pass
+ finally:
+ try:
+ if process:
+ os.killpg(process.pid, signal.SIGKILL)
+ except ProcessLookupError:
+ pass
+ try:
+ pid = int(umlpidfile.read_text())
+ os.killpg(pid, signal.SIGKILL)
+ except (FileNotFoundError, ProcessLookupError):
+ pass
+
+ if process and process.returncode is not None and process.returncode != 0:
+ with self.backendlog.open("a") as f:
+ f.write(f"<Backend exited with error code {process.returncode}>\n")
+
+ try:
+ with self.resultfile.open("r") as f:
+ proxy = json.load(f)
+ except FileNotFoundError:
+ # UML crashed, timed out, etc
+ proxy = None
+
+ return self._convert_results(proxy, tests, result)
+
+ def _parse_status(self, id: str, proxy: dict) -> Tuple[str, str]:
+ if not proxy:
+ return "ERROR", "No result. UML or backend crashed?\n"
+
+ try:
+ _, tb = next(e for e in proxy["successes"] if e[0] == id)
+ return "ok", ""
+ except StopIteration:
+ pass
+
+ try:
+ _, tb = next(e for e in proxy["errors"] if e[0] == id)
+ return "ERROR", tb
+ except StopIteration:
+ pass
+
+ try:
+ _, tb = next(e for e in proxy["failures"] if e[0] == id)
+ return "FAIL", tb
+ except StopIteration:
+ pass
+
+ # setupClass, etc
+ if proxy["errors"]:
+ _, tb = proxy["errors"][0]
+ return "ERROR", tb
+
+ raise Exception("Unable to parse status")
+
+ def _get_log(
+ self, name: str, parser: LogParser, id: str, full_if_none: bool
+ ) -> Optional[str]:
+ testloglines = parser.get_testcase_log(id)
+ tb = None
+ if testloglines:
+ tb = "\n".join([f"{name} log:"] + [" " + line for line in testloglines])
+ elif full_if_none and not parser.has_any():
+ if parser.raw:
+ tb = "\n".join(
+ [f"Full {name} log:", textwrap.indent(parser.raw, " ").rstrip()]
+ )
+ else:
+ tb = f"\nNo {name} log found."
+
+ return tb
+
+ def _convert_results(
+ self,
+ proxy: dict,
+ tests: list[UMLTestCase],
+ result: TestResult,
+ ) -> TestResult:
+ umllog = LogParser(self.umllog)
+ backendlog = LogParser(self.backendlog)
+
+ first_fail = True
+ for test in tests:
+ assert isinstance(test, unittest.TestCase)
+
+ id = test.id()
+ if not id.startswith("roadtest."):
+ id = f"roadtest.{id}"
+
+ status, tb = self._parse_status(id, proxy)
+ if status != "ok":
+ parts = []
+
+ backendtb = self._get_log("Backend", backendlog, id, first_fail)
+ if backendtb:
+ parts.append(backendtb)
+
+ umltb = self._get_log("UML", umllog, id, first_fail)
+ if umltb:
+ parts.append(umltb)
+
+ # In the case of no START/STOP markers at all in the logs, we include
+ # the full logs, but only do it in the first failing test case to
+ # reduce noise.
+ first_fail = False
+ tb = "\n\n".join(parts + [tb])
+
+ if status == "ERROR":
+ result.errors.append((test, tb))
+ elif status == "FAIL":
+ result.failures.append((test, tb))
+
+ print(f"{test} ... {status}")
+ result.testsRun += 1
+
+ return result
diff --git a/tools/testing/roadtest/roadtest/core/sysfs.py b/tools/testing/roadtest/roadtest/core/sysfs.py
new file mode 100644
index 000000000000..64228978718e
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/sysfs.py
@@ -0,0 +1,77 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import contextlib
+from pathlib import Path
+from typing import Iterator
+
+
+# Path.write_text() is inappropriate since Python calls write(2)
+# a second time if the first one returns an error, if the file
+# was opened as text.
+def write_str(path: Path, val: str) -> None:
+ path.write_bytes(val.encode())
+
+
+def write_int(path: Path, val: int) -> None:
+ write_str(path, str(val))
+
+
+def write_float(path: Path, val: float) -> None:
+ write_str(path, str(val))
+
+
+def read_str(path: Path) -> str:
+ return path.read_text().rstrip()
+
+
+def read_int(path: Path) -> int:
+ return int(read_str(path))
+
+
+def read_float(path: Path) -> float:
+ return float(read_str(path))
+
+
+class I2CDevice:
+ def __init__(self, addr: int, bus: int = 0) -> None:
+ self.id = f"{bus}-{addr:04x}"
+ self.path = Path(f"/sys/bus/i2c/devices/{self.id}")
+
+
+class PlatformDevice:
+ def __init__(self, name: str) -> None:
+ self.id = name
+ self.path = Path(f"/sys/bus/platform/devices/{self.id}")
+
+
+class I2CDriver:
+ def __init__(self, driver: str) -> None:
+ self.driver = driver
+ self.path = Path(f"/sys/bus/i2c/drivers/{driver}")
+
+ @contextlib.contextmanager
+ def bind(self, addr: int, bus: int = 0) -> Iterator[I2CDevice]:
+ dev = I2CDevice(addr, bus)
+ write_str(self.path / "bind", dev.id)
+
+ try:
+ yield dev
+ finally:
+ write_str(self.path / "unbind", dev.id)
+
+
+class PlatformDriver:
+ def __init__(self, driver: str) -> None:
+ self.driver = driver
+ self.path = Path(f"/sys/bus/platform/drivers/{driver}")
+
+ @contextlib.contextmanager
+ def bind(self, addr: str) -> Iterator[PlatformDevice]:
+ dev = PlatformDevice(addr)
+ write_str(self.path / "bind", dev.id)
+
+ try:
+ yield dev
+ finally:
+ write_str(self.path / "unbind", dev.id)
diff --git a/tools/testing/roadtest/roadtest/core/test_control.py b/tools/testing/roadtest/roadtest/core/test_control.py
new file mode 100644
index 000000000000..a8cf9105eb52
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/test_control.py
@@ -0,0 +1,35 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest import TestCase
+
+from .control import ControlReader, ControlWriter
+
+
+class TestControl(TestCase):
+ def test_control(self) -> None:
+ with TemporaryDirectory() as tmpdir:
+ work = Path(tmpdir)
+ reader = ControlReader(work)
+ writer = ControlWriter(work)
+
+ values = []
+
+ def append(new: int) -> None:
+ nonlocal values
+ values.append(new)
+
+ vars = {"append": append}
+ writer.write_cmd("append(1)")
+
+ reader.process(vars)
+ self.assertEqual(values, [1])
+
+ writer.write_cmd("append(2)")
+ writer.write_log("append(4)")
+ writer.write_cmd("append(3)")
+
+ reader.process(vars)
+ self.assertEqual(values, [1, 2, 3])
diff --git a/tools/testing/roadtest/roadtest/core/test_devicetree.py b/tools/testing/roadtest/roadtest/core/test_devicetree.py
new file mode 100644
index 000000000000..db61fd24b39a
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/test_devicetree.py
@@ -0,0 +1,31 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from . import devicetree
+
+
+class TestDevicetree(unittest.TestCase):
+ def test_compile(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ tmpdir = Path(tmp)
+ # We don't have the ksrcdir so we can't test if includes work.
+ dt = devicetree.Devicetree(tmpdir, tmpdir)
+
+ dt.assemble(
+ [
+ devicetree.DtFragment(
+ src="""
+&i2c {
+ foo = <1>;
+};
+ """
+ )
+ ]
+ )
+ dt.compile("test.dtb")
+ dtb = tmpdir / "test.dtb"
+ self.assertTrue((dtb).exists())
diff --git a/tools/testing/roadtest/roadtest/core/test_hardware.py b/tools/testing/roadtest/roadtest/core/test_hardware.py
new file mode 100644
index 000000000000..eb09b317e258
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/test_hardware.py
@@ -0,0 +1,41 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest import TestCase
+
+from roadtest.backend.mock import MockBackend
+
+from .hardware import Hardware
+
+
+class TestHardware(TestCase):
+ def test_mock(self) -> None:
+ with TemporaryDirectory() as tmpdir:
+ work = Path(tmpdir)
+
+ backend = MockBackend(work)
+ hw = Hardware(bus="dummy", work=work)
+
+ backend.reg_write(0x1, 0xDEAD)
+ backend.reg_write(0x2, 0xBEEF)
+ mock = hw.update_mock()
+ mock.assert_reg_write_once(self, 0x1, 0xDEAD)
+
+ backend.reg_write(0x1, 0xCAFE)
+ mock = hw.update_mock()
+ with self.assertRaises(AssertionError):
+ mock.assert_reg_write_once(self, 0x1, 0xDEAD)
+
+ mock.assert_last_reg_write(self, 0x1, 0xCAFE)
+
+ self.assertEqual(mock.get_last_reg_write(0x1), 0xCAFE)
+ self.assertEqual(mock.get_last_reg_write(0x2), 0xBEEF)
+
+ with self.assertRaises(IndexError):
+ self.assertEqual(mock.get_last_reg_write(0x3), 0x0)
+
+ mock.reset_mock()
+ with self.assertRaises(AssertionError):
+ mock.assert_last_reg_write(self, 0x2, 0xBEEF)
diff --git a/tools/testing/roadtest/roadtest/core/test_log.py b/tools/testing/roadtest/roadtest/core/test_log.py
new file mode 100644
index 000000000000..6988ff4419db
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/test_log.py
@@ -0,0 +1,54 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from unittest import TestCase
+
+from .log import LogParser
+
+
+class TestLog(TestCase):
+ def test_parser(self) -> None:
+ with NamedTemporaryFile() as tmpfile:
+ path = Path(tmpfile.name)
+
+ path.write_text(
+ """
+xyz START<finished>
+finished1
+finished2
+STOP<finished>
+START<empty>
+STOP<empty>
+START<foo> monkey STOP<foo>
+START<unfinished>
+unfinished1
+unfinished2"""
+ )
+
+ parser = LogParser(path)
+ self.assertEqual(
+ parser.get_testcase_log("finished"), ["finished1", "finished2"]
+ )
+
+ self.assertEqual(
+ parser.get_testcase_log("unfinished"),
+ ["unfinished1", "unfinished2", LogParser.DNF_MESSAGE],
+ )
+
+ self.assertEqual(
+ parser.get_testcase_log("notpresent"),
+ [],
+ )
+
+ self.assertEqual(
+ parser.get_testcase_log("enpty"),
+ [],
+ )
+
+ # Shouldn't happen since we print from the kernel?
+ self.assertEqual(
+ parser.get_testcase_log("foo"),
+ [],
+ )
diff --git a/tools/testing/roadtest/roadtest/core/test_opslog.py b/tools/testing/roadtest/roadtest/core/test_opslog.py
new file mode 100644
index 000000000000..bd594c587032
--- /dev/null
+++ b/tools/testing/roadtest/roadtest/core/test_opslog.py
@@ -0,0 +1,27 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright Axis Communications AB
+
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest import TestCase
+
+from .opslog import OpsLogReader, OpsLogWriter
+
+
+class TestOpsLOg(TestCase):
+ def test_opslog(self) -> None:
+ with TemporaryDirectory() as tmpdir:
+ work = Path(tmpdir)
+ writer = OpsLogWriter(work)
+ reader = OpsLogReader(work)
+
+ self.assertEqual(reader.read_next(), [])
+
+ writer.write("1")
+ writer.write("2")
+
+ self.assertEqual(reader.read_next(), ["1", "2"])
+ self.assertEqual(reader.read_next(), [])
+
+ writer.write("3")
+ self.assertEqual(reader.read_next(), ["3"])
diff --git a/tools/testing/roadtest/roadtest/tests/__init__.py b/tools/testing/roadtest/roadtest/tests/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
--
2.34.1