Balsam_167/tests/can_boot/run_test.py

"""Проверяет существующее ядро, реальный клиент SETGUI и остановку 167 перед CAN Boot."""  # Никакие операции с физической платой не выполняются.
import ctypes  # Передаёт CAN-кадры Python-клиента настоящему C-ядру и аппаратному адаптеру.
import json  # Выбирает имя реального BIN по паспорту текущего prebuild, а не по зашитому SHA прошлой сборки.
import os  # Читает выбранный host-компилятор и путь общей библиотеки.
from pathlib import Path  # Находит production исходники независимо от каталога запуска.
import re  # Извлекает существующую защиту Flash для повторно используемой модели DAC.
import subprocess  # Выполняет C-проверки отдельными процессами с контролем ошибок.
import sys  # Подключает тот же клиент protocan, который использует SETGUI.
import tempfile  # Изолирует временные EXE/DLL и извлечённые production-фрагменты.
from types import SimpleNamespace  # Предоставляет клиенту образ с данными, версией и CRC32.
import zlib  # Независимо вычисляет заявленный CRC32 транспортного файла.

ROOT = Path(__file__).resolve().parents[2]  # Корень 167 содержит прошивку и её собранный SCI8.
TEMPLATES = Path(os.environ.get("TEMPLATES_ROOT", ROOT.parent / "newProject/SETGUI/third_party/templates"))  # Проверяет реально компилируемую общую зависимость.
CORE = TEMPLATES / "c/protocan-boot"  # Содержит production-ядро, F28335-порт и его регрессии.
CC = os.environ.get("CC", "C:/msys64/ucrt64/bin/gcc.exe")  # Позволяет использовать тот же GCC, что остальные аппаратные модели проекта.
os.environ["PATH"] = str(Path(CC).parent) + os.pathsep + os.environ["PATH"]  # Делает DLL выбранного GCC доступными только процессам проверки.
sys.path.insert(0, str(TEMPLATES / "python"))  # Импортирует действующие codec и transfer-класс SETGUI без собственной реализации протокола.
from protocan.can_boot import CanBootTarget, CanBootTransfer  # Сравнивает обе стороны реально используемого протокола.
from protocan.transport import FrameParser, build_frame  # Повторно использует разбор кадров моста приложения SETGUI.

