Все разделы › XII. Сети и DevOps для инженера

STM32F103, STM32F407 · уровень средний · 50 мин чтения

110б. Плата на столе в конвейере: pytest, последовательный порт и свой исполнитель

Конвейер из разделов 110 и 110а собирает образ, гоняет модульные тесты на компьютере и следит за размером. В таблице уровней из раздела 184.1 есть и следующая ступень — «на плате со стендом», но до сих пор она оставалась строчкой в таблице. Здесь она собрана целиком: плата лежит на столе рядом с небольшим компьютером, этот компьютер — исполнитель конвейера с тегом bench, и на каждую правку он прошивает плату и гоняет тесты через последовательный порт.

Тесты пишутся на pytest. Всё, что касается платы, спрятано в фикстуру board: прошить, открыть порт, дождаться строки старта, сбросить перед тестом. Сами тесты получаются по три строки: команда — ответ — проверка. Порт, скорость, образ и программатор приходят параметрами командной строки, поэтому один и тот же набор обслуживает любую плату на столе.

Подопытный прибор — термостат из раздела 196 с консолью по UART. Скрипт из раздела 115в уже гонял его через pyserial; там это был скрипт испытаний, здесь — тесты в конвейере. Все прогоны раздела сделаны без живой платы: роль платы играет прошивка в QEMU, у которого USART1 выведен на псевдотерминал. Для pyserial псевдотерминал выглядит так же, как /dev/ttyUSB0, поэтому код тестов и фикстуры один и тот же. Прошивка через st-flash и openocd показана в коде, но с настоящим программатором не запускалась.

Опоры: логика термостата — файл раздела 196 без единой правки; платформа для QEMU — по образцу раздела 115в; стартовый код и скрипт компоновщика — из 115в; заголовки CMSIS для F407. Опыты — в образе stm-arm:1: Python 3.11.2, pyserial 3.5, pytest 9.1.1, QEMU 7.2.22 (машина netduinoplus2, это STM32F405 с той же периферией USART1 и SysTick, что у F407), arm-none-eabi-gcc 12.2.1. Команды st-flash и openocd — как в разделе 35б. Идея — статья в блоге Golioth об автоматических тестах на железе с pytest (blog.golioth.io); текст, код и числа здесь свои.

Как это называется в документации

Оригинал Перевод

hardware-in-the-loop (HIL), test bench стенд: тесты гоняются на настоящей плате

fixture, conftest.py фикстура: подготовка и уборка для теста; файл с фикстурами набора

command-line option, pytest_addoption параметр командной строки pytest

session scope, function scope фикстура на весь прогон, на каждый тест

self-hosted runner, tags, labels свой исполнитель; теги в GitLab, метки в GitHub

resource_group, concurrency группа: не больше одного задания за раз

JUnit XML report отчёт о тестах в формате JUnit, его понимают GitLab и GitHub

pseudo-terminal (pty) псевдотерминал: пара «программа — порт» без железа

udev rule, SYMLINK правило udev: постоянное имя устройства

dialout группа Linux, которой принадлежат последовательные порты

Оригинал	Перевод
hardware-in-the-loop (HIL), test bench	стенд: тесты гоняются на настоящей плате
fixture, `conftest.py`	фикстура: подготовка и уборка для теста; файл с фикстурами набора
command-line option, `pytest_addoption`	параметр командной строки pytest
session scope, function scope	фикстура на весь прогон, на каждый тест
self-hosted runner, tags, labels	свой исполнитель; теги в GitLab, метки в GitHub
`resource_group`, `concurrency`	группа: не больше одного задания за раз
JUnit XML report	отчёт о тестах в формате JUnit, его понимают GitLab и GitHub
pseudo-terminal (pty)	псевдотерминал: пара «программа — порт» без железа
udev rule, `SYMLINK`	правило udev: постоянное имя устройства
`dialout`	группа Linux, которой принадлежат последовательные порты

110б.1 Где стоит стенд и что он ловит

