# Phase 1 - MacOS tested

This commit is contained in:
Dmitry Akimov 2026-07-02 18:03:14 +03:00
parent da2681245b
commit b55ec8532c
7 changed files with 976 additions and 0 deletions

Binary file not shown.

View file

@ -0,0 +1,378 @@
"""
flash_backend.py синхронное ядро прошивки MIMXRT1052 на spsdk Python API.
Прямой порт tools/host/flash_usb.py (subprocess sdphost/blhost) на прямые
вызовы spsdk (SDP/McuBoot/HabImage), провалидированные в Фазе 0
(spike_hab.py, spike_flash.py байт-в-байт и на живом железе, macOS+Windows).
Zero Textual/asyncio импортов модуль полностью синхронный и тестируемый
без event loop (см. test_flash_backend.py). Async-обвязка (asyncio.to_thread
+ run_coroutine_threadsafe) забота Flasher (Фаза 2), не этого модуля.
Открытые вопросы/допущения (см. сопроводительное сообщение в чате):
- Одна сессия McuBoot на весь flash()/erase_chip(), а не переоткрытие
на каждую операцию, как в CLI flash_usb.py.
- Имена FlashProgress.phase свои (не парсинг stdout blhost).
- Иерархия исключений предложение автора модуля, не зафиксирована
отдельно в MONOLITH_APP_PLAN.md.
"""
from __future__ import annotations
import hashlib
import logging
import os
import tempfile
import time
from pathlib import Path
from typing import Callable, Optional
from spsdk.image.hab.hab_image import HabImage
from spsdk.mboot import McuBoot, MbootUSBInterface
from spsdk.sdp import SDP, SdpUSBInterface
from spsdk.utils.config import Config
from .models import FlashProgress
from .usb_ports import UsbId, resolve_serial_port
logger = logging.getLogger(__name__)
ProgressCallback = Callable[[FlashProgress], None]
# ─── Пути ────────────────────────────────────────────────────────────────
# tools/production/app/flash_backend.py → корень репозитория
REPO_ROOT = Path(__file__).resolve().parents[3]
_HOST_DCD_DIR = REPO_ROOT / "tools" / "host" / "dcd"
FLASHLOADER_BIN = _HOST_DCD_DIR / "ivt_flashloader.bin"
REAL_DCD_BIN = _HOST_DCD_DIR / "dcd.bin"
def fcb_blob_path(fcb_filename: str) -> Path:
"""Путь к готовому FCB-блобу (tools/host/dcd/w25q128_fdcb.bin и т.п.)."""
return _HOST_DCD_DIR / fcb_filename
# ─── USB VID:PID (те же переменные окружения, что уже приняты в проекте) ──
_BOOTROM_VID = int(os.environ.get("BOOTROM_VID", "0x1fc9"), 16)
_BOOTROM_PID = int(os.environ.get("BOOTROM_PID", "0x0130"), 16)
_FLASHLOADER_VID = int(os.environ.get("FLASHLOADER_VID", "0x15a2"), 16)
_FLASHLOADER_PID = int(os.environ.get("FLASHLOADER_PID", "0x0073"), 16)
_CDC_VID = int(os.environ.get("SERVICE_CDC_VID", "0x1996"), 16)
_CDC_PID = int(os.environ.get("SERVICE_CDC_PID", "0x00ad"), 16)
_SDP_DEVICE_ID = f"0x{_BOOTROM_VID:04x}:0x{_BOOTROM_PID:04x}"
_FLASHLOADER_DEVICE_ID = f"0x{_FLASHLOADER_VID:04x}:0x{_FLASHLOADER_PID:04x}"
# ─── Аппаратные константы (см. flash_usb.py — значения не менялись) ──────
FLASH_BASE = 0x60000000
HAB_OFFSET = 0x1000
FLASHLOADER_LOAD_ADDR = 0x20001C00
FLEXSPI_OPTION_ADDR = 0x2000
FLEXSPI_OPTION_VALUE = 0xC0000007
FLEXSPI_FCB_VALUE = 0xF000000F # tag=0xF → Write FCB command (auto-config)
FLEXSPI_MEMORY_ID = 9
ERASE_ALL_TIMEOUT_MS = 200_000 # эквивалент blhost -t 200000 (см. Фазу 0, ⚠В2)
FLASHLOADER_WAIT_TIMEOUT_S = 10.0
_HAB_OPTIONS_TEMPLATE = [
"options:",
" flags: 0x00",
" startAddress: 0x60000000",
" ivtOffset: 0x1000",
" initialLoadSize: 0x2000",
" family: mimxrt1050",
]
# ─── Исключения ────────────────────────────────────────────────────────────
class FlashBackendError(Exception):
"""Базовая ошибка flash_backend."""
class DeviceNotFoundError(FlashBackendError):
"""SDP-устройство не найдено при попытке загрузить Flashloader."""
class FlashLoaderTimeoutError(FlashBackendError):
"""Flashloader не ответил за FLASHLOADER_WAIT_TIMEOUT_S после jump_and_run."""
class HabBuildError(FlashBackendError):
"""Ошибка сборки HAB-образа через HabImage (см. build_custom_hab)."""
# ─── Detection (Р7 — spsdk API вместо pyusb) ───────────────────────────────
def detect_sdp() -> bool:
"""True если виден BootROM SDP (см. _SDP_DEVICE_ID)."""
return len(SdpUSBInterface.scan(device_id=_SDP_DEVICE_ID)) > 0
def detect_cdc() -> bool:
"""True если виден CDC firmware_test (список портов, см. usb_ports.py)."""
return resolve_serial_port(UsbId(_CDC_VID, _CDC_PID)) is not None
def _emit(
progress_cb: Optional[ProgressCallback], phase: str, percent: int, message: str
) -> None:
if progress_cb is not None:
progress_cb(FlashProgress(phase=phase, percent=percent, message=message))
# ─── Flashloader bring-up ───────────────────────────────────────────────────
def wait_for_flashloader(
timeout_s: float = FLASHLOADER_WAIT_TIMEOUT_S,
) -> MbootUSBInterface:
"""Опрашивает MbootUSBInterface.scan() пока Flashloader не поднимется.
:raises FlashLoaderTimeoutError: если не ответил за timeout_s.
"""
deadline = time.monotonic() + timeout_s
while time.monotonic() < deadline:
found = MbootUSBInterface.scan(device_id=_FLASHLOADER_DEVICE_ID)
if found:
return found[0]
time.sleep(0.5)
raise FlashLoaderTimeoutError(
f"Flashloader не ответил за {timeout_s:.0f}с. Проверь BOOT_MOD пины и подключение."
)
def load_flashloader(
progress_cb: Optional[ProgressCallback] = None,
) -> MbootUSBInterface:
"""Загружает Flashloader через SDP, если ещё не запущен.
:raises DeviceNotFoundError: SDP-устройство не найдено (и Flashloader тоже не поднят).
:raises FlashLoaderTimeoutError: см. wait_for_flashloader().
"""
already = MbootUSBInterface.scan(device_id=_FLASHLOADER_DEVICE_ID)
if already:
logger.info("Flashloader уже запущен — пропускаем загрузку")
return already[0]
sdp_devices = SdpUSBInterface.scan(device_id=_SDP_DEVICE_ID)
if not sdp_devices:
raise DeviceNotFoundError(
f"SDP-устройство не найдено ({_SDP_DEVICE_ID}). Плата в BootROM-режиме?"
)
if not FLASHLOADER_BIN.exists():
raise FlashBackendError(f"Не найден: {FLASHLOADER_BIN}")
_emit(
progress_cb,
"load_flashloader",
0,
f"Загрузка Flashloader через SDP ({_SDP_DEVICE_ID})",
)
data = FLASHLOADER_BIN.read_bytes()
with SDP(sdp_devices[0]) as sdp:
sdp.write_file(FLASHLOADER_LOAD_ADDR, data)
sdp.jump_and_run(FLASHLOADER_LOAD_ADDR)
iface = wait_for_flashloader()
_emit(progress_cb, "load_flashloader", 100, "Flashloader готов")
return iface
# ─── FlexSPI / FCB ───────────────────────────────────────────────────────
def configure_flexspi(mboot: McuBoot) -> None:
"""Инициализирует FlexSPI NOR контроллер (см. flash_usb.py::configure_flexspi)."""
mboot.fill_memory(FLEXSPI_OPTION_ADDR, 4, FLEXSPI_OPTION_VALUE)
ok = mboot.configure_memory(FLEXSPI_OPTION_ADDR, FLEXSPI_MEMORY_ID)
if not ok:
raise FlashBackendError("configure_memory (FlexSPI init) вернул False")
def write_fcb_auto(mboot: McuBoot) -> None:
"""Auto-config FCB через magic option word (см. flash_usb.py::write_fcb).
Надёжно проверен только для W25Q128 см. docstring оригинала.
"""
mboot.fill_memory(FLEXSPI_OPTION_ADDR, 4, FLEXSPI_FCB_VALUE)
ok = mboot.configure_memory(FLEXSPI_OPTION_ADDR, FLEXSPI_MEMORY_ID)
if not ok:
raise FlashBackendError("configure_memory (FCB write) вернул False")
def write_fcb_explicit(mboot: McuBoot, fcb_path: Path) -> None:
"""Пишет буквальный FCB-блоб (512 байт) в Flash[FLASH_BASE] (custom-бинари).
См. flash_usb.py::write_fcb_explicit nxpimage не кладёт FCB в HAB-образ,
поэтому для произвольных чипов нужен явный блоб под конкретный memory chip.
"""
if not fcb_path.exists():
raise FlashBackendError(f"FCB-файл не найден: {fcb_path}")
data = fcb_path.read_bytes()
ok = mboot.write_memory(FLASH_BASE, data, mem_id=0)
if not ok:
raise FlashBackendError(f"write_memory(FCB {fcb_path.name}) вернул False")
# ─── Прошивка / RAM-load / erase ────────────────────────────────────────────
def flash(
hab_bin: Path,
*,
ram_only: bool = False,
fcb_path: Optional[Path] = None,
progress_cb: Optional[ProgressCallback] = None,
) -> None:
"""Прошить HAB-образ в Flash либо загрузить в RAM (см. flash_usb.py::flash).
:param hab_bin: Готовый HAB-образ (.bin). Для custom-бинарей см.
build_custom_hab().
:param ram_only: Загрузить в RAM через SDP, во Flash не писать.
:param fcb_path: Явный FCB-блоб (custom-бинари). None auto-config
(write_fcb_auto, только для штатных firmware_test/production).
:raises FlashBackendError: и подклассы на любой ошибке.
"""
if not hab_bin.exists():
raise FlashBackendError(f"Файл не найден: {hab_bin}")
if ram_only:
_emit(progress_cb, "load_flashloader", 0, f"Загрузка в RAM: {hab_bin.name}")
sdp_devices = SdpUSBInterface.scan(device_id=_SDP_DEVICE_ID)
if not sdp_devices:
raise DeviceNotFoundError(f"SDP-устройство не найдено ({_SDP_DEVICE_ID})")
addr = FLASH_BASE + HAB_OFFSET
with SDP(sdp_devices[0]) as sdp:
sdp.write_file(addr, hab_bin.read_bytes())
sdp.jump_and_run(addr)
_emit(progress_cb, "done", 100, "Загружено в RAM")
return
iface = load_flashloader(progress_cb)
write_addr = FLASH_BASE + HAB_OFFSET
hab_size = hab_bin.stat().st_size
erase_size = ((HAB_OFFSET + hab_size + 0xFFF) // 0x1000) * 0x1000
with McuBoot(iface) as mboot:
_emit(progress_cb, "configure", 0, "Конфигурация FlexSPI NOR")
configure_flexspi(mboot)
_emit(
progress_cb, "erase", 0, f"Стирание 0x{FLASH_BASE:08X} + {erase_size} байт"
)
ok = mboot.flash_erase_region(FLASH_BASE, erase_size, mem_id=0)
if not ok:
raise FlashBackendError("flash_erase_region вернул False")
_emit(progress_cb, "fcb", 0, "Запись FCB")
if fcb_path is not None:
write_fcb_explicit(mboot, fcb_path)
else:
write_fcb_auto(mboot)
_emit(progress_cb, "write", 0, f"Запись {hab_bin.name} ({hab_size} байт)")
def _on_progress(current: int, total: int) -> None:
percent = int(current * 100 / total) if total else 0
_emit(progress_cb, "write", percent, f"{current}/{total} байт")
data = hab_bin.read_bytes()
ok = mboot.write_memory(
write_addr, data, mem_id=0, progress_callback=_on_progress
)
if not ok:
raise FlashBackendError("write_memory (HAB-образ) вернул False")
_emit(progress_cb, "reset", 0, "Reset")
mboot.reset(reopen=False)
_emit(progress_cb, "done", 100, "Прошивка завершена успешно")
def erase_chip(progress_cb: Optional[ProgressCallback] = None) -> None:
"""Полная очистка Flash (см. flash_usb.py::erase_chip). После erase FCB
тоже стёрт плата не загрузится до следующей прошивки."""
iface = load_flashloader(progress_cb)
with McuBoot(iface) as mboot:
_emit(progress_cb, "configure", 0, "Конфигурация FlexSPI NOR")
configure_flexspi(mboot)
_emit(progress_cb, "erase", 0, "Полная очистка Flash (~30с)")
iface.device.timeout = ERASE_ALL_TIMEOUT_MS # см. Фазу 0, ⚠В2
ok = mboot.flash_erase_all(mem_id=FLEXSPI_MEMORY_ID)
if not ok:
raise FlashBackendError("flash_erase_all вернул False")
_emit(progress_cb, "reset", 0, "Reset")
mboot.reset(reopen=False)
_emit(progress_cb, "done", 100, "Chip erase завершён")
# ─── HAB-сборка для custom-бинарей (Фаза 0, ⚠В1 — HabImage вместо nxpimage) ─
def _make_hab_config(input_bin: Path, dcd_bin: Optional[Path], work_dir: Path) -> Path:
"""YAML-конфиг nxpimage hab с АБСОЛЮТНЫМИ путями (Р6 — временный файл
может лежать где угодно, не обязательно рядом с реальными hab/*.yaml)."""
lines = list(_HAB_OPTIONS_TEMPLATE)
if dcd_bin is not None:
lines.append(f" DCDFilePath: {dcd_bin.resolve().as_posix()}")
lines.append(f'inputImageFile: "{input_bin.resolve().as_posix()}"')
lines.append("sections: []")
yaml_path = (
work_dir / f"hab_{hashlib.sha1(input_bin.name.encode()).hexdigest()[:8]}.yaml"
)
yaml_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return yaml_path
def build_custom_hab(
raw_bin: Path,
use_dcd: bool,
progress_cb: Optional[ProgressCallback] = None,
) -> Path:
"""Собрать HAB-образ из сырого бинарника через HabImage (spsdk API).
Заменяет flasher.py::_build_custom_hab (subprocess nxpimage). Логика
сборки байт-в-байт идентична nxpimage CLI см. Фазу 0 (golden-тест).
:return: путь к собранному *.hab.bin во временной директории
(tempfile.gettempdir(), Р6). Вызывающий код отвечает за unlink
после использования так же, как сейчас делает Flasher.
:raises HabBuildError: при ошибке валидации/сборки.
"""
_emit(progress_cb, "hab_build", 0, "Сборка HAB-образа (HabImage)")
dcd_bin = REAL_DCD_BIN if use_dcd else None
if use_dcd and not dcd_bin.exists():
raise HabBuildError(f"DCD запрошен, но файл не найден: {dcd_bin}")
work_dir = Path(tempfile.mkdtemp(prefix="tui_hab_"))
try:
yaml_path = _make_hab_config(raw_bin, dcd_bin, work_dir)
cfg = Config.create_from_file(str(yaml_path))
schemas = HabImage.get_validation_schemas_from_cfg(cfg)
cfg.check(schemas, check_unknown_props=True)
hab = HabImage.load_from_config(cfg)
hab.post_export(cfg.config_dir)
data = hab.export()
except HabBuildError:
raise
except Exception as exc: # noqa: BLE001 — оборачиваем весь spsdk-зоопарк исключений
raise HabBuildError(f"Сборка HAB не удалась: {exc}") from exc
out_path = work_dir / f"{raw_bin.stem}_hab.bin"
out_path.write_bytes(data)
_emit(progress_cb, "hab_build", 100, f"HAB-образ собран: {len(data)} байт")
return out_path

View file

@ -0,0 +1,85 @@
"""
usb_ports.py резолвер serial-портов по VID:PID (Р8).
Принцип (см. MONOLITH_APP_PLAN.md §Р8): имя порта не переносимо даже в
пределах одной ОС (перевтыкание в другой физический разъём меняет имя:
cu.usbmodemXXXX / COMn / ttyACMn), поэтому единственный стабильный
идентификатор устройства VID:PID. Порядок резолва:
1. Явный override через переменную окружения (escape hatch).
2. Иначе поиск по VID:PID среди serial.tools.list_ports.comports().
3. Одно совпадение его имя порта.
4. Ноль совпадений None (штатное состояние: устройство не подключено,
для TUI это WaitingScreen, а не ошибка).
5. Несколько совпадений первое + warning в лог (дизамбигуация по
serial_number задел на будущее, см. MONOLITH_APP_PLAN.md §О3).
Модуль намеренно не хранит собственных VID/PID-констант они остаются
там же, где были (firmware_client.py, m5_client.py, flash_backend.py),
каждый со своим fallback-дефолтом, как уже принято в проекте.
"""
from __future__ import annotations
import logging
import os
from dataclasses import dataclass
from typing import Optional
import serial.tools.list_ports
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class UsbId:
"""VID:PID пара — единственный стабильный идентификатор USB-устройства."""
vid: int
pid: int
def __str__(self) -> str:
return f"{self.vid:04X}:{self.pid:04X}"
def resolve_serial_port(usb_id: UsbId, env_var: Optional[str] = None) -> Optional[str]:
"""
Найти serial-порт устройства по VID:PID.
:param usb_id: Идентичность устройства (VID:PID).
:param env_var: Имя переменной окружения-override (escape hatch, п.1
алгоритма). Если задана и непуста возвращается
as-is без обращения к list_ports; значение не
валидируется (ответственность вызывающего кода).
:return: Имя порта (например, "/dev/cu.usbmodem1101" или "COM8"),
либо None, если устройство не найдено.
"""
if env_var:
override = os.environ.get(env_var, "").strip()
if override:
logger.debug(
"resolve_serial_port(%s): override из %s = %s",
usb_id,
env_var,
override,
)
return override
matches = [
info.device
for info in serial.tools.list_ports.comports()
if info.vid == usb_id.vid and info.pid == usb_id.pid
]
if not matches:
return None
if len(matches) > 1:
logger.warning(
"Несколько устройств %s: %s — используем первое (%s). ",
usb_id,
matches,
matches[0],
)
return matches[0]

View file

@ -0,0 +1,90 @@
# Windows Host Results
## Step 1: spike_hab.py
```bash
production git:(feature-tui-monolith) uv run pytest spike/spike_hab.py -v -s
================================================================================================================ test session starts ================================================================================================================
platform darwin -- Python 3.14.6, pytest-9.1.1, pluggy-1.6.0 -- /Users/von_akimow/Desktop/TFT_ENV/tft_manufacture_test/tools/production/.venv/bin/python3
cachedir: .pytest_cache
rootdir: /Users/von_akimow/Desktop/TFT_ENV/tft_manufacture_test/tools/production
configfile: pyproject.toml
collected 2 items
spike/spike_hab.py::test_golden_hab_no_dcd PASSED
spike/spike_hab.py::test_golden_hab_with_dcd PASSED
================================================================================================================= 2 passed in 1.07s =================================================================================================================
```
## Step 2: spike_flash.py --ports-only
```bash
# подключены M5StamPLC + плата с firmware_test
➜ production git:(feature-tui-monolith) ✗ uv run python spike/spike_flash.py --ports-only
=== Serial-порты (serial.tools.list_ports.comports) ===
/dev/cu.debug-console VID:PID=----:---- 'n/a' serial=None
/dev/cu.Bluetooth-Incoming-Port VID:PID=----:---- 'n/a' serial=None
/dev/cu.usbmodem11101 VID:PID=303A:4001 'Espressif Device' serial='3cdc75edcf6c0000'
/dev/cu.usbmodemGUXFBWDJBWTGQ3 VID:PID=1FC9:0143 'MCU-LINK (r0FB) CMSIS-DAP V3.172' serial='GUXFBWDJBWTGQ'
/dev/cu.usbmodem11301 VID:PID=1996:00AD 'MU LLC' serial=None
Сверь: firmware_test ожидается как 1996:00AD. Для M5StampPLC запиши VID:PID из вывода выше — это и есть измерение О1 (см. MONOLITH_APP_PLAN.md §О1 и DEV_ARCH.md §5).
```
## Step 3: spike_flash.py
```bash
# подключены M5StamPLC + плата в режиме SDP
➜ production git:(feature-tui-monolith) ✗ uv run python spike/spike_flash.py
=== Serial-порты (serial.tools.list_ports.comports) ===
/dev/cu.debug-console VID:PID=----:---- 'n/a' serial=None
/dev/cu.Bluetooth-Incoming-Port VID:PID=----:---- 'n/a' serial=None
/dev/cu.usbmodem11101 VID:PID=303A:4001 'Espressif Device' serial='3cdc75edcf6c0000'
/dev/cu.usbmodemGUXFBWDJBWTGQ3 VID:PID=1FC9:0143 'MCU-LINK (r0FB) CMSIS-DAP V3.172' serial='GUXFBWDJBWTGQ'
Сверь: firmware_test ожидается как 1996:00AD. Для M5StampPLC запиши VID:PID из вывода выше — это и есть измерение О1 (см. MONOLITH_APP_PLAN.md §О1 и DEV_ARCH.md §5).
=== SDP/Flashloader (SDP=0x1fc9:0x0130, Flashloader=0x15a2:0x0073) ===
Загрузка Flashloader через SDP (0x1fc9:0x0130)
Ожидание Flashloader (до 10с).... OK
✅ get_property(CURRENT_VERSION) = [1258424320]
=== configure_memory (FlexSPI NOR) ===
configure_memory: OK
✅ Гейт 0 (SDP/Flashloader): пройден
```
## Step 4: spike_flash.py --erase-all
```bash
# Подключены M5StamPLC + плата в режиме SDP
➜ production git:(feature-tui-monolith) ✗ uv run python spike/spike_flash.py --erase-all
=== Serial-порты (serial.tools.list_ports.comports) ===
/dev/cu.debug-console VID:PID=----:---- 'n/a' serial=None
/dev/cu.Bluetooth-Incoming-Port VID:PID=----:---- 'n/a' serial=None
/dev/cu.usbmodem11101 VID:PID=303A:4001 'Espressif Device' serial='3cdc75edcf6c0000'
/dev/cu.usbmodemGUXFBWDJBWTGQ3 VID:PID=1FC9:0143 'MCU-LINK (r0FB) CMSIS-DAP V3.172' serial='GUXFBWDJBWTGQ'
Сверь: firmware_test ожидается как 1996:00AD. Для M5StampPLC запиши VID:PID из вывода выше — это и есть измерение О1 (см. MONOLITH_APP_PLAN.md §О1 и DEV_ARCH.md §5).
=== SDP/Flashloader (SDP=0x1fc9:0x0130, Flashloader=0x15a2:0x0073) ===
Flashloader уже запущен — пропускаем загрузку
✅ get_property(CURRENT_VERSION) = [1258424320]
=== configure_memory (FlexSPI NOR) ===
configure_memory: OK
⚠️ Chip erase сотрёт FCB — плата не загрузится до повторной прошивки.
Наберите ERASE для подтверждения: ERASE
Выставляю timeout=200000мс (эквивалент blhost -t 200000)
flash_erase_all: OK
✅ Гейт 0 (SDP/Flashloader): пройден
```

View file

View file

@ -0,0 +1,423 @@
"""
test_flash_backend.py юнит-тесты app/flash_backend.py.
Два уровня:
1. SDP/McuBoot/detect моки (unittest.mock через monkeypatch). Железо не
нужно, спсdk-транспорт не трогаем проверяем только нашу логику
(последовательность вызовов, обработку ошибок, форму FlashProgress).
2. build_custom_hab() НЕ мокается. Использует настоящий HabImage на
детерминированном dummy-бинаре, как spike_hab.py в Фазе 0 (там же
доказана byte-exact идентичность nxpimage). Мокать spsdk здесь было бы
ложной уверенностью сама суть проверки в том, что реальный HabImage
не падает и не расходится по конфигу.
"""
from __future__ import annotations
import hashlib
from pathlib import Path
from typing import List
from unittest.mock import MagicMock, Mock
import pytest
from app import flash_backend as fb
from app.models import FlashProgress
# ─── Общие фикстуры ──────────────────────────────────────────────────────
@pytest.fixture
def events() -> List[FlashProgress]:
return []
def _collector(events: List[FlashProgress]):
return events.append
def _phases(events: List[FlashProgress]) -> list[str]:
return [e.phase for e in events]
# ─── detect_sdp / detect_cdc ────────────────────────────────────────────
def test_detect_sdp_found(monkeypatch):
monkeypatch.setattr(fb.SdpUSBInterface, "scan", Mock(return_value=[Mock()]))
assert fb.detect_sdp() is True
def test_detect_sdp_not_found(monkeypatch):
monkeypatch.setattr(fb.SdpUSBInterface, "scan", Mock(return_value=[]))
assert fb.detect_sdp() is False
def test_detect_cdc_found(monkeypatch):
monkeypatch.setattr(
fb, "resolve_serial_port", Mock(return_value="/dev/cu.usbmodem1")
)
assert fb.detect_cdc() is True
def test_detect_cdc_not_found(monkeypatch):
monkeypatch.setattr(fb, "resolve_serial_port", Mock(return_value=None))
assert fb.detect_cdc() is False
# ─── wait_for_flashloader ────────────────────────────────────────────────
def test_wait_for_flashloader_immediate(monkeypatch):
iface = Mock()
monkeypatch.setattr(fb.MbootUSBInterface, "scan", Mock(return_value=[iface]))
assert fb.wait_for_flashloader(timeout_s=1.0) is iface
def test_wait_for_flashloader_timeout(monkeypatch):
monkeypatch.setattr(fb.MbootUSBInterface, "scan", Mock(return_value=[]))
monkeypatch.setattr(fb.time, "sleep", Mock()) # не ждать реально
with pytest.raises(fb.FlashLoaderTimeoutError):
fb.wait_for_flashloader(timeout_s=0.001)
# ─── load_flashloader ────────────────────────────────────────────────────
def test_load_flashloader_already_running(monkeypatch):
iface = Mock()
scan_mboot = Mock(return_value=[iface])
scan_sdp = Mock() # не должен вызываться
monkeypatch.setattr(fb.MbootUSBInterface, "scan", scan_mboot)
monkeypatch.setattr(fb.SdpUSBInterface, "scan", scan_sdp)
result = fb.load_flashloader()
assert result is iface
scan_sdp.assert_not_called()
def test_load_flashloader_no_sdp_device(monkeypatch):
monkeypatch.setattr(fb.MbootUSBInterface, "scan", Mock(return_value=[]))
monkeypatch.setattr(fb.SdpUSBInterface, "scan", Mock(return_value=[]))
with pytest.raises(fb.DeviceNotFoundError):
fb.load_flashloader()
def test_load_flashloader_missing_bin(monkeypatch):
monkeypatch.setattr(fb.MbootUSBInterface, "scan", Mock(return_value=[]))
monkeypatch.setattr(fb.SdpUSBInterface, "scan", Mock(return_value=[Mock()]))
monkeypatch.setattr(fb, "FLASHLOADER_BIN", Mock(exists=Mock(return_value=False)))
with pytest.raises(fb.FlashBackendError, match="Не найден"):
fb.load_flashloader()
def test_load_flashloader_happy_path(monkeypatch, events):
sdp_iface = Mock()
flashloader_iface = Mock()
monkeypatch.setattr(fb.MbootUSBInterface, "scan", Mock(return_value=[]))
monkeypatch.setattr(fb.SdpUSBInterface, "scan", Mock(return_value=[sdp_iface]))
fake_bin = Mock(
exists=Mock(return_value=True), read_bytes=Mock(return_value=b"\x00" * 16)
)
monkeypatch.setattr(fb, "FLASHLOADER_BIN", fake_bin)
mock_sdp_ctx = MagicMock()
mock_sdp_ctx.__enter__.return_value = mock_sdp_ctx
monkeypatch.setattr(fb, "SDP", Mock(return_value=mock_sdp_ctx))
monkeypatch.setattr(
fb, "wait_for_flashloader", Mock(return_value=flashloader_iface)
)
result = fb.load_flashloader(progress_cb=_collector(events))
assert result is flashloader_iface
mock_sdp_ctx.write_file.assert_called_once_with(
fb.FLASHLOADER_LOAD_ADDR, b"\x00" * 16
)
mock_sdp_ctx.jump_and_run.assert_called_once_with(fb.FLASHLOADER_LOAD_ADDR)
assert _phases(events) == ["load_flashloader", "load_flashloader"]
assert events[-1].percent == 100
# ─── configure_flexspi / write_fcb_* ─────────────────────────────────────
def test_configure_flexspi_ok():
mboot = Mock(configure_memory=Mock(return_value=True))
fb.configure_flexspi(mboot)
mboot.fill_memory.assert_called_once_with(
fb.FLEXSPI_OPTION_ADDR, 4, fb.FLEXSPI_OPTION_VALUE
)
mboot.configure_memory.assert_called_once_with(
fb.FLEXSPI_OPTION_ADDR, fb.FLEXSPI_MEMORY_ID
)
def test_configure_flexspi_fail():
mboot = Mock(configure_memory=Mock(return_value=False))
with pytest.raises(fb.FlashBackendError):
fb.configure_flexspi(mboot)
def test_write_fcb_auto_uses_magic_word():
mboot = Mock(configure_memory=Mock(return_value=True))
fb.write_fcb_auto(mboot)
mboot.fill_memory.assert_called_once_with(
fb.FLEXSPI_OPTION_ADDR, 4, fb.FLEXSPI_FCB_VALUE
)
def test_write_fcb_explicit_missing_file(tmp_path):
mboot = Mock()
with pytest.raises(fb.FlashBackendError, match="не найден"):
fb.write_fcb_explicit(mboot, tmp_path / "nope.bin")
def test_write_fcb_explicit_ok(tmp_path):
fcb = tmp_path / "w25q128_fdcb.bin"
fcb.write_bytes(b"\xab" * 512)
mboot = Mock(write_memory=Mock(return_value=True))
fb.write_fcb_explicit(mboot, fcb)
mboot.write_memory.assert_called_once_with(fb.FLASH_BASE, b"\xab" * 512, mem_id=0)
def test_write_fcb_explicit_write_fails(tmp_path):
fcb = tmp_path / "w25q128_fdcb.bin"
fcb.write_bytes(b"\xab" * 512)
mboot = Mock(write_memory=Mock(return_value=False))
with pytest.raises(fb.FlashBackendError):
fb.write_fcb_explicit(mboot, fcb)
# ─── flash() ──────────────────────────────────────────────────────────────
def _mock_mcuboot_ctx(monkeypatch, **method_returns):
"""Подменяет fb.McuBoot на context-manager мок с заданными return_value
у fill_memory/configure_memory/flash_erase_region/write_memory (все True
по умолчанию, кроме явно переопределённых)."""
defaults = dict(
configure_memory=True,
flash_erase_region=True,
flash_erase_all=True,
write_memory=True,
)
defaults.update(method_returns)
ctx = MagicMock()
ctx.__enter__.return_value = ctx
for name, ret in defaults.items():
getattr(ctx, name).return_value = ret
monkeypatch.setattr(fb, "McuBoot", Mock(return_value=ctx))
return ctx
def test_flash_missing_file(events, tmp_path):
with pytest.raises(fb.FlashBackendError, match="Файл не найден"):
fb.flash(tmp_path / "nope_hab.bin", progress_cb=_collector(events))
assert events == []
def test_flash_ram_only(monkeypatch, events, tmp_path):
hab_bin = tmp_path / "fw_hab.bin"
hab_bin.write_bytes(b"\xd1" + b"\x00" * 63)
sdp_iface = Mock()
monkeypatch.setattr(fb.SdpUSBInterface, "scan", Mock(return_value=[sdp_iface]))
mcuboot_mock = Mock()
monkeypatch.setattr(fb, "McuBoot", mcuboot_mock)
mock_sdp_ctx = MagicMock()
mock_sdp_ctx.__enter__.return_value = mock_sdp_ctx
monkeypatch.setattr(fb, "SDP", Mock(return_value=mock_sdp_ctx))
fb.flash(hab_bin, ram_only=True, progress_cb=_collector(events))
addr = fb.FLASH_BASE + fb.HAB_OFFSET
mock_sdp_ctx.write_file.assert_called_once_with(addr, hab_bin.read_bytes())
mock_sdp_ctx.jump_and_run.assert_called_once_with(addr)
mcuboot_mock.assert_not_called()
assert _phases(events) == ["load_flashloader", "done"]
def test_flash_ram_only_no_sdp(monkeypatch, events, tmp_path):
hab_bin = tmp_path / "fw_hab.bin"
hab_bin.write_bytes(b"\xd1")
monkeypatch.setattr(fb.SdpUSBInterface, "scan", Mock(return_value=[]))
with pytest.raises(fb.DeviceNotFoundError):
fb.flash(hab_bin, ram_only=True, progress_cb=_collector(events))
def test_flash_happy_path_auto_fcb(monkeypatch, events, tmp_path):
hab_bin = tmp_path / "fw_hab.bin"
hab_bin.write_bytes(b"\xd1" + b"\x00" * 127)
flashloader_iface = Mock()
monkeypatch.setattr(fb, "load_flashloader", Mock(return_value=flashloader_iface))
ctx = _mock_mcuboot_ctx(monkeypatch)
fb.flash(hab_bin, progress_cb=_collector(events))
ctx.flash_erase_region.assert_called_once()
args, kwargs = ctx.flash_erase_region.call_args
assert args[0] == fb.FLASH_BASE
assert kwargs["mem_id"] == 0
ctx.write_memory.assert_called_once()
wargs, wkwargs = ctx.write_memory.call_args
assert wargs[0] == fb.FLASH_BASE + fb.HAB_OFFSET
assert wargs[1] == hab_bin.read_bytes()
assert wkwargs["mem_id"] == 0
assert "progress_callback" in wkwargs
ctx.reset.assert_called_once_with(reopen=False)
assert _phases(events) == ["configure", "erase", "fcb", "write", "reset", "done"]
assert events[-1].percent == 100
def test_flash_happy_path_explicit_fcb(monkeypatch, events, tmp_path):
hab_bin = tmp_path / "fw_hab.bin"
hab_bin.write_bytes(b"\xd1" + b"\x00" * 15)
fcb_bin = tmp_path / "w25q512_fdcb.bin"
fcb_bin.write_bytes(b"\xab" * 512)
monkeypatch.setattr(fb, "load_flashloader", Mock(return_value=Mock()))
ctx = _mock_mcuboot_ctx(monkeypatch)
fb.flash(hab_bin, fcb_path=fcb_bin, progress_cb=_collector(events))
# explicit FCB → write_memory вызывается ДВАЖДЫ: FCB-блоб + сам HAB-образ
assert ctx.write_memory.call_count == 2
first_call_args = ctx.write_memory.call_args_list[0].args
assert first_call_args[0] == fb.FLASH_BASE
assert first_call_args[1] == b"\xab" * 512
def test_flash_write_memory_fails(monkeypatch, events, tmp_path):
hab_bin = tmp_path / "fw_hab.bin"
hab_bin.write_bytes(b"\xd1")
monkeypatch.setattr(fb, "load_flashloader", Mock(return_value=Mock()))
_mock_mcuboot_ctx(monkeypatch, write_memory=False)
with pytest.raises(fb.FlashBackendError, match="write_memory"):
fb.flash(hab_bin, progress_cb=_collector(events))
# reset/done не должны наступить после ошибки записи
assert "reset" not in _phases(events)
assert "done" not in _phases(events)
def test_flash_progress_callback_reports_bytes(monkeypatch, events, tmp_path):
"""write_memory реально дёргает progress_callback(current, total) — см.
Фазу 0. Проверяем, что наш адаптер конвертирует это в FlashProgress."""
hab_bin = tmp_path / "fw_hab.bin"
payload = b"\xd1" + b"\x00" * 999
hab_bin.write_bytes(payload)
monkeypatch.setattr(fb, "load_flashloader", Mock(return_value=Mock()))
ctx = _mock_mcuboot_ctx(monkeypatch)
def _fake_write_memory(address, data, mem_id=0, progress_callback=None):
progress_callback(len(data) // 2, len(data))
progress_callback(len(data), len(data))
return True
ctx.write_memory.side_effect = _fake_write_memory
fb.flash(hab_bin, progress_cb=_collector(events))
write_events = [e for e in events if e.phase == "write" and "/" in e.message]
assert len(write_events) == 2
assert write_events[-1].percent == 100
# ─── erase_chip() ─────────────────────────────────────────────────────────
def test_erase_chip_happy_path(monkeypatch, events):
flashloader_iface = MagicMock()
monkeypatch.setattr(fb, "load_flashloader", Mock(return_value=flashloader_iface))
ctx = _mock_mcuboot_ctx(monkeypatch)
fb.erase_chip(progress_cb=_collector(events))
assert flashloader_iface.device.timeout == fb.ERASE_ALL_TIMEOUT_MS
ctx.flash_erase_all.assert_called_once_with(mem_id=fb.FLEXSPI_MEMORY_ID)
ctx.reset.assert_called_once_with(reopen=False)
assert _phases(events) == ["configure", "erase", "reset", "done"]
def test_erase_chip_fails(monkeypatch, events):
monkeypatch.setattr(fb, "load_flashloader", Mock(return_value=MagicMock()))
_mock_mcuboot_ctx(monkeypatch, flash_erase_all=False)
with pytest.raises(fb.FlashBackendError, match="flash_erase_all"):
fb.erase_chip(progress_cb=_collector(events))
assert "done" not in _phases(events)
# ─── build_custom_hab() — НЕ мокается, реальный HabImage ─────────────────
def _dummy_bin(tmp_path: Path, size: int = 512) -> Path:
seed = hashlib.sha256(b"test-flash-backend-golden").digest()
data = (seed * (size // len(seed) + 1))[:size]
p = tmp_path / "dummy_app.bin"
p.write_bytes(data)
return p
def test_build_custom_hab_no_dcd(tmp_path, events):
raw = _dummy_bin(tmp_path)
out = fb.build_custom_hab(raw, use_dcd=False, progress_cb=_collector(events))
assert out.exists()
assert out.stat().st_size > 0
assert _phases(events) == ["hab_build", "hab_build"]
assert events[-1].percent == 100
def test_build_custom_hab_with_dcd(tmp_path):
if not fb.REAL_DCD_BIN.exists():
pytest.skip(f"Реальный DCD не найден: {fb.REAL_DCD_BIN}")
raw = _dummy_bin(tmp_path)
out = fb.build_custom_hab(raw, use_dcd=True)
assert out.exists()
assert out.stat().st_size > 0
def test_build_custom_hab_dcd_requested_but_missing(monkeypatch, tmp_path):
raw = _dummy_bin(tmp_path)
monkeypatch.setattr(fb, "REAL_DCD_BIN", tmp_path / "nonexistent_dcd.bin")
with pytest.raises(fb.HabBuildError, match="DCD"):
fb.build_custom_hab(raw, use_dcd=True)
def test_build_custom_hab_missing_input(tmp_path):
with pytest.raises(fb.HabBuildError):
fb.build_custom_hab(tmp_path / "does_not_exist.bin", use_dcd=False)
def test_build_custom_hab_bytes_match_golden(tmp_path):
"""Дополнительная перекрёстная проверка со spike_hab.py: два независимых
вызова build_custom_hab с одинаковым входом дают одинаковый результат
(детерминированность), а не просто 'не упало'."""
raw = _dummy_bin(tmp_path)
out1 = fb.build_custom_hab(raw, use_dcd=False)
out2 = fb.build_custom_hab(raw, use_dcd=False)
assert out1.read_bytes() == out2.read_bytes()