image_path = Path(os.environ["CAN167_IMAGE"]) if "CAN167_IMAGE" in os.environ else ROOT / "Bin/Can167/release" / ("firmware_" + json.loads((ROOT / "Bin/Can167/generated/firmware_identity.json").read_text(encoding="utf-8"))["build_id"] + ".bin")  # Явный путь либо паспорт prebuild однозначно выбирают текущий BIN при любом Build ID.
data = image_path.read_bytes()  # Реальный TI SCI8 обязателен для сквозной проверки клиента и адресов.
with tempfile.TemporaryDirectory(prefix="can167-") as tmp:  # Сохраняет только результаты текущего запуска и удаляет их после освобождения DLL.
    tmp = Path(tmp)  # Формирует переносимые пути всех временных объектов теста.
    loader_source = (CORE / "projects/ccs/f28335/main.c").read_text(encoding="utf-8")  # Читает production CCS-main, где recovery определяет адрес после reset.
    loader_address = loader_source[loader_source.index("static unsigned int boot_periph_id("):loader_source.index("static bool ram_word(")]  # Берёт точную функцию чтения перемычек без дублирования её алгоритма в тесте.
    (tmp / "production_boot_address.inc").write_text(loader_address, encoding="utf-8")  # Изолирует аппаратную функцию загрузчика в текущем временном каталоге.
    application_source = (ROOT / "Source/Internal/peripher.c").read_text(encoding="utf-8")  # Источник аппаратного Mode остаётся существующим get_Mode приложения 167.
    application_address = re.search(r"Mode\s*=\s*1;.*?Mode\s*\+=\s*8\s*\*\s*!GpioDataRegs\.GPBDAT\.bit\.GPIO51;", application_source, re.S)  # Извлекает все пять production-назначений до определения Desk.
    assert application_address is not None  # Отсутствие ожидаемого источника Mode требует обновить аппаратную проверку, а не подставить тестовый адрес.
    (tmp / "production_application_address.inc").write_text(application_address.group() + "\n", encoding="utf-8")  # Сохраняет извлечённый фрагмент приложения без перекодирования русских комментариев.
    address_exe = tmp / "address_gpio.exe"  # Отдельная программа перебирает все положения четырёх аппаратных перемычек.
    subprocess.run([CC, "-std=c99", "-Wall", "-Wextra", "-Werror", "-I" + str(tmp), str(ROOT / "tests/can_boot/address_gpio_test.c"), "-o", str(address_exe)], check=True)  # Компилирует оба точных production-фрагмента вместе с минимальными GPIO-регистрами модели.
    subprocess.run([str(address_exe)], check=True, timeout=10)  # Проверяет совпадение Mode приложения и адреса recovery для всех 16 комбинаций.
    flags = [CC, "-std=c99", "-O2", "-Wall", "-Wextra", "-Werror", "-Wno-misleading-indentation", "-Wno-error=unused-variable", "-DPCAN_BOOT_WIDE_BYTES"]  # Проверяет широкие октеты C28x; допускает существующий стиль однострочных upstream-тестов и неиспользуемые регистры аппаратных заглушек.
    cases = [("core", [], ["src/pcan_boot.c", "tests/test_pcan_boot.c"]), ("store", [], ["ports/tms320f28335/f28335_boot_store.c", "tests/test_f28335_boot_store.c"]), ("loader", [], ["src/pcan_boot.c", "ports/tms320f28335/f28335_boot_store.c", "tests/test_f28335_can_boot.c"]), ("application", ["-DF28335_BOOT_LOADER=0"], ["src/pcan_boot.c", "ports/tms320f28335/f28335_boot_store.c", "tests/test_f28335_can_boot.c"])]  # Сохраняет upstream-сценарии сессий, обрыва питания, пробного запуска, отката и подтверждения main.
    for name, defines, sources in cases:  # Выполняет существующие регрессии общей библиотеки без изменения её файлов.
        exe = tmp / (name + ".exe")  # Разделяет результаты четырёх вариантов.
        subprocess.run([*[f for f in flags if name != "core" or f != "-DPCAN_BOOT_WIDE_BYTES"], "-Iinclude", "-Itests/f28335_stubs", *defines, *sources, "-o", str(exe)], cwd=CORE, check=True)  # Компилирует действующие C-тесты вместе с production алгоритмами.
        subprocess.run([str(exe), *([str(image_path.resolve())] if name == "store" else [])], check=True, timeout=30)  # Store дополнительно обязан принять настоящий SCI8 приложения 167.
    flash = (ROOT / "Source/Internal/flash_rs.c").read_text(encoding="utf-8")  # Повторно используемый DAC-тест включает настоящую read-only защиту Flash.
    guard = re.search(r"^static unsigned int flash_load_guard\([^\n]*\n\{[^\n]*\n[^\n]*\n\}", flash, re.M)  # Берёт существующее тело функции вместо копирования её логики в тест.
    (tmp / "production_flash_guard.inc").write_text(guard.group() + "\n", encoding="utf-8")  # Сохраняет извлечённый фрагмент в UTF-8 только внутри тестового каталога.
    exe = tmp / "load_stop.exe"  # Отдельная проверка реального DAC.c моделирует физические START/STOP и код 4 мА.
    subprocess.run([*flags, "-Itests/can_boot/stubs", "-Itests/load_control/stubs", "-ISource/Internal/Include", "-I" + str(tmp), "tests/can_boot/load_stop_test.c", "-o", str(exe)], cwd=ROOT, check=True)  # Включает Boot-путь модели без замены самого автомата нагрузки.
    subprocess.run([str(exe)], check=True, timeout=10)  # Зависание STOP либо повторный START завершают проверку ошибкой.
    library = tmp / "transfer.dll"  # Экспортирует аппаратную модель общего порта для существующего Python-клиента.
    subprocess.run([*flags, "-shared", "-I" + str(CORE), "-I" + str(CORE / "tests/f28335_stubs"), str(CORE / "src/pcan_boot.c"), str(CORE / "ports/tms320f28335/f28335_boot_store.c"), str(ROOT / "tests/can_boot/transfer_model.c"), "-o", str(library)], check=True)  # Линкует реальное ядро с реальным портом F28335 в режиме загрузчика.
    model = ctypes.CDLL(str(library))  # Получает только тестовые аппаратные операции, не обращаясь к установленному драйверу CAN.
    octets = ctypes.c_uint * 8  # PCAN_BOOT_WIDE_BYTES использует unsigned int для октета, как аппаратный C28x-порт.
    model.model_frame.argtypes = [ctypes.c_uint32, ctypes.POINTER(ctypes.c_uint), ctypes.c_uint, ctypes.POINTER(ctypes.c_uint32), ctypes.POINTER(ctypes.c_uint)]  # Сохраняет размеры ID и массивов при вызове DLL.
    model.model_reset()  # Первое обновление происходит с пустой EEPROM без подтверждённого слота.
    target = CanBootTarget(device_type=1, device=8, product_type=0x2833, hardware_min=1, hardware_max=1)  # Проверяет закреплённый Excel адрес задатчика PERIPH 8 с прежним аппаратным кодом F28335.
    def transfer(crc=None):  # Выполняет полную последовательность клиента через production callbacks EEPROM.
        image = SimpleNamespace(data=data, crc32=zlib.crc32(data) if crc is None else crc, version=0x01010002)  # Версия соответствует ручной 1.1.2 текущей 167.
        client = CanBootTransfer(image, target)  # Использует неизменённый клиент SETGUI, включая окно и выбор DATA_A/B.
        pending = client.start()  # Реальная первая команда — ENTER_BOOT с ненулевой session.
        while pending:  # Каждое следующее окно появляется только после подходящего production-статуса.
            requests, pending = pending, []  # Не переписывает логику сессии либо расчёт следующего блока в модели.
            for raw in requests:  # Передаёт 16 кадров окна последовательно, как настоящий CAN-мост.
                frame = FrameParser().feed(raw.encode())[0]  # Проверяет также транспортную упаковку существующего Python-клиента.
                reply_id, reply = ctypes.c_uint32(), octets()  # Предоставляет место для реального MBOX4 ответа аппаратного порта.
                count = model.model_frame(frame.can_id, octets(*frame.data), len(frame.data), ctypes.byref(reply_id), reply)  # Выполняет настоящий parser, CRC и запись неактивного слота.
                if count:  # В середине окна MCU не должен выдавать лишние ACK.
                    response = FrameParser().feed(build_frame(reply_id.value, bytes(reply[:]), to_can=False).encode())[0]  # Возвращает ответ через тот же codec, что физический мост SETGUI.
                    next_frames, _ = client.handle_status(response)  # Только клиент решает, какие команды следует отправить далее.
                    pending.extend(next_frames)  # Сохраняет очередь следующего окна либо VERIFY/COMMIT/REBOOT.
        assert client.finished and client.percent == 100  # Не считает передачу завершённой по одному ACK без полного протокола.
    try:  # Освобождение DLL требуется даже при неуспешной проверке, чтобы удалить временный каталог Windows.
        transfer()  # Первая передача записывает A и публикует его как pending.
        assert model.model_select() == 0 and model.model_confirm(0)  # Только пробный запуск даёт право стать подтверждённым active.
        model.model_reopen()  # Следующая сессия должна увидеть active A из энергонезависимого журнала.
        try:  # Повреждённая заявленная CRC не должна испортить ранее подтверждённый образ.
            transfer(zlib.crc32(data) ^ 1)  # Передаёт те же данные с намеренно неверной метаданной CRC32.
            raise AssertionError("CRC mismatch accepted")  # Отсутствие отказа означает нарушение проверки целостности.
        except RuntimeError as error:  # Клиент обязан получить точный CRC_ERROR от производственного ядра.
            assert "0x06" in str(error)  # Ошибка другого этапа не подтверждает работу CRC-проверки.
        model.model_reopen(); transfer()  # Новая полная передача записывает B, сохраняя подтверждённый A.
        assert model.model_select() == 1 and model.model_select() == 0  # Повторный reset без CONFIRM откатывает неудачный B к A.
        assert model.model_parameters_intact()  # Ни одно окно, журнал или откат не меняет область параметров.
        print("PASS: actual SETGUI client -> F28335 port -> real SCI8, A/B, CRC rejection, trial rollback, parameter protection")  # Сообщает результат сквозной программной проверки в переносимом выводе консоли Windows.
    finally:  # Закрывает только библиотеку, созданную этой проверкой.
        close = ctypes.WinDLL("kernel32", use_last_error=True).FreeLibrary  # Освобождает тестовую DLL перед удалением её каталога.
        close.argtypes = [ctypes.c_void_p]; close.restype = ctypes.c_int  # Передаёт полный размер указателя процесса Windows.
        assert close(model._handle)  # Не оставляет заблокированный временный файл после завершения тестов.