Модульные тесты из 110 и 184 не видят железа: таймингов, прерываний, настоящего UART, того, что прибор вообще стартует после прошивки. Стенд видит именно это, но платит временем и тем, что плата одна. Отсюда устройство конвейера:

Задание	Где идёт	Что делает
`build`	любой исполнитель, в контейнере	собирает стендовый образ, кладёт его в артефакт
`hil-thermo-a`	только исполнитель с тегами `bench` и `thermo-a`	прошивает плату A, гоняет pytest, отдаёт отчёт JUnit
`hil-thermo-b`	исполнитель с тегами `bench` и `thermo-b`	то же на плате B

Исполнитель с платой — обычная машина: небольшой компьютер или Raspberry Pi с программатором ST-LINK и переходником USB–UART (или портом ST-LINK, раздел 115б.1). На ней стоят служба исполнителя, Python с pytest и pyserial и программа прошивки. Задание идёт прямо на этой машине, без контейнера: так проще всего добраться до USB.

Код стенда делится на три слоя, и только нижний знает, есть ли плата на самом деле:

Файл	Что в нём
`test_thermo.py`	тесты: команда, ответ, проверка
`conftest.py`	параметры командной строки и фикстуры `bench`, `board`
`boardlib.py`	`Board` — порт и строки; прошивальщики: st-flash, openocd, QEMU

110б.2 Плата для теста: строка с тайм-аутом и ожидание по образцу

Главное в работе с прибором из теста — никогда не ждать вечно. Зависшая плата не должна вешать конвейер: каждое чтение идёт с тайм-аутом, и если нужной строки нет, тест падает с понятным сообщением — что ждали и что пришло вместо этого.

Чтение одной строки. pyserial читает до \n или до конца тайм-аута; обрывок без \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

Ожидание по регулярному выражению. Прибор может прислать что-то без запроса (в термостате — «АВАРИЯ: датчик» или «cfg: заводские» после старта), поэтому чужие строки пропускаются, а последние пять попадают в текст ошибки:

    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)

Сообщение BoardTimeout попадает в отчёт целиком. Тайм-аут в самом тесте pytest записывает как падение (failure), а в подготовке, то есть в фикстуре, — как ошибку (error); разница видна в 110б.7 на медленном старте. В опыте со стендом так выглядит строка отчёта о зависшей плате:

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

Порт открывается с exclusive=True: pyserial ставит на него блокировку, и второй открывающий получает отказ сразу, а не делит поток байтов (раздел 115б.7). Прошивка и сброс — через прошивальщика, после них стенд ждёт строку старта:

    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 Прошить и сбросить: st-flash, openocd и QEMU

У каждого прошивальщика два действия: flash(image) и reset(). Для настоящей платы — программатор, выбранный по серийному номеру: на столе может лежать несколько ST-LINK, и прошить надо именно свой.

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 берёт только .bin и адрес (раздел 35б.3), поэтому сборка кладёт рядом с .elf и .bin. openocd прошивает прямо из .elf той же строкой program … verify reset exit; серийный номер программатора задаёт команда 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

Без платы её место занимает QEMU. «Прошить» — значит перезапустить QEMU с новым образом, «сбросить» — послать system_reset в его монитор через сокет. Порт QEMU создаёт сам (-serial pty) и называет его в первой строке вывода, например /dev/pts/0; номер каждый раз свой, поэтому стенд держит постоянную ссылку и подменяет её одним os.replace — так же, как udev держит /dev/bench/thermo-a для настоящей платы (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)

Прошивка для стенда — термостат 196 с платформой для QEMU. От платформы 115в она отличается тремя вещами: после старта печатает строку boot thermo196 v1.4.2, отвечает на ver версией, которую задал конвейер, и в стендовой сборке (-DBENCH_HOOKS) понимает hang — зависает намертво с запрещёнными прерываниями. Так моделируется ошибка, после которой прибор молчит. Разбор строки консоли:

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;
    }
}

