diff --git a/tools/production/app/firmware_test_hab.bin b/tools/production/app/firmware_test_hab.bin new file mode 100644 index 0000000..1090ebc Binary files /dev/null and b/tools/production/app/firmware_test_hab.bin differ diff --git a/tools/production/app/flash_backend.py b/tools/production/app/flash_backend.py new file mode 100644 index 0000000..d527d8b --- /dev/null +++ b/tools/production/app/flash_backend.py @@ -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 diff --git a/tools/production/app/usb_ports.py b/tools/production/app/usb_ports.py new file mode 100644 index 0000000..ed78b4d --- /dev/null +++ b/tools/production/app/usb_ports.py @@ -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] diff --git a/tools/production/spike/output_macos.md b/tools/production/spike/output_macos.md new file mode 100644 index 0000000..52d4934 --- /dev/null +++ b/tools/production/spike/output_macos.md @@ -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): пройден +``` diff --git a/output_windows.md b/tools/production/spike/output_windows.md similarity index 100% rename from output_windows.md rename to tools/production/spike/output_windows.md diff --git a/tools/production/tests/__init__.py b/tools/production/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/production/tests/test_flash_backend.py b/tools/production/tests/test_flash_backend.py new file mode 100644 index 0000000..5021296 --- /dev/null +++ b/tools/production/tests/test_flash_backend.py @@ -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()