All sections › XII. Networks and DevOps for the engineer

STM32F103, STM32F407 · level intermediate · 50 min read

110б. A board on the desk in the pipeline: pytest, the serial port and your own runner

The pipeline from sections 110 and 110а builds the image, runs unit tests on the computer and keeps an eye on the size. The table of levels in section 184.1 also has the next step — "on the board, with a test bench" — but until now it stayed a row in a table. Here it is built in full: the board lies on the desk next to a small computer, that computer is a pipeline runner with the tag bench, and on every change it flashes the board and runs tests over the serial port.

The tests are written with pytest. Everything about the board is hidden in the fixture board: flash, open the port, wait for the boot line, reset before the test. The tests themselves come out three lines each: command — reply — check. The port, baud rate, image and programmer come in as command-line options, so the same suite serves any board on the desk.

The device under test is the thermostat from section 196 with a UART console. The script from section 115в already drove it through pyserial; there it was a test script, here it is tests in a pipeline. All runs in this section were done without a live board: the board's role is played by firmware in QEMU, which has USART1 routed to a pseudo-terminal. To pyserial a pseudo-terminal looks just like /dev/ttyUSB0, so the test and fixture code is the same. Flashing through st-flash and openocd is shown in the code, but it was not run with a real programmer.

Foundations: the thermostat logic is the file from section 196 without a single change; the platform for QEMU follows the pattern of section 115в; the startup code and linker script come from 115в; CMSIS headers for the F407. Experiments run in the stm-arm:1 image: Python 3.11.2, pyserial 3.5, pytest 9.1.1, QEMU 7.2.22 (machine netduinoplus2, which is an STM32F405 with the same USART1 and SysTick peripherals as the F407), arm-none-eabi-gcc 12.2.1. The st-flash and openocd commands are as in section 35б. The idea comes from a Golioth blog post on automated tests on hardware with pytest (blog.golioth.io); the text, code and numbers here are our own.

What it is called in the documentation

Term Meaning

hardware-in-the-loop (HIL), test bench tests run on a real board

fixture, conftest.py setup and teardown for a test; the file with the suite's fixtures

command-line option, pytest_addoption a pytest command-line parameter

session scope, function scope a fixture for the whole run, for each test

self-hosted runner, tags, labels your own runner; tags in GitLab, labels in GitHub

resource_group, concurrency a group: no more than one job at a time

JUnit XML report a test report in JUnit format, understood by GitLab and GitHub

pseudo-terminal (pty) a "program — port" pair without hardware

udev rule, SYMLINK a udev rule: a persistent device name

dialout the Linux group that owns serial ports

Term	Meaning
hardware-in-the-loop (HIL), test bench	tests run on a real board
fixture, `conftest.py`	setup and teardown for a test; the file with the suite's fixtures
command-line option, `pytest_addoption`	a pytest command-line parameter
session scope, function scope	a fixture for the whole run, for each test
self-hosted runner, tags, labels	your own runner; tags in GitLab, labels in GitHub
`resource_group`, `concurrency`	a group: no more than one job at a time
JUnit XML report	a test report in JUnit format, understood by GitLab and GitHub
pseudo-terminal (pty)	a "program — port" pair without hardware
udev rule, `SYMLINK`	a udev rule: a persistent device name
`dialout`	the Linux group that owns serial ports

110б.1 Where the bench sits and what it catches

The unit tests from 110 and 184 do not see the hardware: timings, interrupts, a real UART, whether the device starts at all after flashing. The bench sees exactly that, but pays with time and with the fact that there is only one board. Hence the layout of the pipeline:

Job	Where it runs	What it does
`build`	any runner, in a container	builds the bench image and puts it in an artifact
`hil-thermo-a`	only a runner with tags `bench` and `thermo-a`	flashes board A, runs pytest, returns a JUnit report
`hil-thermo-b`	a runner with tags `bench` and `thermo-b`	the same on board B

The runner with the board is an ordinary machine: a small computer or a Raspberry Pi with an ST-LINK programmer and a USB–UART adapter (or the ST-LINK port, section 115б.1). It has the runner service, Python with pytest and pyserial, and the flashing program. The job runs directly on this machine, without a container: that is the simplest way to reach USB.

The bench code splits into three layers, and only the bottom one knows whether a board actually exists:

File	What is in it
`test_thermo.py`	tests: command, reply, check
`conftest.py`	command-line options and the fixtures `bench`, `board`
`boardlib.py`	`Board` — the port and lines; flashers: st-flash, openocd, QEMU

110б.2 A board for the test: a line with a timeout and waiting for a pattern

The main rule when talking to a device from a test is never wait forever. A hung board must not hang the pipeline: every read has a timeout, and if the expected line does not come, the test fails with a clear message — what was expected and what arrived instead.

Reading one line. pyserial reads up to \n or until the timeout expires; a fragment without \n counts as "no line":

    def readline(self, timeout):
        """Одна строка без \\r\\n или None, если за timeout секунд целой строки не пришло."""
        self.ser.timeout = timeout
        raw = self.ser.read_until(b"\n")
        if not raw.endswith(b"\n"):
            return None
        s = raw.rstrip(b"\r\n").decode("utf-8", errors="replace")
        self.log.append((round(time.monotonic() - self.t0, 3), s))
        return s

Waiting for a regular expression. The device may send something without being asked (the thermostat sends a sensor alarm line or the factory-defaults cfg: line after start-up), so foreign lines are skipped, and the last five go into the error text:

    def expect(self, pattern, timeout=1.0):
        """Ждать строку, подходящую под регулярное выражение; чужие строки пропустить."""
        rx = re.compile(pattern)
        end = time.monotonic() + timeout
        seen = []
        while True:
            left = end - time.monotonic()
            if left <= 0:
                raise BoardTimeout("нет /%s/ за %.1f с; пришло: %r" % (pattern, timeout, seen[-5:]))
            s = self.readline(left)
            if s is None:
                continue
            m = rx.search(s)
            if m:
                return m
            seen.append(s)

A command with a reply is: flush the input buffer, send, wait:

    def ask(self, cmd, pattern, timeout=1.0):
        """Команда → ответ по образцу. Что пришло до команды, выбросить."""
        self.ser.reset_input_buffer()
        self.send(cmd)
        return self.expect(pattern, timeout)

The BoardTimeout message goes into the report in full. A timeout in the test itself is recorded by pytest as a failure, and in setup, that is in a fixture, as an error; the difference shows up in 110б.7 with a slow start. In the bench experiment, this is how the report line for a hung board looks:

boardlib.BoardTimeout: нет /^t=(-?\d+\.\d) set=(-?\d+\.\d) (\S+)$/ за 1.0 с; пришло: []

The port is opened with exclusive=True: pyserial puts a lock on it, and a second opener is refused at once instead of sharing the byte stream (section 115б.7). Flashing and reset go through the flasher; after them the bench waits for the boot line:

    def flash(self, image):
        self.close()                           # порт мог исчезнуть вместе с платой — открыть заново
        self.flasher.flash(image)
        self.open()
        return self.expect(r"^boot thermo196 v(\S+)$", self.boot_timeout)

    def reset(self):
        self.ser.reset_input_buffer()
        self.flasher.reset()
        return self.expect(r"^boot thermo196 v(\S+)$", self.boot_timeout)

110б.3 Flash and reset: st-flash, openocd and QEMU

Every flasher has two actions: flash(image) and reset(). For a real board it is a programmer selected by serial number: there may be several ST-LINKs on the desk, and you must flash exactly your own.

class StFlashFlasher:
    """Настоящая плата через ST-LINK и stlink-tools: программатор выбирается по серийному номеру."""

    def __init__(self, probe):
        self.probe = probe

    def flash(self, image):
        binary = image[:-4] + ".bin" if image.endswith(".elf") else image
        subprocess.run(["st-flash", "--serial", self.probe, "--reset", "write", binary, "0x08000000"],
                       check=True, timeout=60)

    def reset(self):
        subprocess.run(["st-flash", "--serial", self.probe, "reset"], check=True, timeout=20)

    def stop(self):
        pass

st-flash takes only a .bin and an address (section 35б.3), so the build puts a .bin next to the .elf. openocd flashes directly from the .elf with the same line program … verify reset exit; the programmer's serial number is set by the command adapter serial:

class OpenOcdFlasher:
    """Настоящая плата через openocd 0.12: тот же ST-LINK, но прошивка прямо из ELF."""

    def __init__(self, probe, target="target/stm32f1x.cfg"):
        self.base = ["openocd", "-f", "interface/stlink.cfg", "-c", "adapter serial %s" % probe, "-f", target]

    def flash(self, image):
        subprocess.run(self.base + ["-c", "program %s verify reset exit" % image], check=True, timeout=60)

    def reset(self):
        subprocess.run(self.base + ["-c", "init; reset run; exit"], check=True, timeout=20)

    def stop(self):
        pass

Without a board, QEMU takes its place. "Flash" means restarting QEMU with the new image, "reset" means sending system_reset to its monitor over a socket. QEMU creates the port itself (-serial pty) and names it in the first line of output, for example /dev/pts/0; the number differs each time, so the bench keeps a persistent link and swaps it with a single os.replace — the same way udev keeps /dev/bench/thermo-a for a real board (110б.7):

class QemuFlasher:
    """«Плата» — QEMU netduinoplus2. Прошить = перезапустить QEMU с образом; сбросить = system_reset.
    Порт: QEMU создаёт /dev/pts/N (номер каждый раз свой), стенд держит постоянную ссылку port."""

    def __init__(self, port, workdir="/tmp"):
        self.port, self.proc = port, None
        self.mon = os.path.join(workdir, os.path.basename(port) + ".mon")

    def flash(self, image):
        self.stop()
        self.proc = subprocess.Popen(
            ["qemu-system-arm", "-M", "netduinoplus2", "-nographic", "-kernel", image,
             "-serial", "pty", "-monitor", "unix:%s,server=on,wait=off" % self.mon],
            stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
        m = re.search(r"char device redirected to (/dev/pts/\d+)", self.proc.stdout.readline())
        if not m:
            raise RuntimeError("QEMU не назвал псевдотерминал")
        self.pty = m.group(1)
        tmp = self.port + ".new"               # подменить ссылку разом, как это делает udev
        if os.path.lexists(tmp):
            os.remove(tmp)
        os.symlink(self.pty, tmp)
        os.replace(tmp, self.port)

    def reset(self):
        with socket.socket(socket.AF_UNIX) as s:
            s.connect(self.mon)
            s.sendall(b"system_reset\n")
            time.sleep(0.05)                   # дать монитору принять команду до закрытия сокета

    def stop(self):
        if self.proc is not None:
            self.proc.kill()
            self.proc.wait()
            self.proc = None

def make_flasher(name, port, probe):
    if name == "qemu":
        return QemuFlasher(port)
    if name == "st-flash":
        return StFlashFlasher(probe)
    if name == "openocd":
        return OpenOcdFlasher(probe)
    raise ValueError("неизвестный --flasher: %s" % name)

The bench firmware is thermostat 196 with a platform for QEMU. It differs from the 115в platform in three things: after start-up it prints the line boot thermo196 v1.4.2, it answers ver with the version set by the pipeline, and in the bench build (-DBENCH_HOOKS) it understands hang — it hangs for good with interrupts disabled. This models a fault after which the device goes silent. Parsing a console line:

static void rx(struct thermo *t, char c)
{
    if (c == '\r' || c == '\n') {
        line[len] = 0;
        if (!pass && same(line, "ver")) {
            plat_puts("ver=" FW_VERSION "\n");
#ifdef BENCH_HOOKS
        } else if (!pass && same(line, "hang")) {
            __disable_irq();                              /* ни SysTick, ни UART — прибор молчит */
            for (;;) {
            }
#endif
        } else {
            for (unsigned i = 0; !pass && i < len; i++)
                thermo_rx(t, line[i]);
            thermo_rx(t, c);
        }
        len = pass = 0;
    } else if (pass) {
        thermo_rx(t, c);
    } else if (len < sizeof line - 1) {
        line[len++] = c;
    } else {
        for (unsigned i = 0; i < len; i++)
            thermo_rx(t, line[i]);
        thermo_rx(t, c);
        pass = 1;
        len = 0;
    }
}

The image builds without warnings (-Wall -Wextra -Werror): 1964 bytes of code and constants, 8 bytes of .data, 128 bytes of .bss, the .bin is 1972 bytes.

110б.4 Fixtures and command-line options

Everything that distinguishes one board from another comes in as options: --port, --baud, --image, --flasher, --probe (the programmer's serial number), --fw-version (what must end up on the board) and --boot-timeout. The defaults for the port, image and programmer are taken from environment variables — they are convenient to set in the job settings.

def pytest_addoption(parser):
    g = parser.getgroup("board", "плата на стенде")
    g.addoption("--port", default=os.environ.get("BOARD_PORT"), help="порт платы: /dev/bench/..., COM5")
    g.addoption("--baud", type=int, default=115200, help="скорость порта")
    g.addoption("--image", default=os.environ.get("FW_IMAGE"), help="что прошить: .elf или .bin")
    g.addoption("--flasher", default="st-flash", choices=["st-flash", "openocd", "qemu"])
    g.addoption("--probe", default=os.environ.get("PROBE_SERIAL", ""), help="серийный номер программатора")
    g.addoption("--fw-version", default=os.environ.get("FW_VERSION"), help="какую версию ждать от платы")
    g.addoption("--boot-timeout", type=float, default=3.0, help="сколько ждать строку старта, с")
    g.addoption("--no-reset", action="store_true", help="не сбрасывать плату перед каждым тестом")

There are two fixtures, and they live for different times. bench lives for the whole run: flashing is the slowest operation, so it is done once. The teardown is registered before flashing, so QEMU (or the port) is closed even when the board did not start:

@pytest.fixture(scope="session")
def bench(request):
    """Прошить один раз на весь прогон: прошивка — самое долгое."""
    o = request.config.option
    if not o.port or not o.image:
        pytest.exit("нужны --port и --image (или BOARD_PORT и FW_IMAGE)", returncode=2)
    flasher = make_flasher(o.flasher, o.port, o.probe)
    request.addfinalizer(flasher.stop)     # и когда плата не стартовала: QEMU не должен остаться
    b = Board(o.port, o.baud, flasher, o.boot_timeout)
    request.addfinalizer(b.close)
    b.boot = b.flash(o.image)
    return b

board is needed by every test and resets the board before it. This is the bench's main protection: a hang or a changed setpoint in one test does not carry over to the next.

@pytest.fixture
def board(bench, request):
    """Каждому тесту — плата сразу после сброса: чужое зависание или уставка не переходят дальше."""
    if not request.config.option.no_reset:
        bench.boot = bench.reset()
    yield bench

The suite settings file is pytest.ini: the JUnit report in xunit2 format, the demo experiment from 110б.5 is not part of a normal run:

[pytest]
markers =
    demo: опыт раздела 110б — зависание без pytest.raises, в обычный прогон не входит
addopts = -m "not demo" -p no:cacheprovider
junit_family = xunit2

110б.5 Thermostat tests and a hung board

Six tests, one of them with four command variants — nine runs in total:

"""Раздел 110б: термостат 196 на стенде. Каждый тест: команда — строка — проверка."""
import pytest

from boardlib import BoardTimeout

GET = r"^t=(-?\d+\.\d) set=(-?\d+\.\d) (\S+)$"


def test_version(board, fw_version):
    ver = board.ask("ver", r"^ver=(\S+)$").group(1)
    assert ver == board.boot.group(1)            # строка старта и команда говорят одно
    if fw_version:
        assert ver == fw_version                 # на плате именно то, что собрал конвейер


def test_get(board):
    t, sp, state = board.ask("get", GET).groups()
    assert 5.0 <= float(t) <= 35.0
    assert sp == "21.0"                          # после сброса — заводская уставка
    assert state in ("ждёт", "нагрев")


def test_set_roundtrip(board):
    try:
        assert board.ask("set 22.5", r"^(ok|\?.*)$").group(1) == "ok"
        assert board.ask("get", GET).group(2) == "22.5"
    finally:
        board.ask("set 21.0", r"^ok$")           # на плате уставка во Flash: сброс её не вернёт


@pytest.mark.parametrize("cmd", ["foo", "set 99", "set", "hyst 9"])
def test_bad_command(board, cmd):
    assert board.ask(cmd, r"^(ok|\?.*)$").group(1).startswith("?")


def test_long_line(board):
    assert board.ask("x" * 40, r"^\?.*$").group(0) == "? строка длиннее 31 символа"


def test_hang_then_reset(board):
    board.send("hang")
    with pytest.raises(BoardTimeout):
        board.ask("get", GET, timeout=1.0)       # плата молчит — тест не ждёт вечно
    board.reset()
    assert board.ask("get", GET)                 # после сброса снова отвечает

What is checked:

Test	Command	What we expect
`test_version`	`ver`	the version matches the boot line and the one the pipeline built
`test_get`	`get`	temperature 5…35 °C, factory setpoint 21.0, state "idle" or "heating"
`test_set_roundtrip`	`set 22.5`, `get`	`ok`, then setpoint 22.5; at the end the setpoint is restored
`test_bad_command`	`foo`, `set 99`, `set`, `hyst 9`	the reply starts with `?`
`test_long_line`	40 characters `x`	`? line longer than 31 characters`
`test_hang_then_reset`	`hang`, `get`	no reply within 1 s, after a reset `get` answers

The bench job in QEMU, pytest output:

collected 12 items / 3 deselected / 9 selected

bench/test_thermo.py .........                                           [100%]

---------------- generated xml file: /tmp/repo/report/junit.xml ----------------
======================= 9 passed, 3 deselected in 1.55s ========================

Time per test, from the JUnit report:

Test	Time, s
`test_version` (with flashing)	0.072
`test_get`	0.051
`test_set_roundtrip`	0.052
`test_bad_command`, each of the four	0.051
`test_long_line`	0.053
`test_hang_then_reset`	1.104
total (pytest)	1.55
the whole job (Python start-up, test collection, report)	2.17

Almost all of the run time is one timeout: the hang test waits for a reply for 1 s. The other tests in QEMU take 0.05 s each, and almost all of that time is the 0.05 s pause after the reset command to the monitor. On a real board flashing and reset through a programmer take longer; those times were not measured here.

Why reset before each test. Three tests of the demo experiment: the first sends hang and does not catch the timeout — this is what a real firmware bug looks like. The next two are ordinary get and ver:

"""Раздел 110б, опыт: тест, после которого плата висит. Запуск: pytest bench -m demo [--no-reset]."""
import pytest

from test_thermo import GET

pytestmark = pytest.mark.demo


def test_a_hangs(board):
    board.send("hang")
    board.ask("get", GET, timeout=1.0)           # так выглядит настоящее зависание: ответа нет


def test_b_get(board):
    board.ask("get", GET)


def test_c_version(board):
    board.ask("ver", r"^ver=\S+$")

Run	Result	Time
with a reset before each test	1 failed, 2 passed	1.41 s
`--no-reset`	all 3 failed, each on a 1 s timeout	3.28 s

Without a reset, one hung board brings down every test after it, and the report does not show which one was first. With a reset, exactly the test on which the board hung fails.

110б.6 The pipeline: your own runner with the bench tag

The bench job calls a single script — you can also run it by hand on the bench machine:

#!/bin/sh
# Раздел 110б: задание стенда. BOARD_PORT, FLASHER, PROBE_SERIAL — из переменных задания,
# build/ — образ и версия от задания build. Код выхода — код pytest.
set -eu
. build/version.env
mkdir -p report
set +e
python3 -m pytest bench --port "$BOARD_PORT" --baud 115200 --image build/thermo.elf \
    --flasher "$FLASHER" --probe "${PROBE_SERIAL:-}" --fw-version "$FW_VERSION" \
    --junitxml report/junit.xml -o junit_suite_name="hil-$(basename "$BOARD_PORT")" \
    > report/pytest.txt 2>&1
rc=$?
set -e
cat report/pytest.txt
exit $rc

The report/junit.xml report goes to the service even when tests fail — that is what it is for. The exit code is pytest's exit code: a test fails, the job is red. The suite name in the report is taken from the port name (hil-thermo-a), so that the reports of two boards do not get mixed up.

The pipeline file for GitLab (in the repository it is .gitlab-ci.yml):

# Раздел 110б: .gitlab-ci.yml — сборка в контейнере на любом исполнителе,
# тесты на плате — только на исполнителе с тегом bench, к которому она подключена.
stages: [build, bench]

build:
  stage: build
  image: registry.example.com/tools/stm-arm:1
  tags: [docker]
  variables:
    GIT_DEPTH: "0"                    # git describe нужна история с тегами
  script:
    - sh ci/build_fw.sh
  artifacts:
    paths: [build/]
    expire_in: 1 week

.hil:
  stage: bench
  needs: [build]
  timeout: 10 minutes
  interruptible: false                # не обрывать на середине прошивки
  variables:
    FLASHER: st-flash
  script:
    - sh ci/bench_job.sh
  artifacts:
    when: always                      # отчёт нужен как раз когда тесты упали
    paths: [report/]
    reports:
      junit: report/junit.xml
    expire_in: 2 weeks

hil-thermo-a:
  extends: .hil
  tags: [bench, thermo-a]             # исполнитель, к которому подключена плата A
  resource_group: bench-thermo-a      # одна плата — одно задание за раз
  variables:
    BOARD_PORT: /dev/bench/thermo-a
    PROBE_SERIAL: $PROBE_THERMO_A     # серийный номер ST-LINK — в настройках CI/CD

hil-thermo-b:
  extends: .hil
  tags: [bench, thermo-b]
  resource_group: bench-thermo-b
  variables:
    BOARD_PORT: /dev/bench/thermo-b
    PROBE_SERIAL: $PROBE_THERMO_B

Four keys turn a job into a bench job:

Key	Why
`tags: [bench, thermo-a]`	the job is picked up only by a runner that has both tags
`resource_group`	a second job for the same board waits until the first one finishes
`needs: [build]`	the image arrives as an artifact from the build; nothing is built on the bench machine
`artifacts: when: always` + `reports: junit`	the report is kept even on failure; GitLab shows the tests in the merge request

The same for GitHub Actions (.github/workflows/bench.yml); a tag here is called a label, and the queue for the board is set by concurrency:

# Раздел 110б: .github/workflows/bench.yml — то же для GitHub Actions.
name: bench
on:
  push:
    branches: [main]
  pull_request:

jobs:
  build:
    runs-on: ubuntu-latest
    container: registry.example.com/tools/stm-arm:1
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
      - run: sh ci/build_fw.sh
      - uses: actions/upload-artifact@v4
        with:
          name: fw
          path: build/

  hil:
    needs: build
    runs-on: [self-hosted, bench, thermo-a]
    timeout-minutes: 10
    concurrency:
      group: bench-thermo-a
      cancel-in-progress: false
    env:
      BOARD_PORT: /dev/bench/thermo-a
      FLASHER: st-flash
      PROBE_SERIAL: ${{ vars.PROBE_THERMO_A }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/download-artifact@v4
        with:
          name: fw
          path: build/
      - run: sh ci/bench_job.sh
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: report-thermo-a
          path: report/

The runner was not registered in this section. In GitLab you create it in the project settings (CI/CD → Runners) with tags bench and thermo-a, and on the bench machine you register the gitlab-runner service with the shell executor — the job then runs directly on that machine. In GitHub the labels bench and thermo-a are set when you configure your own runner.

No service was at hand, so both files were checked with our own script ci_lint.py: the YAML parses, all keys come from lists of known ones, references in needs and extends point to existing jobs, stages are listed in stages, and every job with the bench tag has a resource_group, a JUnit report and when: always. Output on these files:

gitlab_stages=build bench
gitlab_jobs=build hil-thermo-a hil-thermo-b
gitlab_bench_jobs=hil-thermo-a hil-thermo-b
gitlab_resource_groups=bench-thermo-a bench-thermo-b
github_on=pull_request push
github_jobs=build hil
github_self_hosted=hil:self-hosted,bench,thermo-a
lint_errors=0

The same script on broken copies finds the typo tag: instead of tags: (with it the job would lose the bench tag and could go to any runner), two boards in one group, and a report without when: always. This does not replace a check by the service itself: gitlab-ci-lint and actionlint were not run on these files.

110б.7 Pitfalls

The port name changed — the tests talk to someone else's board. The number in /dev/ttyACM0 is the connection order, not the socket (section 115б.2). In the experiment two "boards" x and y, started in the order x, y, got /dev/pts/0 and /dev/pts/1; in the order y, x — the other way round. The bench needs a persistent address: a udev rule by the programmer's serial number, one link per board.

# Раздел 110б: /etc/udev/rules.d/99-bench.rules на машине-исполнителе.
# Порт платы называется по серийному номеру её ST-LINK, а не по порядку подключения.
# Свои номера: udevadm info -a -n /dev/ttyACM0 | grep '{serial}'
SUBSYSTEM=="tty", ATTRS{idVendor}=="0483", ATTRS{serial}=="SN_THERMO_A", SYMLINK+="bench/thermo-a", GROUP="dialout", MODE="0660"
SUBSYSTEM=="tty", ATTRS{idVendor}=="0483", ATTRS{serial}=="SN_THERMO_B", SYMLINK+="bench/thermo-b", GROUP="dialout", MODE="0660"

The rules in this section were not tested on a machine with udev; the serial numbers here are placeholders. The pipeline uses only the persistent name: /dev/bench/thermo-a.

Several boards on one machine. Each board gets its own port, its own programmer serial number, its own tag and its own group. Two bench jobs on two "boards" in the experiment ran in parallel: each took 1.548 s in pytest, both together 2.09 s of wall-clock time; a single whole job takes 2.17 s. If two jobs do open the same port, exclusive=True will not let the second one open it silently: Could not exclusively lock port.

The port "hung" after reflashing. A port opened before reflashing leads nowhere: in the experiment a write to the old port object gave SerialException: write failed: [Errno 5] Input/output error. The STM32's own USB CDC behaves the same: on reset the device drops off the bus and comes back as a new device (section 115б.6). That is why Board.flash closes the port before flashing and opens it afterwards — a new object on the persistent name answered at once.

A command sent before the boot line is lost. The firmware has not enabled UART reception yet, and the test is already writing. In the experiment, get sent right after start-up got no reply — only boot thermo196 v1.4.2 and the factory-defaults cfg: line arrived. That is why flash and reset wait for the boot line and only then return.

Slow firmware does not make it within the timeout. A build with a start-up delay of 1.5 s (sensor warm-up) and --boot-timeout 1: the test did not start, and the report shows a setup error, not a failure: failed on setup with "boardlib.BoardTimeout: no /^boot thermo196 v(\S+)$/ within 1.0 s". With --boot-timeout 3 the test passed, but took 3.020 s: a slow start is paid twice — at flashing and at the reset before the test, and with a reset before every test — on every one. Take timeouts with a margin over the measured start-up time, and do not repeat a long device setup without need.

Permission denied. The port belongs to the dialout group with mode 0660, and the runner service runs as its own user. In the experiment the port was switched to root:dialout 0660: a user without the group got [Errno 13] could not open port … Permission denied, and after usermod -aG dialout it opened. Group membership applies to new sessions: the runner service is restarted after that (section 194.5). The ST-LINK programmer needs access to the USB device itself — usually the udev rules from the stlink-tools or openocd package provide it.

A reset did not restore the settings. In the bench platform the setpoint lives in RAM and a reset returns the factory 21.0. On the BluePill from 196 it is in Flash and survives a reset. A test that changes settings restores them itself — the finally in test_set_roundtrip.

The bench build went into a release. hang is a bench command; it must not be in a release build. It is enabled only by -DBENCH_HOOKS; the release is built separately, as in section 110.

GitHub pushed a job out of the queue. With concurrency the queue holds one pending job: a new pending job cancels the previous one (according to the GitHub Actions documentation, not tested in this section). For the bench this means that intermediate commits will not be checked on the board when changes come often. In GitLab resource_group keeps a queue of all jobs.

The board hung, and there is no watchdog. A reset through the programmer saves the test, but not the device in the field. The bench catches a hang with a timeout; the device needs its own watchdog (section 33а).

110б.8 Summary

The bench is a pipeline runner with the bench tag, with a board, a programmer and a port attached to it; the build runs on any runner, and the bench receives a ready image.
The board fixture hides everything about the board: flash once per run (bench, for the whole run), reset before each test (board), read lines with a timeout and wait for a regular expression. The port, baud rate, image and programmer are the options --port, --baud, --image, --flasher.
The run in QEMU: 12 tests collected, 9 selected, 9 passed in 1.55 s, the whole job took 2.17 s; of that, 1.104 s is the hang test with a timeout of 1 s. The image is 1964 bytes of code, the .bin is 1972 bytes.
A hung board: with a reset before each test 1 test of 3 fails, without a reset all 3 do (3.28 s instead of 1.41 s).
A slow start of 1.5 s: a 1 s timeout gives a setup error, with 3 s the test passed in 3.020 s.
Two boards in parallel take 2.09 s versus 2.17 s for one; one board gets one job at a time: resource_group in GitLab, concurrency in GitHub, exclusive=True in pyserial.
Pitfalls: port names follow the connection order (a udev rule by serial number), the old port after reflashing (Errno 5), a command before the boot line, dialout permissions (Errno 13).

Full sources

Everything you need to repeat the bench: the firmware, the board library, fixtures, tests, job scripts and pipeline files, the YAML check and the section's experiments. The thermostat logic thermo.c and thermo.h comes from section 196, the startup code and linker script from 115в.

plat_bench.c — the thermostat platform for the bench in QEMU: the boot line, ver, hang.

/* Раздел 110б: платформа термостата 196 для стенда в конвейере — QEMU netduinoplus2 (STM32F405,
 * USART1 и SysTick как у F407). Логика — proj196/thermo.c без изменений; модель комнаты — как в 115в.
 * Отличия от платформы 115в — то, что нужно стенду:
 *   строка «boot thermo196 v<версия>» после старта — по ней стенд понимает, что прибор жив;
 *   ver   — «ver=<версия>», версия приходит из конвейера: -DFW_VERSION="\"1.4.2\"";
 *   hang  — только в стендовой сборке (-DBENCH_HOOKS): прибор зависает намертво, как при ошибке;
 *   -DBOOT_DELAY_MS=1500 — медленный старт (прогрев датчика), строка приветствия приходит позже. */
#include "stm32f4xx.h"
#include "thermo.h"

#ifndef FW_VERSION
#define FW_VERSION "0.0.0-dev"
#endif
#ifndef BOOT_DELAY_MS
#define BOOT_DELAY_MS 0u
#endif
#define CORE_HZ 168000000u        /* частота ядра в QEMU netduinoplus2 */

static volatile uint32_t ms;
void SysTick_Handler(void) { ms++; }

/* ── модель комнаты в миллионных долях градуса, шаг 100 мс (как в 115в) ── */
static int32_t room_u = 15000000;         /* 15 °C */
static int heater;
static unsigned rng = 12345;

static void room_step(void)
{
    room_u += (5000000 - room_u) / 18000 + (heater ? 30000000 / 18000 : 0);
}

int16_t plat_read_temp(void)
{
    rng = rng * 1103515245u + 12345u;
    int noise = (int)((rng >> 16) % 3) - 1;
    return (int16_t)((room_u + 50000) / 100000 + noise);
}

void plat_relay(int on) { heater = on; }
uint32_t plat_now_ms(void) { return ms; }

static void putc1(char c)
{
    while (!(USART1->SR & USART_SR_TXE)) {
    }
    USART1->DR = (uint8_t)c;
}

void plat_puts(const char *s)
{
    for (; *s; s++) {
        if (*s == '\n')
            putc1('\r');
        putc1(*s);
    }
}

/* настройки — в ОЗУ: сброс возвращает заводские (на плате они во Flash и сброс переживают) */
static struct thermo_cfg saved;
static int saved_ok;
int plat_cfg_load(struct thermo_cfg *c) { if (!saved_ok) return -1; *c = saved; return 0; }
void plat_cfg_save(const struct thermo_cfg *c) { saved = *c; saved_ok = 1; }

static int same(const char *a, const char *b)
{
    while (*a && *a == *b) {
        a++;
        b++;
    }
    return *a == *b;
}

/* свои команды платформа выполняет сама, остальное целиком уходит в логику */
static char line[40];
static unsigned len, pass;

static void rx(struct thermo *t, char c)
{
    if (c == '\r' || c == '\n') {
        line[len] = 0;
        if (!pass && same(line, "ver")) {
            plat_puts("ver=" FW_VERSION "\n");
#ifdef BENCH_HOOKS
        } else if (!pass && same(line, "hang")) {
            __disable_irq();                              /* ни SysTick, ни UART — прибор молчит */
            for (;;) {
            }
#endif
        } else {
            for (unsigned i = 0; !pass && i < len; i++)
                thermo_rx(t, line[i]);
            thermo_rx(t, c);
        }
        len = pass = 0;
    } else if (pass) {
        thermo_rx(t, c);
    } else if (len < sizeof line - 1) {
        line[len++] = c;
    } else {
        for (unsigned i = 0; i < len; i++)
            thermo_rx(t, line[i]);
        thermo_rx(t, c);
        pass = 1;
        len = 0;
    }
}

int main(void)
{
    static struct thermo th;
    RCC->APB2ENR |= RCC_APB2ENR_USART1EN;
    USART1->BRR = 0x2D9;                                  /* APB2 84 МГц / 115 200 */
    USART1->CR1 = USART_CR1_UE | USART_CR1_TE | USART_CR1_RE;
    SysTick->LOAD = CORE_HZ / 1000u - 1u;
    SysTick->VAL = 0;
    SysTick->CTRL = SysTick_CTRL_CLKSOURCE_Msk | SysTick_CTRL_TICKINT_Msk | SysTick_CTRL_ENABLE_Msk;
#if BOOT_DELAY_MS
    while (ms < BOOT_DELAY_MS) {                          /* медленный старт */
    }
#endif
    plat_puts("boot thermo196 v" FW_VERSION "\n");
    thermo_init(&th);
    uint32_t next = ms;
    for (;;) {
        if (USART1->SR & USART_SR_RXNE)
            rx(&th, (char)USART1->DR);
        if ((int32_t)(ms - next) >= 0) {
            next += 100;
            room_step();
            thermo_tick(&th);
        }
    }
}

boardlib.py — the board for pytest: the port, lines with a timeout, flashers.

"""Раздел 110б: плата на столе для pytest — порт, строки с тайм-аутом, прошивка и сброс.

Board не знает, чем прошивают: это делает «прошивальщик» (Flasher). Для настоящей платы —
st-flash или openocd по серийному номеру программатора, для опытов без платы — QEMU, у которого
USART1 выведен на псевдотерминал (-serial pty): pyserial открывает его как обычный /dev/ttyUSB0."""
import os, re, socket, subprocess, time

import serial


class BoardTimeout(AssertionError):
    """Нужной строки не пришло за отведённое время. В тексте — что пришло вместо неё."""


class Board:
    def __init__(self, port, baud, flasher, boot_timeout=3.0):
        self.port, self.baud, self.flasher = port, baud, flasher
        self.boot_timeout = boot_timeout
        self.ser = None
        self.log = []                          # все строки от платы: (секунды от старта, текст)
        self.t0 = time.monotonic()

    # ── порт ──────────────────────────────────────────────────────────────
    def open(self):
        self.close()
        self.ser = serial.Serial(self.port, self.baud, timeout=0.1, exclusive=True)
        self.ser.reset_input_buffer()

    def close(self):
        if self.ser is not None:
            self.ser.close()
            self.ser = None

    def send(self, line):
        self.ser.write(line.encode("ascii") + b"\n")

    def readline(self, timeout):
        """Одна строка без \\r\\n или None, если за timeout секунд целой строки не пришло."""
        self.ser.timeout = timeout
        raw = self.ser.read_until(b"\n")
        if not raw.endswith(b"\n"):
            return None
        s = raw.rstrip(b"\r\n").decode("utf-8", errors="replace")
        self.log.append((round(time.monotonic() - self.t0, 3), s))
        return s

    def expect(self, pattern, timeout=1.0):
        """Ждать строку, подходящую под регулярное выражение; чужие строки пропустить."""
        rx = re.compile(pattern)
        end = time.monotonic() + timeout
        seen = []
        while True:
            left = end - time.monotonic()
            if left <= 0:
                raise BoardTimeout("нет /%s/ за %.1f с; пришло: %r" % (pattern, timeout, seen[-5:]))
            s = self.readline(left)
            if s is None:
                continue
            m = rx.search(s)
            if m:
                return m
            seen.append(s)

    def ask(self, cmd, pattern, timeout=1.0):
        """Команда → ответ по образцу. Что пришло до команды, выбросить."""
        self.ser.reset_input_buffer()
        self.send(cmd)
        return self.expect(pattern, timeout)

    # ── прошивка и сброс ──────────────────────────────────────────────────
    def flash(self, image):
        self.close()                           # порт мог исчезнуть вместе с платой — открыть заново
        self.flasher.flash(image)
        self.open()
        return self.expect(r"^boot thermo196 v(\S+)$", self.boot_timeout)

    def reset(self):
        self.ser.reset_input_buffer()
        self.flasher.reset()
        return self.expect(r"^boot thermo196 v(\S+)$", self.boot_timeout)


class QemuFlasher:
    """«Плата» — QEMU netduinoplus2. Прошить = перезапустить QEMU с образом; сбросить = system_reset.
    Порт: QEMU создаёт /dev/pts/N (номер каждый раз свой), стенд держит постоянную ссылку port."""

    def __init__(self, port, workdir="/tmp"):
        self.port, self.proc = port, None
        self.mon = os.path.join(workdir, os.path.basename(port) + ".mon")

    def flash(self, image):
        self.stop()
        self.proc = subprocess.Popen(
            ["qemu-system-arm", "-M", "netduinoplus2", "-nographic", "-kernel", image,
             "-serial", "pty", "-monitor", "unix:%s,server=on,wait=off" % self.mon],
            stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
        m = re.search(r"char device redirected to (/dev/pts/\d+)", self.proc.stdout.readline())
        if not m:
            raise RuntimeError("QEMU не назвал псевдотерминал")
        self.pty = m.group(1)
        tmp = self.port + ".new"               # подменить ссылку разом, как это делает udev
        if os.path.lexists(tmp):
            os.remove(tmp)
        os.symlink(self.pty, tmp)
        os.replace(tmp, self.port)

    def reset(self):
        with socket.socket(socket.AF_UNIX) as s:
            s.connect(self.mon)
            s.sendall(b"system_reset\n")
            time.sleep(0.05)                   # дать монитору принять команду до закрытия сокета

    def stop(self):
        if self.proc is not None:
            self.proc.kill()
            self.proc.wait()
            self.proc = None


class StFlashFlasher:
    """Настоящая плата через ST-LINK и stlink-tools: программатор выбирается по серийному номеру."""

    def __init__(self, probe):
        self.probe = probe

    def flash(self, image):
        binary = image[:-4] + ".bin" if image.endswith(".elf") else image
        subprocess.run(["st-flash", "--serial", self.probe, "--reset", "write", binary, "0x08000000"],
                       check=True, timeout=60)

    def reset(self):
        subprocess.run(["st-flash", "--serial", self.probe, "reset"], check=True, timeout=20)

    def stop(self):
        pass


class OpenOcdFlasher:
    """Настоящая плата через openocd 0.12: тот же ST-LINK, но прошивка прямо из ELF."""

    def __init__(self, probe, target="target/stm32f1x.cfg"):
        self.base = ["openocd", "-f", "interface/stlink.cfg", "-c", "adapter serial %s" % probe, "-f", target]

    def flash(self, image):
        subprocess.run(self.base + ["-c", "program %s verify reset exit" % image], check=True, timeout=60)

    def reset(self):
        subprocess.run(self.base + ["-c", "init; reset run; exit"], check=True, timeout=20)

    def stop(self):
        pass


def make_flasher(name, port, probe):
    if name == "qemu":
        return QemuFlasher(port)
    if name == "st-flash":
        return StFlashFlasher(probe)
    if name == "openocd":
        return OpenOcdFlasher(probe)
    raise ValueError("неизвестный --flasher: %s" % name)

conftest.py — command-line options and the fixtures bench and board.

"""Раздел 110б: фикстуры стенда. Один набор тестов — на любую плату: всё про плату приходит
параметрами командной строки.

    pytest bench --port /dev/bench/thermo-a --image build/thermo.elf --flasher st-flash --probe 066DFF...
    pytest bench --port /tmp/bench/thermo-a --image fw.elf --flasher qemu"""
import os

import pytest

from boardlib import Board, make_flasher


def pytest_addoption(parser):
    g = parser.getgroup("board", "плата на стенде")
    g.addoption("--port", default=os.environ.get("BOARD_PORT"), help="порт платы: /dev/bench/..., COM5")
    g.addoption("--baud", type=int, default=115200, help="скорость порта")
    g.addoption("--image", default=os.environ.get("FW_IMAGE"), help="что прошить: .elf или .bin")
    g.addoption("--flasher", default="st-flash", choices=["st-flash", "openocd", "qemu"])
    g.addoption("--probe", default=os.environ.get("PROBE_SERIAL", ""), help="серийный номер программатора")
    g.addoption("--fw-version", default=os.environ.get("FW_VERSION"), help="какую версию ждать от платы")
    g.addoption("--boot-timeout", type=float, default=3.0, help="сколько ждать строку старта, с")
    g.addoption("--no-reset", action="store_true", help="не сбрасывать плату перед каждым тестом")


@pytest.fixture(scope="session")
def bench(request):
    """Прошить один раз на весь прогон: прошивка — самое долгое."""
    o = request.config.option
    if not o.port or not o.image:
        pytest.exit("нужны --port и --image (или BOARD_PORT и FW_IMAGE)", returncode=2)
    flasher = make_flasher(o.flasher, o.port, o.probe)
    request.addfinalizer(flasher.stop)     # и когда плата не стартовала: QEMU не должен остаться
    b = Board(o.port, o.baud, flasher, o.boot_timeout)
    request.addfinalizer(b.close)
    b.boot = b.flash(o.image)
    return b


@pytest.fixture
def board(bench, request):
    """Каждому тесту — плата сразу после сброса: чужое зависание или уставка не переходят дальше."""
    if not request.config.option.no_reset:
        bench.boot = bench.reset()
    yield bench


@pytest.fixture
def fw_version(request):
    return request.config.option.fw_version

pytest.ini — suite settings.

[pytest]
markers =
    demo: опыт раздела 110б — зависание без pytest.raises, в обычный прогон не входит
addopts = -m "not demo" -p no:cacheprovider
junit_family = xunit2

test_thermo.py — thermostat tests.

"""Раздел 110б: термостат 196 на стенде. Каждый тест: команда — строка — проверка."""
import pytest

from boardlib import BoardTimeout

GET = r"^t=(-?\d+\.\d) set=(-?\d+\.\d) (\S+)$"


def test_version(board, fw_version):
    ver = board.ask("ver", r"^ver=(\S+)$").group(1)
    assert ver == board.boot.group(1)            # строка старта и команда говорят одно
    if fw_version:
        assert ver == fw_version                 # на плате именно то, что собрал конвейер


def test_get(board):
    t, sp, state = board.ask("get", GET).groups()
    assert 5.0 <= float(t) <= 35.0
    assert sp == "21.0"                          # после сброса — заводская уставка
    assert state in ("ждёт", "нагрев")


def test_set_roundtrip(board):
    try:
        assert board.ask("set 22.5", r"^(ok|\?.*)$").group(1) == "ok"
        assert board.ask("get", GET).group(2) == "22.5"
    finally:
        board.ask("set 21.0", r"^ok$")           # на плате уставка во Flash: сброс её не вернёт


@pytest.mark.parametrize("cmd", ["foo", "set 99", "set", "hyst 9"])
def test_bad_command(board, cmd):
    assert board.ask(cmd, r"^(ok|\?.*)$").group(1).startswith("?")


def test_long_line(board):
    assert board.ask("x" * 40, r"^\?.*$").group(0) == "? строка длиннее 31 символа"


def test_hang_then_reset(board):
    board.send("hang")
    with pytest.raises(BoardTimeout):
        board.ask("get", GET, timeout=1.0)       # плата молчит — тест не ждёт вечно
    board.reset()
    assert board.ask("get", GET)                 # после сброса снова отвечает

test_cascade.py — the "hung board without a reset" experiment.

"""Раздел 110б, опыт: тест, после которого плата висит. Запуск: pytest bench -m demo [--no-reset]."""
import pytest

from test_thermo import GET

pytestmark = pytest.mark.demo


def test_a_hangs(board):
    board.send("hang")
    board.ask("get", GET, timeout=1.0)           # так выглядит настоящее зависание: ответа нет


def test_b_get(board):
    board.ask("get", GET)


def test_c_version(board):
    board.ask("ver", r"^ver=\S+$")

build_fw.sh — the bench build: the build job.

#!/bin/sh
# Раздел 110б: стендовая сборка термостата — задание build и опыт раздела.
#   OUT — куда положить образ; FW_VERSION — версия (в конвейере из git describe); EXTRA — свои флаги.
#   BENCH_HOOKS включает команду hang: такая сборка — только для стенда, не для выпуска.
set -eu
OUT=${OUT:-build}
FW_VERSION=${FW_VERSION:-$(git describe --tags --always 2>/dev/null || echo 0.0.0-dev)}
mkdir -p "$OUT"
arm-none-eabi-gcc -mcpu=cortex-m4 -mthumb -Os -Wall -Wextra -Werror -ffreestanding \
    -ffunction-sections -fdata-sections -DSTM32F407xx -DBENCH_HOOKS \
    -DFW_VERSION="\"$FW_VERSION\"" ${EXTRA:-} -Ifw -Ifw/cmsis/device -Ifw/cmsis/core \
    fw/start_f4.c fw/plat_bench.c fw/thermo.c -T fw/f4.ld -nostdlib -Wl,--gc-sections \
    -o "$OUT/thermo.elf"
arm-none-eabi-objcopy -O binary "$OUT/thermo.elf" "$OUT/thermo.bin"
arm-none-eabi-size "$OUT/thermo.elf"
echo "FW_VERSION=$FW_VERSION" > "$OUT/version.env"

bench_job.sh — the bench job.

#!/bin/sh
# Раздел 110б: задание стенда. BOARD_PORT, FLASHER, PROBE_SERIAL — из переменных задания,
# build/ — образ и версия от задания build. Код выхода — код pytest.
set -eu
. build/version.env
mkdir -p report
set +e
python3 -m pytest bench --port "$BOARD_PORT" --baud 115200 --image build/thermo.elf \
    --flasher "$FLASHER" --probe "${PROBE_SERIAL:-}" --fw-version "$FW_VERSION" \
    --junitxml report/junit.xml -o junit_suite_name="hil-$(basename "$BOARD_PORT")" \
    > report/pytest.txt 2>&1
rc=$?
set -e
cat report/pytest.txt
exit $rc

gitlab-ci.yml — the GitLab pipeline; in the repository it is named .gitlab-ci.yml.

# Раздел 110б: .gitlab-ci.yml — сборка в контейнере на любом исполнителе,
# тесты на плате — только на исполнителе с тегом bench, к которому она подключена.
stages: [build, bench]

build:
  stage: build
  image: registry.example.com/tools/stm-arm:1
  tags: [docker]
  variables:
    GIT_DEPTH: "0"                    # git describe нужна история с тегами
  script:
    - sh ci/build_fw.sh
  artifacts:
    paths: [build/]
    expire_in: 1 week

.hil:
  stage: bench
  needs: [build]
  timeout: 10 minutes
  interruptible: false                # не обрывать на середине прошивки
  variables:
    FLASHER: st-flash
  script:
    - sh ci/bench_job.sh
  artifacts:
    when: always                      # отчёт нужен как раз когда тесты упали
    paths: [report/]
    reports:
      junit: report/junit.xml
    expire_in: 2 weeks

hil-thermo-a:
  extends: .hil
  tags: [bench, thermo-a]             # исполнитель, к которому подключена плата A
  resource_group: bench-thermo-a      # одна плата — одно задание за раз
  variables:
    BOARD_PORT: /dev/bench/thermo-a
    PROBE_SERIAL: $PROBE_THERMO_A     # серийный номер ST-LINK — в настройках CI/CD

hil-thermo-b:
  extends: .hil
  tags: [bench, thermo-b]
  resource_group: bench-thermo-b
  variables:
    BOARD_PORT: /dev/bench/thermo-b
    PROBE_SERIAL: $PROBE_THERMO_B

bench.yml — the same for GitHub Actions, .github/workflows/bench.yml.

# Раздел 110б: .github/workflows/bench.yml — то же для GitHub Actions.
name: bench
on:
  push:
    branches: [main]
  pull_request:

jobs:
  build:
    runs-on: ubuntu-latest
    container: registry.example.com/tools/stm-arm:1
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
      - run: sh ci/build_fw.sh
      - uses: actions/upload-artifact@v4
        with:
          name: fw
          path: build/

  hil:
    needs: build
    runs-on: [self-hosted, bench, thermo-a]
    timeout-minutes: 10
    concurrency:
      group: bench-thermo-a
      cancel-in-progress: false
    env:
      BOARD_PORT: /dev/bench/thermo-a
      FLASHER: st-flash
      PROBE_SERIAL: ${{ vars.PROBE_THERMO_A }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/download-artifact@v4
        with:
          name: fw
          path: build/
      - run: sh ci/bench_job.sh
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: report-thermo-a
          path: report/

99-bench.rules — a udev rule: persistent port names by serial number.

# Раздел 110б: /etc/udev/rules.d/99-bench.rules на машине-исполнителе.
# Порт платы называется по серийному номеру её ST-LINK, а не по порядку подключения.
# Свои номера: udevadm info -a -n /dev/ttyACM0 | grep '{serial}'
SUBSYSTEM=="tty", ATTRS{idVendor}=="0483", ATTRS{serial}=="SN_THERMO_A", SYMLINK+="bench/thermo-a", GROUP="dialout", MODE="0660"
SUBSYSTEM=="tty", ATTRS{idVendor}=="0483", ATTRS{serial}=="SN_THERMO_B", SYMLINK+="bench/thermo-b", GROUP="dialout", MODE="0660"

ci_lint.py — a check of the pipeline files without the service.

"""Раздел 110б: проверка файлов конвейера без сервиса — YAML читается, ключи из списка известных.

    python ci_lint.py ci/gitlab-ci.yml ci/bench.yml

Это не замена `gitlab-ci-lint` и `actionlint`: проверяются опечатки в ключах, ссылки needs/extends,
этапы и то, ради чего файлы писались: тесты на плате идут только на исполнителе с тегом bench,
по одному заданию на плату, отчёт JUnit сохраняется и при падении."""
import sys

import yaml

GL_TOP = {"stages", "variables", "default", "include", "workflow", "image", "services", "cache",
          "before_script", "after_script"}
GL_JOB = {"stage", "image", "services", "tags", "variables", "script", "before_script", "after_script",
          "artifacts", "needs", "dependencies", "rules", "only", "except", "when", "timeout", "retry",
          "interruptible", "resource_group", "extends", "cache", "environment", "parallel",
          "allow_failure", "coverage", "trigger", "release", "secrets", "id_tokens", "hooks", "pages"}
GL_ART = {"paths", "exclude", "expire_in", "expose_as", "name", "public", "reports", "untracked", "when"}
GH_TOP = {"name", "run-name", "on", "jobs", "env", "permissions", "concurrency", "defaults"}
GH_JOB = {"name", "needs", "runs-on", "container", "services", "steps", "env", "if", "timeout-minutes",
          "concurrency", "strategy", "outputs", "permissions", "environment", "defaults",
          "continue-on-error", "uses", "with", "secrets"}
GH_STEP = {"name", "id", "uses", "run", "with", "env", "if", "shell", "working-directory",
           "continue-on-error", "timeout-minutes"}
errors = []


def need(cond, msg):
    if not cond:
        errors.append(msg)


def gitlab(path):
    doc = yaml.safe_load(open(path, encoding="utf-8"))
    stages = doc.get("stages", ["build", "test", "deploy"])
    jobs = {k: v for k, v in doc.items() if k not in GL_TOP}
    for name, job in jobs.items():
        need(isinstance(job, dict), "%s: задание должно быть словарём" % name)
        need(not set(job) - GL_JOB, "%s: неизвестные ключи %s" % (name, sorted(set(job) - GL_JOB)))
        need(not set(job.get("artifacts", {})) - GL_ART, "%s: неизвестные ключи artifacts" % name)
    full = {}
    for name, job in jobs.items():                       # extends: свои ключи поверх шаблона
        base = dict(jobs.get(job.get("extends"), {}))
        need("extends" not in job or job["extends"] in jobs, "%s: нет шаблона %s" % (name, job.get("extends")))
        base.update(job)
        full[name] = base
    real = {k: v for k, v in full.items() if not k.startswith(".")}
    for name, job in real.items():
        need("script" in job, "%s: нет script" % name)
        need(job.get("stage", "test") in stages, "%s: этапа %s нет в stages" % (name, job.get("stage")))
        need(all(n in real for n in job.get("needs", [])), "%s: needs ссылается на несуществующее" % name)
        if "bench" in job.get("tags", []):
            need("resource_group" in job, "%s: плата без resource_group — два задания на одну плату" % name)
            need(job.get("artifacts", {}).get("when") == "always", "%s: отчёт не сохранится при падении" % name)
            need("junit" in job.get("artifacts", {}).get("reports", {}), "%s: нет отчёта JUnit" % name)
    bench = sorted(n for n, j in real.items() if "bench" in j.get("tags", []))
    groups = sorted(real[n]["resource_group"] for n in bench if "resource_group" in real[n])
    print("gitlab_stages=%s" % " ".join(stages))
    print("gitlab_jobs=%s" % " ".join(sorted(real)))
    print("gitlab_bench_jobs=%s" % " ".join(bench))
    print("gitlab_resource_groups=%s" % " ".join(groups))
    need(len(groups) == len(set(groups)), "две платы в одной resource_group")


def github(path):
    doc = yaml.safe_load(open(path, encoding="utf-8"))
    if True in doc:                                      # YAML 1.1: голое on читается как True
        doc["on"] = doc.pop(True)
    need(not set(doc) - GH_TOP, "workflow: неизвестные ключи %s" % sorted(set(doc) - GH_TOP))
    jobs = doc.get("jobs", {})
    for name, job in jobs.items():
        need(not set(job) - GH_JOB, "%s: неизвестные ключи %s" % (name, sorted(set(job) - GH_JOB)))
        needs = job.get("needs", [])
        needs = [needs] if isinstance(needs, str) else needs
        need(all(n in jobs for n in needs), "%s: needs ссылается на несуществующее" % name)
        for st in job.get("steps", []):
            need(not set(st) - GH_STEP, "%s: неизвестные ключи шага %s" % (name, sorted(set(st) - GH_STEP)))
            need(("uses" in st) != ("run" in st), "%s: в шаге нужен ровно один из uses и run" % name)
    hil = [n for n, j in jobs.items() if isinstance(j.get("runs-on"), list) and "self-hosted" in j["runs-on"]]
    for n in hil:
        need("bench" in jobs[n]["runs-on"], "%s: свой исполнитель без метки bench" % n)
        need(jobs[n].get("concurrency", {}).get("cancel-in-progress") is False, "%s: прошивку оборвут" % n)
    print("github_on=%s" % " ".join(sorted(doc["on"])))
    print("github_jobs=%s" % " ".join(sorted(jobs)))
    print("github_self_hosted=%s" % " ".join("%s:%s" % (n, ",".join(jobs[n]["runs-on"])) for n in hil))


gitlab(sys.argv[1])
github(sys.argv[2])
for e in errors:
    print("error: " + e)
print("lint_errors=%d" % len(errors))
sys.exit(1 if errors else 0)

exp110b.py — port experiments: names, the old port, locking, dialout.

"""Раздел 110б: опыты с портом стенда (stm-arm:1, pyserial 3.5). Печатает строки ключ=значение.

    python3 exp110b.py IMAGE

1) номер псевдотерминала зависит от порядка запуска плат, ссылка стенда — нет;
2) команда, посланная до строки старта, теряется;
3) старый открытый порт после перепрошивки: запись падает, помогает только открыть заново;
4) второй открывающий тот же порт с exclusive=True — отказ (два задания на одну плату);
5) права: порт root:dialout 0660 — пользователь без группы dialout не открывает, с группой — открывает."""
import os, subprocess, sys, time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "bench"))
import serial  # noqa: E402
from boardlib import Board, QemuFlasher  # noqa: E402

IMAGE = sys.argv[1]
PORT = "/tmp/bench/exp"
os.makedirs("/tmp/bench", exist_ok=True)


def kv(k, v):
    print("%s=%s" % (k, v), flush=True)


# 1. имя порта: номер псевдотерминала — порядок «подключения», как у /dev/ttyACM0 и ttyACM1
fx, fy = QemuFlasher("/tmp/bench/x"), QemuFlasher("/tmp/bench/y")
order = []
for first, second in ((fx, fy), (fy, fx)):
    first.flash(IMAGE)
    second.flash(IMAGE)
    order.append("x=%s y=%s" % (fx.pty, fy.pty))
    fx.stop()
    fy.stop()
kv("exp_order_xy", order[0])
kv("exp_order_yx", order[1])

fl = QemuFlasher(PORT)
b = Board(PORT, 115200, fl)
b.flash(IMAGE)
kv("exp_link", "%s -> %s" % (PORT, os.readlink(PORT)))

# 2. команда до строки старта: прошивка ещё не слушает порт
fl.flash(IMAGE)
early = serial.Serial(PORT, 115200, timeout=0.1)
early.write(b"get\n")
got, end = [], time.monotonic() + 1.0
while time.monotonic() < end:
    ln = early.readline()
    if ln.endswith(b"\n"):
        got.append(ln.decode("utf-8").strip())
kv("exp_early", " | ".join(got))
early.close()

# 3. старый открытый порт после перепрошивки
old = serial.Serial(PORT, 115200, timeout=0.5)
b.flash(IMAGE)                                   # плата перепрошита, старый объект порта остался
try:
    old.write(b"get\n")
    kv("exp_stale", "read %d bytes" % len(old.read(64)))
except (serial.SerialException, OSError) as e:
    kv("exp_stale", "%s: %s" % (type(e).__name__, e))
old.close()
kv("exp_reopen", b.ask("get", r"^t=\S+ set=\S+ \S+$").group(0))
b.close()

# 4. два открывающих на один порт
first = serial.Serial(PORT, 115200, timeout=0.2, exclusive=True)
try:
    serial.Serial(PORT, 115200, timeout=0.2, exclusive=True)
    kv("exp_second_open", "opened")
except serial.SerialException as e:
    kv("exp_second_open", str(e))
first.close()

# 5. права dialout: так выглядит /dev/ttyACM0 на Linux
real = os.path.realpath(PORT)
subprocess.run(["chgrp", "dialout", real], check=True)
os.chmod(real, 0o660)
st = os.stat(real)
kv("exp_perm", "%o group=%d" % (st.st_mode & 0o777, st.st_gid))
subprocess.run(["useradd", "-M", "ci"], check=True)
probe = ("import serial\ntry:\n    serial.Serial('%s', 115200).close(); print('opened')\n"
         "except serial.SerialException as e:\n    print(e)" % PORT)
env = dict(os.environ)


def as_ci():
    r = subprocess.run(["su", "ci", "-s", "/bin/sh", "-c", "python3 -c \"%s\"" % probe.replace('"', '\\"')],
                       capture_output=True, text=True, env=env)
    return (r.stdout + r.stderr).strip()


kv("exp_groups_before", subprocess.run(["id", "-nG", "ci"], capture_output=True, text=True).stdout.strip())
kv("exp_open_before", as_ci())
subprocess.run(["usermod", "-aG", "dialout", "ci"], check=True)
kv("exp_groups_after", subprocess.run(["id", "-nG", "ci"], capture_output=True, text=True).stdout.strip())
kv("exp_open_after", as_ci())
fl.stop()
kv("exp_done", 1)

run110b.sh — the whole run of the section in a container.

#!/bin/sh
# Раздел 110б: стенд термостата в stm-arm:1 — сборка, задание стенда, опыты. Печатает ключ=значение.
# /v — этот каталог (hil110b), /p — proj196 (thermo.c, thermo.h), /s — python115c (start_f4.c, f4.ld),
# /di, /ci — заголовки CMSIS (STM32F4xx DFP 2.17.1, CMSIS 5.9.0), /py — pyserial 3.5 и pytest 9.1.1.
set -u
export LC_ALL=C.UTF-8 PYTHONIOENCODING=utf-8 PYTHONPATH=/py PYTHONDONTWRITEBYTECODE=1
R=/tmp/repo; rm -rf $R /tmp/bench; mkdir -p $R/fw/cmsis/device $R/fw/cmsis/core /tmp/bench
cp /v/fw/plat_bench.c /p/thermo.c /p/thermo.h /s/start_f4.c /s/f4.ld $R/fw/
cp /di/stm32f4xx.h /di/stm32f407xx.h /di/system_stm32f4xx.h $R/fw/cmsis/device/
cp /ci/*.h $R/fw/cmsis/core/
cp -r /v/bench /v/ci $R/
cd $R
echo "== versions"
echo "python=$(python3 --version | cut -d' ' -f2)"
echo "pyserial=$(python3 -c 'import serial; print(serial.__version__)')"
echo "pytest=$(python3 -c 'import pytest; print(pytest.__version__)')"
echo "qemu=$(qemu-system-arm --version | head -1)"
echo "gcc=$(arm-none-eabi-gcc --version | head -1)"

summary() {   # итог из отчёта JUnit: тесты, упавшие, ошибки, время
    python3 - "$1" <<'PY'
import sys, xml.etree.ElementTree as ET
r = ET.parse(sys.argv[1]).getroot()
s = r if r.tag == "testsuite" else r.find("testsuite")
print("junit_suite=%s tests=%s failures=%s errors=%s skipped=%s time=%s" % (
    s.get("name"), s.get("tests"), s.get("failures"), s.get("errors"), s.get("skipped"), s.get("time")))
for c in s.iter("testcase"):
    bad = c.find("failure") if c.find("failure") is not None else c.find("error")
    print("case %s %s %s%s" % (c.get("name"), c.get("time"), "FAIL" if bad is not None else "ok",
                               (" | " + bad.get("message", "").splitlines()[0][:150]) if bad is not None else ""))
PY
}

echo "== build"
FW_VERSION=1.4.2 sh ci/build_fw.sh 2>&1; echo "build_exit=$?"
OUT=build-slow FW_VERSION=1.4.2 EXTRA=-DBOOT_DELAY_MS=1500 sh ci/build_fw.sh > /dev/null 2>&1; echo "build_slow_exit=$?"
ls -l build/thermo.bin | awk '{print "bin_bytes=" $5}'
cat build/version.env

echo "== job thermo-a"
t0=$(date +%s.%N)
BOARD_PORT=/tmp/bench/thermo-a FLASHER=qemu sh ci/bench_job.sh; echo "job_exit=$?"
t1=$(date +%s.%N)
echo "job_wall=$(python3 -c "print('%.2f' % ($t1 - $t0))")"
summary report/junit.xml
cp report/junit.xml /tmp/junit-a.xml

echo "== cascade reset"
python3 -m pytest bench -m demo --port /tmp/bench/c1 --image build/thermo.elf --flasher qemu -q \
    --junitxml /tmp/c1.xml > /tmp/c1.txt 2>&1; echo "cascade_reset_exit=$?"
tail -1 /tmp/c1.txt; summary /tmp/c1.xml
echo "== cascade no-reset"
python3 -m pytest bench -m demo --port /tmp/bench/c2 --image build/thermo.elf --flasher qemu -q --no-reset \
    --junitxml /tmp/c2.xml > /tmp/c2.txt 2>&1; echo "cascade_noreset_exit=$?"
tail -1 /tmp/c2.txt; summary /tmp/c2.xml

for T in 1 3; do
    echo "== slow boot, boot-timeout $T"
    python3 -m pytest bench -k version --port /tmp/bench/slow$T --image build-slow/thermo.elf --flasher qemu -q \
        --boot-timeout $T --junitxml /tmp/slow$T.xml > /tmp/slow$T.txt 2>&1; echo "slow_${T}_exit=$?"
    tail -1 /tmp/slow$T.txt; summary /tmp/slow$T.xml
done

echo "== two boards"
for B in a b; do rm -rf /tmp/r$B; cp -r $R /tmp/r$B; rm -rf /tmp/r$B/report; done
t0=$(date +%s.%N)
(cd /tmp/ra && BOARD_PORT=/tmp/bench/thermo-a FLASHER=qemu sh ci/bench_job.sh > /tmp/ja.txt 2>&1; echo $? > /tmp/ja.rc) &
(cd /tmp/rb && BOARD_PORT=/tmp/bench/thermo-b FLASHER=qemu sh ci/bench_job.sh > /tmp/jb.txt 2>&1; echo $? > /tmp/jb.rc) &
wait
t1=$(date +%s.%N)
echo "two_exit=$(cat /tmp/ja.rc) $(cat /tmp/jb.rc)"
echo "two_wall=$(python3 -c "print('%.2f' % ($t1 - $t0))")"
summary /tmp/ra/report/junit.xml | head -1
summary /tmp/rb/report/junit.xml | head -1

echo "left_qemu_tests=$(pgrep -c qemu-system-arm)"
echo "== port"
python3 /v/exp110b.py $R/build/thermo.elf
echo "left_qemu=$(pgrep -c qemu-system-arm)"
echo "done=1"

Next: section 111 "Containers and infrastructure" — what an image and a container are from the inside, Compose and a reverse proxy for the services that grow around a device.

Open in the interactive handbook