Образ собран без предупреждений (-Wall -Wextra -Werror): 1964 байта кода и констант, 8 байт .data, 128 байт .bss, .bin — 1972 байта.

110б.4 Фикстуры и параметры командной строки

Всё, что отличает одну плату от другой, приходит параметрами: --port, --baud, --image, --flasher, --probe (серийный номер программатора), --fw-version (что должно оказаться на плате) и --boot-timeout. Значения по умолчанию для порта, образа и программатора берутся из переменных окружения — их удобно задавать в настройках задания.

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="не сбрасывать плату перед каждым тестом")

Фикстур две, и у них разный срок жизни. bench живёт весь прогон: прошивка — самая долгая операция, её делают один раз. Уборка зарегистрирована до прошивки, поэтому QEMU (или порт) закрывается и тогда, когда плата не стартовала:

@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 нужна каждому тесту и сбрасывает плату перед ним. Это главная защита стенда: зависание или изменённая уставка в одном тесте не переходят в следующий.

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

Файл настроек набора — pytest.ini: отчёт JUnit в формате xunit2, опыт demo из 110б.5 в обычный прогон не входит:

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

110б.5 Тесты термостата и зависшая плата

Шесть тестов, один из них с четырьмя вариантами команды — всего девять прогонов:

"""Раздел 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_version`	`ver`	версия совпадает со строкой старта и с той, что собрал конвейер
`test_get`	`get`	температура 5…35 °C, заводская уставка 21.0, состояние «ждёт» или «нагрев»
`test_set_roundtrip`	`set 22.5`, `get`	`ok`, затем уставка 22.5; в конце уставка возвращается
`test_bad_command`	`foo`, `set 99`, `set`, `hyst 9`	ответ начинается с `?`
`test_long_line`	40 символов `x`	`? строка длиннее 31 символа`
`test_hang_then_reset`	`hang`, `get`	ответа нет за 1 с, после сброса `get` отвечает

Задание стенда в QEMU, вывод pytest:

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 ========================

Время по тестам — из отчёта JUnit:

Тест	Время, с
`test_version` (с прошивкой)	0,072
`test_get`	0,051
`test_set_roundtrip`	0,052
`test_bad_command`, каждый из четырёх	0,051
`test_long_line`	0,053
`test_hang_then_reset`	1,104
всего (pytest)	1,55
всё задание (запуск Python, сбор тестов, отчёт)	2,17

Почти всё время прогона — один тайм-аут: тест зависания ждёт ответ 1 с. Остальные тесты в QEMU идут по 0,05 с, и почти всё это время — пауза в 0,05 с после команды сброса в монитор. На настоящей плате прошивка и сброс через программатор идут дольше; эти времена здесь не мерились.

Зачем сброс перед каждым тестом. Три теста опыта demo: первый посылает hang и не ловит тайм-аут — так выглядит настоящая ошибка прошивки. Два следующих — обычные get и 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+$")

Прогон	Итог	Время
со сбросом перед тестом	1 упал, 2 прошли	1,41 с
`--no-reset`	упали все 3, каждый по тайм-ауту в 1 с	3,28 с

Без сброса одна зависшая плата роняет все тесты после себя, и в отчёте не видно, какой был первым. Со сбросом падает ровно тот тест, на котором плата повисла.

110б.6 Конвейер: свой исполнитель с тегом bench

Задание стенда вызывает один скрипт — его же можно запустить руками на машине со стендом:

#!/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

Отчёт report/junit.xml уходит в сервис даже при упавших тестах — для этого он и нужен. Код выхода — код pytest: упал тест — задание красное. Имя набора в отчёте — по имени порта (hil-thermo-a), чтобы отчёты двух плат не смешались.

Файл конвейера для GitLab (в репозитории — .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

Четыре ключа делают из задания стенд:

Ключ	Зачем
`tags: [bench, thermo-a]`	задание возьмёт только исполнитель, у которого есть оба тега
`resource_group`	второе задание на ту же плату ждёт, пока не закончится первое
`needs: [build]`	образ приходит артефактом из сборки, на машине со стендом ничего не собирается
`artifacts: when: always` + `reports: junit`	отчёт сохраняется и при падении; GitLab показывает тесты в запросе слияния

То же для GitHub Actions (.github/workflows/bench.yml); тег здесь называется меткой, а очередь на плату задаёт 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/

Исполнитель в разделе не регистрировался. В GitLab его создают в настройках проекта (раздел CI/CD → исполнители) с тегами bench и thermo-a, а на машине со стендом регистрируют службу gitlab-runner с исполнителем shell — задание тогда идёт прямо на этой машине. В GitHub метки bench и thermo-a задают при настройке своего исполнителя.

Сервиса под рукой не было, поэтому оба файла проверены своим скриптом ci_lint.py: YAML читается, все ключи — из списков известных, ссылки needs и extends ведут на существующие задания, этапы есть в stages, у каждого задания с тегом bench есть resource_group, отчёт JUnit и when: always. Вывод на этих файлах:

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

Тот же скрипт на испорченных копиях находит опечатку tag: вместо tags: (с ней задание потеряло бы тег bench и могло уйти на любой исполнитель), две платы в одной группе и отчёт без when: always. Это не замена проверке самим сервисом: gitlab-ci-lint и actionlint на этих файлах не запускались.

110б.7 Грабли

Имя порта поменялось — тесты стучатся в чужую плату. Номер в /dev/ttyACM0 — порядок подключения, а не разъём (раздел 115б.2). В опыте две «платы» x и y, запущенные в порядке x, y, получили /dev/pts/0 и /dev/pts/1; в порядке y, x — наоборот. Стенду нужен постоянный адрес: правило udev по серийному номеру программатора, по ссылке на каждую плату.

# Раздел 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"

Правила в разделе не проверялись на машине с udev; серийные номера здесь подставные. В конвейере — только постоянное имя: /dev/bench/thermo-a.

Несколько плат на одной машине. Каждой плате — свой порт, свой серийный номер программатора, свой тег и своя группа. Два задания стенда на двух «платах» в опыте шли параллельно: каждое по 1,548 с в pytest, обе вместе — 2,09 с по часам; одно задание целиком — 2,17 с. Если два задания всё-таки откроют один порт, exclusive=True не даст второму открыть его молча: Could not exclusively lock port.

Порт «завис» после перепрошивки. Порт, открытый до перепрошивки, ведёт в никуда: в опыте запись в старый объект порта дала SerialException: write failed: [Errno 5] Input/output error. У USB CDC самого STM32 то же: прибор при сбросе пропадает с шины и возвращается новым устройством (раздел 115б.6). Поэтому Board.flash закрывает порт до прошивки и открывает после — новый объект по постоянному имени ответил сразу.

Команда до строки старта теряется. Прошивка ещё не включила приём UART, а тест уже пишет. В опыте get, посланная сразу после запуска, ответа не получила — пришли только boot thermo196 v1.4.2 и cfg: заводские. Поэтому flash и reset ждут строку старта и только потом возвращают управление.

Медленная прошивка не успевает к тайм-ауту. Сборка с задержкой старта 1,5 с (прогрев датчика) и --boot-timeout 1: тест не начался, в отчёте — ошибка подготовки, а не падение: failed on setup with "boardlib.BoardTimeout: нет /^boot thermo196 v(\S+)$/ за 1.0 с". С --boot-timeout 3 тест прошёл, но шёл 3,020 с: медленный старт платится дважды — при прошивке и при сбросе перед тестом, а при сбросе перед каждым тестом — на каждом. Тайм-ауты берут с запасом от измеренного времени старта, а долгую подготовку прибора не повторяют без нужды.

Permission denied. Порт принадлежит группе dialout с правами 0660, а служба исполнителя работает от своего пользователя. В опыте порт переведён в root:dialout 0660: пользователь без группы получил [Errno 13] could not open port … Permission denied, после usermod -aG dialout — открыл. Группа действует в новых сеансах: службу исполнителя после этого перезапускают (раздел 194.5). Программатору ST-LINK нужен доступ к самому устройству USB — его обычно дают правила udev из пакета stlink-tools или openocd.

Сброс не вернул настройки. В стендовой платформе уставка лежит в ОЗУ и сброс возвращает заводские 21.0. На BluePill из 196 она во Flash и сброс переживает. Тест, который меняет настройки, возвращает их сам — finally в test_set_roundtrip.

Стендовая сборка ушла в выпуск. hang — команда для стенда, в выпускной сборке её быть не должно. Она включается только -DBENCH_HOOKS; выпуск собирается отдельно, как в разделе 110.

GitHub вытеснил задание из очереди. У concurrency в очереди держится одно ожидающее задание: новое ожидающее отменяет прежнее (по документации GitHub Actions, в разделе не проверялось). Для стенда это значит, что промежуточные коммиты при частых правках на плате не проверятся. В GitLab resource_group держит очередь из всех заданий.

Плата зависла, а сторожевого таймера нет. Сброс через программатор спасает тест, но не прибор в работе. Стенд ловит зависание тайм-аутом; прибору нужен свой сторожевой таймер (раздел 33а).

110б.8 Итоги

Стенд — это исполнитель конвейера с тегом bench, к которому подключены плата, программатор и порт; сборка идёт на любом исполнителе, на стенд приходит готовый образ.
Фикстура board прячет всё про плату: прошить один раз на прогон (bench, на весь прогон), сбросить перед каждым тестом (board), читать строки с тайм-аутом и ждать по регулярному выражению. Порт, скорость, образ и программатор — параметры --port, --baud, --image, --flasher.
Прогон в QEMU: собрано 12 тестов, выбрано 9, 9 прошли за 1,55 с, задание целиком — 2,17 с; из них 1,104 с — тест зависания с тайм-аутом 1 с. Образ — 1964 байта кода, .bin — 1972 байта.
Зависшая плата: со сбросом перед тестом падает 1 тест из 3, без сброса — все 3 (3,28 с вместо 1,41 с).
Медленный старт 1,5 с: тайм-аут 1 с — ошибка подготовки, 3 с — тест прошёл за 3,020 с.
Две платы параллельно — 2,09 с против 2,17 с на одну; на одну плату — одно задание: resource_group в GitLab, concurrency в GitHub, exclusive=True в pyserial.
Грабли: имена портов по порядку подключения (правило udev по серийнику), старый порт после перепрошивки (Errno 5), команда до строки старта, права dialout (Errno 13).

Полные исходники

Всё, что нужно, чтобы повторить стенд: прошивка, библиотека платы, фикстуры, тесты, скрипты задания и файлы конвейера, проверка YAML и опыты раздела. Логика термостата thermo.c и thermo.h — из раздела 196, стартовый код и скрипт компоновщика — из 115в.

plat_bench.c — платформа термостата для стенда в QEMU: строка старта, 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 — плата для pytest: порт, строки с тайм-аутом, прошивальщики.

"""Раздел 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 — параметры командной строки и фикстуры bench и 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 — настройки набора.

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

test_thermo.py — тесты термостата.

"""Раздел 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 — опыт «зависшая плата без сброса».

"""Раздел 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 — стендовая сборка: задание build.

#!/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 — задание стенда.

#!/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 — конвейер GitLab; в репозитории лежит под именем .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 — то же для 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 — правило udev: постоянные имена портов по серийному номеру.

# Раздел 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 — проверка файлов конвейера без сервиса.

"""Раздел 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 — опыты с портом: имена, старый порт, блокировка, 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 — весь прогон раздела в контейнере.

#!/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"

Дальше: раздел 111 «Контейнеры и инфраструктура» — что такое образ и контейнер изнутри, Compose и обратный прокси для служб, которыми обрастает прибор.

Открыть в интерактивном справочнике