550 lines
21 KiB
Python
550 lines
21 KiB
Python
"""
|
||
conftest.py — pytest-фикстуры для HIL-тестов MIMXRT1052.
|
||
Вся работа с железом делегирована pyocd_utils.
|
||
Конфигурация читается из env_config (приоритет: env > .env > default).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Generator
|
||
from contextlib import contextmanager
|
||
import pytest
|
||
import serial
|
||
|
||
import env_config as cfg
|
||
from pyocd_utils import flexram_init, load_elf, open_target, run_from_vectors
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CLI-опции pytest (перекрывают .env и os.environ)
|
||
# ---------------------------------------------------------------------------
|
||
def pytest_addoption(parser: pytest.Parser) -> None:
|
||
parser.addoption("--elf", default=None, help="Путь к .elf файлу")
|
||
parser.addoption("--vcom", default=cfg.VCOM_PORT, help="VCOM-порт MCU-Link")
|
||
parser.addoption("--m5-port", default=None, help="M5StampPLC serial port")
|
||
parser.addoption("--no-load", action="store_true", help="ELF уже запущен")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Внутренняя функция загрузки
|
||
# ---------------------------------------------------------------------------
|
||
def _load_elf(request: pytest.FixtureRequest, default_elf: Path) -> None:
|
||
if request.config.getoption("--no-load"):
|
||
log.info("--no-load: пропускаем загрузку ELF")
|
||
return
|
||
|
||
elf = Path(request.config.getoption("--elf") or default_elf)
|
||
assert elf.exists(), f"ELF не найден: {elf}"
|
||
|
||
with open_target(frequency=cfg.PYOCD_FREQUENCY) as target:
|
||
flexram_init(target)
|
||
load_elf(target, str(elf))
|
||
run_from_vectors(target)
|
||
|
||
log.info("ELF загружен: %s", elf.name)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Общая логика UART: открыть порт, дождаться READY
|
||
# ---------------------------------------------------------------------------
|
||
@contextmanager
|
||
def _uart_context(port: str, baud: int, ready_timeout: float):
|
||
ser = serial.Serial(port=port, baudrate=baud, timeout=2.0, write_timeout=1.0)
|
||
try:
|
||
deadline = time.monotonic() + ready_timeout
|
||
ready = False
|
||
while time.monotonic() < deadline:
|
||
line = ser.readline().decode("ascii", errors="replace").strip()
|
||
if line == "READY":
|
||
ready = True
|
||
log.info("Получен READY от прошивки")
|
||
break
|
||
|
||
if not ready:
|
||
pytest.fail(f"Прошивка не отправила READY за {ready_timeout} с")
|
||
|
||
ser.reset_input_buffer()
|
||
yield ser
|
||
finally:
|
||
ser.close() # гарантированно, всегда
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# M5StampPLC — драйвер для pytest (JSON-lines протокол)
|
||
#
|
||
# Определяем РАНЬШЕ фикстур, которые его используют в аннотациях.
|
||
# ---------------------------------------------------------------------------
|
||
class M5Agent:
|
||
"""Драйвер M5StampPLC для pytest (JSON-lines протокол через USB CDC)."""
|
||
|
||
def __init__(self, ser: serial.Serial) -> None:
|
||
self._ser = ser
|
||
|
||
def cmd(self, command: str, **kwargs) -> dict:
|
||
"""Отправить JSON-команду, вернуть parsed-ответ. RuntimeError при ok=false."""
|
||
payload = {"cmd": command, **kwargs}
|
||
self._ser.reset_input_buffer()
|
||
self._ser.write((json.dumps(payload) + "\r\n").encode("ascii"))
|
||
|
||
deadline = time.monotonic() + cfg.M5_TIMEOUT
|
||
while time.monotonic() < deadline:
|
||
raw = self._ser.readline()
|
||
if not raw:
|
||
continue
|
||
text = raw.decode("ascii", errors="replace").strip()
|
||
if not text or text == "READY" or not text.startswith("{"):
|
||
continue
|
||
try:
|
||
resp = json.loads(text)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if not resp.get("ok"):
|
||
raise RuntimeError(f"M5 error: {resp.get('err', '?')} (cmd={command})")
|
||
return resp
|
||
|
||
raise TimeoutError(f"M5: нет ответа на '{command}'")
|
||
|
||
def ping(self) -> None:
|
||
self.cmd("ping")
|
||
|
||
def power(self, state: bool) -> None:
|
||
"""Включить/выключить питание таргета (RLY1)."""
|
||
self.cmd("power", state=state)
|
||
|
||
def opto_set(self, ch: int, state: bool) -> None:
|
||
self.cmd("opto_set", ch=ch, state=state)
|
||
|
||
def opto_all_off(self) -> None:
|
||
self.cmd("opto_all_off")
|
||
|
||
def info(self) -> dict:
|
||
"""Запросить информацию об агенте (включая can_ok)."""
|
||
return self.cmd("info")
|
||
|
||
def can_send(self, can_id: int, data: list, ext: bool = False) -> None:
|
||
"""Отправить CAN-фрейм с шины M5."""
|
||
self.cmd("can_send", id=can_id, data=list(data), ext=ext)
|
||
|
||
def can_recv(self, timeout_ms: int = 500) -> dict:
|
||
"""
|
||
Принять CAN-фрейм на M5.
|
||
Возвращает dict {id, ext, data}.
|
||
Выбрасывает TimeoutError если фрейм не пришёл.
|
||
"""
|
||
resp = self.cmd("can_recv", timeout_ms=timeout_ms)
|
||
# cmd() уже выбрасывает RuntimeError при ok=false (включая timeout от агента)
|
||
return resp
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Фикстура M5StampPLC
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture(scope="module")
|
||
def m5(request: pytest.FixtureRequest) -> Generator[M5Agent, None, None]:
|
||
port = request.config.getoption("--m5-port") or cfg.M5_PORT
|
||
log.info("M5 %s @ %d baud", port, cfg.M5_BAUD)
|
||
|
||
ser = serial.Serial(
|
||
port=port,
|
||
baudrate=cfg.M5_BAUD,
|
||
timeout=cfg.M5_TIMEOUT,
|
||
write_timeout=1.0,
|
||
)
|
||
|
||
# Ждём READY; fallback на ping если агент уже работает
|
||
deadline = time.monotonic() + cfg.READY_TIMEOUT
|
||
ready = False
|
||
while time.monotonic() < deadline:
|
||
line = ser.readline().decode("ascii", errors="replace").strip()
|
||
if line == "READY":
|
||
ready = True
|
||
log.info("M5 READY получен")
|
||
break
|
||
|
||
agent = M5Agent(ser)
|
||
if not ready:
|
||
try:
|
||
agent.ping()
|
||
log.info("M5 уже работает (READY пропущен, ping OK)")
|
||
except Exception:
|
||
ser.close()
|
||
pytest.fail(
|
||
"M5 agent не отвечает — проверьте HIL_M5_PORT и "
|
||
"что m5/agent.py развёрнут (just host::m5-deploy)."
|
||
)
|
||
|
||
# Включаем питание таргета и ждём стабилизации
|
||
agent.power(True)
|
||
log.info("Питание таргета включено, ждём %.1f с", cfg.POWER_ON_SETTLE_S)
|
||
time.sleep(cfg.POWER_ON_SETTLE_S)
|
||
|
||
agent.opto_all_off() # безопасное начальное состояние
|
||
yield agent
|
||
|
||
# Teardown: выключить всё
|
||
try:
|
||
agent.opto_all_off()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
agent.power(False)
|
||
log.info("Питание таргета выключено")
|
||
except Exception:
|
||
pass
|
||
ser.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Фикстуры загрузки (scope=module — один раз на файл с тестами)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def loaded_host_uart(request: pytest.FixtureRequest, m5: M5Agent) -> None:
|
||
"""
|
||
Загрузить test_hil_button.elf на таргет по SWD
|
||
|
||
Явная зависимость от m5 гарантирует порядок:
|
||
1. m5 создаётся первым → питание таргета включено
|
||
2. только потом pyOCD подключается и грузит ELF
|
||
"""
|
||
_load_elf(
|
||
request,
|
||
Path(cfg.BUILD_DIR) / "tests/target/host_uart/test_host_uart.elf",
|
||
)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def loaded_hil_button(request: pytest.FixtureRequest, m5: M5Agent) -> None:
|
||
"""
|
||
Загрузить test_hil_button.elf на таргет по SWD
|
||
|
||
Явная зависимость от m5 гарантирует порядок:
|
||
1. m5 создаётся первым → питание таргета включено
|
||
2. только потом pyOCD подключается и грузит ELF
|
||
"""
|
||
_load_elf(
|
||
request,
|
||
Path(cfg.BUILD_DIR) / "tests/target/hil_button/test_hil_button.elf",
|
||
)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def loaded_hil_opto(request: pytest.FixtureRequest, m5: M5Agent) -> None:
|
||
"""
|
||
Загрузить test_hil_opto.elf на таргет по SWD
|
||
|
||
Явная зависимость от фикстуры m5 гарантирует порядок:
|
||
1. m5 создаётся первым → питание таргета включено
|
||
2. только потом pyOCD подключается и грузит ELF
|
||
Без этой зависимости pytest мог бы попытаться подключиться
|
||
к MCU пока он ещё обесточен.
|
||
"""
|
||
_load_elf(
|
||
request,
|
||
Path(cfg.BUILD_DIR) / "tests/target/hil_opto/test_hil_opto.elf",
|
||
)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def loaded_hil_can(request: pytest.FixtureRequest, m5: M5Agent) -> None:
|
||
"""
|
||
Загрузить test_hil_can.elf на таргет по SWD
|
||
|
||
Явная зависимость от m5 гарантирует порядок:
|
||
1. m5 создаётся первым → питание таргета включено
|
||
2. только потом pyOCD подключается и грузит ELF
|
||
"""
|
||
_load_elf(
|
||
request,
|
||
Path(cfg.BUILD_DIR) / "tests/target/hil_can/test_hil_can.elf",
|
||
)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def loaded_hil_usb_cdc(request: pytest.FixtureRequest, m5: M5Agent) -> None:
|
||
"""
|
||
Загрузить test_hil_usb_cdc.elf на таргет по SWD
|
||
|
||
Явная зависимость от m5 гарантирует порядок:
|
||
1. m5 создаётся первым → питание таргета включено
|
||
2. только потом pyOCD подключается и грузит ELF
|
||
"""
|
||
_load_elf(
|
||
request,
|
||
Path(cfg.BUILD_DIR) / "tests/target/hil_usb_cdc/test_hil_usb_cdc.elf",
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Фикстуры UART (установление соединения с VCOM программатора NXP MCU-Link)
|
||
# Выполняется один раз на тест
|
||
# ---------------------------------------------------------------------------
|
||
_UART_FIXTURE_MAP = {
|
||
"uart": "loaded_host_uart",
|
||
"uart_opto": "loaded_hil_opto",
|
||
"uart_can": "loaded_hil_can",
|
||
"uart_button": "loaded_hil_button",
|
||
"uart_hil_usb_cdc": "loaded_hil_usb_cdc",
|
||
}
|
||
|
||
|
||
def _make_uart_fixture(loaded_name: str):
|
||
@pytest.fixture(scope="module")
|
||
def _fixture(
|
||
request: pytest.FixtureRequest,
|
||
) -> Generator[serial.Serial, None, None]:
|
||
request.getfixturevalue(loaded_name) # триггерит зависимость явно
|
||
port = request.config.getoption("--vcom")
|
||
with _uart_context(port, cfg.VCOM_BAUD, cfg.READY_TIMEOUT) as ser:
|
||
yield ser
|
||
|
||
return _fixture
|
||
|
||
|
||
# Регистрируем все фикстуры в пространстве имён модуля одной строкой
|
||
for _name, _dep in _UART_FIXTURE_MAP.items():
|
||
globals()[_name] = _make_uart_fixture(_dep)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Фикстуры USB CDC (открытие порта, установка DTR)
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture(scope="module")
|
||
def usb_cdc_port(
|
||
uart_hil_usb_cdc,
|
||
) -> Generator[serial.Serial, None, None]:
|
||
"""Открыть USB CDC порт таргета. Ждёт появления порта и DTR ready."""
|
||
import time
|
||
|
||
if not cfg.TARGET_VCOM_PORT:
|
||
pytest.skip("HIL_USB_CDC_PORT not set")
|
||
|
||
# Ждём появления USB CDC порта (enumeration после загрузки ELF).
|
||
deadline = time.monotonic() + cfg.TARGET_VCOM_TIMEOUT
|
||
ser = None
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
ser = serial.Serial(
|
||
port=cfg.TARGET_VCOM_PORT, baudrate=cfg.TARGET_VCOM_BAUD, timeout=0.5
|
||
)
|
||
break
|
||
except serial.SerialException:
|
||
time.sleep(0.3)
|
||
|
||
if ser is None:
|
||
pytest.fail(
|
||
f"USB CDC port {cfg.TARGET_VCOM_PORT} not available after {cfg.TARGET_VCOM_TIMEOUT}s"
|
||
)
|
||
|
||
try:
|
||
# Установить DTR чтобы firmware увидела DTE presence.
|
||
ser.dtr = True
|
||
time.sleep(0.3)
|
||
# Сбросить входной буфер — могут быть мусорные байты от enumeration.
|
||
ser.reset_input_buffer()
|
||
yield ser
|
||
finally:
|
||
ser.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Утилита для тестов
|
||
# ---------------------------------------------------------------------------
|
||
def uart_cmd(ser: serial.Serial, cmd: str) -> str:
|
||
"""Отправить команду, получить одну строку ответа."""
|
||
ser.reset_input_buffer()
|
||
ser.write((cmd.strip() + "\r\n").encode("ascii"))
|
||
resp = ser.readline()
|
||
if not resp:
|
||
raise TimeoutError(f"Нет ответа на команду '{cmd}'")
|
||
return resp.decode("ascii", errors="replace").strip()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# firmware_test CDC — фикстура для тестирования firmware_test через протокол v2
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class FirmwareCdc:
|
||
"""
|
||
Драйвер firmware_test протокола v2 (JSON-lines через USB CDC).
|
||
|
||
Открывает CDC-порт, ждёт session_start, предоставляет методы:
|
||
- run_test(test_id) → запустить тест, дождаться test_result
|
||
- send_confirm(confirm_id, confirmed) → ответить на confirm_request
|
||
- wait_event(type, timeout_s) → ждать событие нужного типа
|
||
"""
|
||
|
||
def __init__(self, ser: serial.Serial) -> None:
|
||
self._ser = ser
|
||
|
||
def _readline(self, timeout_s: float = 5.0) -> dict:
|
||
"""Читать строки до получения валидного JSON. TimeoutError если истёк таймаут."""
|
||
deadline = time.monotonic() + timeout_s
|
||
while time.monotonic() < deadline:
|
||
raw = self._ser.readline()
|
||
if not raw:
|
||
continue
|
||
text = raw.decode("ascii", errors="replace").strip()
|
||
if not text:
|
||
continue
|
||
try:
|
||
return json.loads(text)
|
||
except json.JSONDecodeError:
|
||
log.debug("firmware_cdc: не JSON: %r", text)
|
||
continue
|
||
raise TimeoutError(f"firmware_cdc: нет ответа за {timeout_s} с")
|
||
|
||
def wait_event(self, event_type: str, timeout_s: float = 5.0) -> dict:
|
||
"""
|
||
Ждать событие с заданным полем type.
|
||
Пропускает промежуточные события (progress и т.п.).
|
||
"""
|
||
deadline = time.monotonic() + timeout_s
|
||
while time.monotonic() < deadline:
|
||
remaining = deadline - time.monotonic()
|
||
msg = self._readline(timeout_s=max(remaining, 0.1))
|
||
if msg.get("type") == event_type:
|
||
return msg
|
||
log.debug(
|
||
"firmware_cdc: пропускаем %r (ждём %r)", msg.get("type"), event_type
|
||
)
|
||
raise TimeoutError(
|
||
f"firmware_cdc: событие {event_type!r} не получено за {timeout_s} с"
|
||
)
|
||
|
||
def send(self, payload: dict) -> None:
|
||
"""Отправить JSON-команду на таргет."""
|
||
line = json.dumps(payload, separators=(",", ":")) + "\n"
|
||
self._ser.write(line.encode("ascii"))
|
||
self._ser.flush()
|
||
|
||
def ping(self) -> None:
|
||
"""Проверка связи: ping → pong."""
|
||
self.send({"type": "cmd", "cmd": "ping"})
|
||
self.wait_event("pong")
|
||
|
||
def send_confirm(self, confirm_id: str, confirmed: bool) -> None:
|
||
"""Ответить на confirm_request."""
|
||
self.send({"type": "confirm", "id": confirm_id, "confirmed": confirmed})
|
||
|
||
def run_test(self, test_id: str, timeout_s: float = 60.0) -> dict:
|
||
"""
|
||
Запустить тест по id, вернуть dict test_result.
|
||
|
||
Промежуточные confirm_request игнорируются — их нужно обрабатывать
|
||
отдельно через wait_event("confirm_request") до вызова run_test,
|
||
или через on_confirm callback в специализированных тестах.
|
||
|
||
Для HIL-тестов (opto, can) используй run_hil_test().
|
||
"""
|
||
self.send({"type": "cmd", "cmd": "run", "id": test_id})
|
||
self.wait_event("test_begin", timeout_s=5.0)
|
||
return self.wait_event("test_result", timeout_s=timeout_s)
|
||
|
||
def run_hil_test(
|
||
self,
|
||
test_id: str,
|
||
on_confirm,
|
||
timeout_s: float = 30.0,
|
||
) -> dict:
|
||
"""
|
||
Запустить HIL-тест с обработкой confirm_request.
|
||
|
||
on_confirm(confirm_id: str) → bool вызывается для каждого confirm_request.
|
||
Возвращает True → отправляет confirmed:true, False → confirmed:false.
|
||
|
||
Цикл завершается при получении test_result.
|
||
|
||
Args:
|
||
test_id: идентификатор теста в реестре таргета
|
||
on_confirm: callable(confirm_id: str) → bool
|
||
timeout_s: общий таймаут на весь тест
|
||
"""
|
||
self.send({"type": "cmd", "cmd": "run", "id": test_id})
|
||
self.wait_event("test_begin", timeout_s=5.0)
|
||
|
||
deadline = time.monotonic() + timeout_s
|
||
while time.monotonic() < deadline:
|
||
remaining = deadline - time.monotonic()
|
||
msg = self._readline(timeout_s=max(remaining, 0.1))
|
||
msg_type = msg.get("type")
|
||
|
||
if msg_type == "test_result":
|
||
return msg
|
||
|
||
if msg_type == "confirm_request":
|
||
confirm_id = msg.get("id", "")
|
||
log.info("firmware_cdc: confirm_request %r", confirm_id)
|
||
confirmed = on_confirm(confirm_id)
|
||
self.send_confirm(confirm_id, confirmed)
|
||
continue
|
||
|
||
log.debug("firmware_cdc: промежуточное событие %r", msg_type)
|
||
|
||
raise TimeoutError(
|
||
f"firmware_cdc: test_result для {test_id!r} не получен за {timeout_s} с"
|
||
)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def firmware_cdc(m5: M5Agent) -> Generator[FirmwareCdc, None, None]:
|
||
"""
|
||
Открыть USB CDC порт firmware_test, дождаться session_start.
|
||
|
||
Зависит от m5 — питание таргета уже включено.
|
||
firmware_test должна быть прошита в Flash (via SDP) и запущена.
|
||
|
||
Порт берётся из TARGET_VCOM_PORT (HIL_USB_CDC_PORT в .env).
|
||
"""
|
||
if not cfg.TARGET_VCOM_PORT:
|
||
pytest.skip("HIL_USB_CDC_PORT не задан — пропускаем firmware_cdc тесты")
|
||
|
||
# Ждём появления CDC порта после включения питания
|
||
deadline = time.monotonic() + cfg.TARGET_VCOM_TIMEOUT
|
||
ser = None
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
ser = serial.Serial(
|
||
port=cfg.TARGET_VCOM_PORT,
|
||
baudrate=cfg.TARGET_VCOM_BAUD,
|
||
timeout=1.0,
|
||
write_timeout=1.0,
|
||
)
|
||
break
|
||
except serial.SerialException:
|
||
time.sleep(0.2)
|
||
|
||
if ser is None:
|
||
pytest.fail(
|
||
f"CDC порт {cfg.TARGET_VCOM_PORT} недоступен за {cfg.TARGET_VCOM_TIMEOUT} с"
|
||
)
|
||
|
||
try:
|
||
ser.dtr = True
|
||
time.sleep(0.3)
|
||
ser.reset_input_buffer()
|
||
|
||
cdc = FirmwareCdc(ser)
|
||
|
||
# session_start — одноразовое событие при старте, может быть уже пропущено.
|
||
# Проверяем живость через ping → pong.
|
||
try:
|
||
cdc.ping()
|
||
log.info("firmware_test: ping OK, прошивка активна")
|
||
except (TimeoutError, Exception) as exc:
|
||
pytest.fail(
|
||
f"firmware_test не отвечает на ping — "
|
||
f"прошивка запущена? Порт верный? ({exc})"
|
||
)
|
||
|
||
yield cdc
|
||
finally:
|
||
ser.close()
|