HaloKeymind/test/test_esp32_usb_session_fix.py

"""Build-local native CDC hooks must precede delayed framework events."""
from pathlib import Path
import importlib.util
import os
import tempfile
import unittest

ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location("esp_usb_fix", ROOT / "scripts/esp32_usb_session_fix.py")
FIX = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(FIX)
CDC = '''#include "esp32-hal-tinyusb.h"
TUD_CDC_DESCRIPTOR(*itf, str_index, 0x85, 64, 0x03, 0x84, 64)
void tud_cdc_line_state_cb(uint8_t itf, bool dtr, bool rts)
{
    devices[itf]->_onLineState(dtr, rts);
}
void USBCDC::_onLineState(bool _dtr, bool _rts){
    if(dtr == _dtr && rts == _rts) return;
    dtr = _dtr;
    arduino_usb_event_post();
}
void USBCDC::_onRX(){
    tud_cdc_n_read();
    xQueueSend();
}
void tud_cdc_tx_complete_cb(uint8_t itf){
    devices[itf]->_onTX();
}
'''
USB = '''#include "esp32-hal-tinyusb.h"
void tud_mount_cb(void){
    arduino_usb_event_post();
}
void tud_umount_cb(void){
    arduino_usb_event_post();
}
'''


def installed_tested_tinyusb_cores(packages):
    # This optional installed-source qualification covers the 2.0.17 layout
    # used by the golden anchors. A C6 HWCDC build can replace the default
    # package with 3.x; it must not silently change which TinyUSB SDK is tested.
    result = []
    for package in sorted(packages.glob("framework-arduinoespressif32*")):
        core = package / "cores/esp32"
        if not (core / "USBCDC.cpp").is_file() or not (core / "esp_arduino_version.h").is_file():
            continue
        if FIX.hwcdc_framework_version(core / "USBCDC.cpp") == (2, 0, 17):
            result.append(core)
    return result


class Esp32UsbSessionFixTests(unittest.TestCase):
    def test_owner_hooks_precede_state_changes_and_rx_queue(self):
        patched = FIX.patched_cdc_source(CDC)
        self.assertLess(patched.index("meshEsp32TinyUsbCdcLineState(dtr);"),
                        patched.index("devices[itf]->_onLineState(dtr, rts);"))
        self.assertLess(patched.index("!meshEsp32TinyUsbAcceptRx()"),
                        patched.index("xQueueSend();"))
        self.assertLess(patched.index("meshEsp32TinyUsbTxComplete();"),
                        patched.index("devices[itf]->_onTX();"))
        self.assertIn("usbd_edpt_busy(0, 0x84)", patched)
        self.assertEqual(FIX.patched_cdc_source(patched), patched)
        device = FIX.patched_usb_source(USB)
        for signature in (FIX.MOUNT, FIX.UNMOUNT):
            callback = device[device.index(signature):]
            self.assertLess(callback.index("meshEsp32TinyUsbDeviceSessionBoundary"),
                            callback.index("arduino_usb_event_post"))
        self.assertEqual(FIX.patched_usb_source(device), device)

    def test_changed_sdk_or_endpoint_fails_closed(self):
        for original, changed in (("0x84", "0x82"), (FIX.LINE_STATE, "unknown() {"),
                                  (FIX.RX, "changed_rx() {"),
                                  (FIX.TX_COMPLETE, "changed_tx() {")):
            with self.subTest(changed=original), self.assertRaises(RuntimeError):
                FIX.patched_cdc_source(CDC.replace(original, changed))
        with self.assertRaises(RuntimeError):
            FIX.patched_usb_source(USB.replace(FIX.UNMOUNT, "changed_unmount() {"))
        self.assertEqual(FIX.patched_cdc_source(CDC.replace("\n", "\r\n")),
                         FIX.patched_cdc_source(CDC))

    def test_only_primary_native_cdc_uses_build_local_copy(self):
        class Environment(dict):
            def subst(self, value):
                return self["BUILD_DIR"]
            def File(self, value):
                return value
        class Node:
            def __init__(self, path): self.path = path
            def srcnode(self): return self
            def get_abspath(self): return str(self.path)
        with tempfile.TemporaryDirectory(prefix="meshcore-esp-usb-hook-") as directory:
            directory = Path(directory)
            source = directory / "USBCDC.cpp"
            source.write_text(CDC, encoding="utf-8")
            node = Node(source)
            env = Environment(BUILD_DIR=str(directory / "build"),
                              CPPDEFINES=[("ARDUINO_USB_CDC_ON_BOOT", 1), ("ARDUINO_USB_MODE", 0)])
            result = Path(FIX.replace_source(env, node))
            self.assertNotEqual(result, source)
            self.assertEqual(source.read_text(), CDC)
            self.assertEqual(result.read_text(), FIX.patched_cdc_source(CDC))
            for definitions in ([], [("ARDUINO_USB_CDC_ON_BOOT", 1), ("ARDUINO_USB_MODE", 1)],
                                [("ARDUINO_USB_CDC_ON_BOOT", 0), ("ARDUINO_USB_MODE", 0)]):
                env["CPPDEFINES"] = definitions
                self.assertIs(FIX.replace_source(env, node), node)
            # Selecting a newer HWCDC SDK for optional qualification must
            # never relax the actual mode0 build's changed-layout rejection.
            env["CPPDEFINES"] = [("ARDUINO_USB_CDC_ON_BOOT", 1), ("ARDUINO_USB_MODE", 0)]
            source.write_text(CDC.replace(FIX.RX, "void USBCDC::_onRX() {\n"), encoding="utf-8")
            with self.assertRaises(RuntimeError):
                FIX.replace_source(env, node)

    def test_installed_sdk_selection_uses_exact_version_and_alternate_packages(self):
        with tempfile.TemporaryDirectory(prefix="meshcore-tinyusb-sdk-selection-") as directory:
            packages = Path(directory)
            cores = {}
            for name, version in (("framework-arduinoespressif32", (3, 1, 3)),
                                  ("framework-arduinoespressif32@src-old", (2, 0, 17)),
                                  ("framework-arduinoespressif32@src-new", (3, 3, 11))):
                core = packages / name / "cores/esp32"
                core.mkdir(parents=True)
                (core / "USBCDC.cpp").write_text(CDC, encoding="ascii")
                (core / "esp_arduino_version.h").write_text(
                    "\n".join("#define ESP_ARDUINO_VERSION_" + key + " " + str(value)
                              for key, value in zip(("MAJOR", "MINOR", "PATCH"), version)) + "\n",
                    encoding="ascii")
                cores[version] = core
            self.assertEqual(installed_tested_tinyusb_cores(packages), [cores[(2, 0, 17)]])
            (cores[(2, 0, 17)] / "USBCDC.cpp").unlink()
            self.assertEqual(installed_tested_tinyusb_cores(packages), [])

    def test_installed_sdk_compatibility_when_available(self):
        # Package-free CI still exercises the golden anchors above. Check an
        # installed SDK on Windows and Unix, including a custom PIO core dir.
        core = Path(os.environ.get("PLATFORMIO_CORE_DIR", str(Path.home() / ".platformio")))
        packages = installed_tested_tinyusb_cores(core / "packages")
        if not packages:
            self.skipTest("installed Arduino-ESP32 2.0.17 TinyUSB SDK unavailable")
        for package in packages:
            for name, patcher in (("USBCDC.cpp", FIX.patched_cdc_source), ("USB.cpp", FIX.patched_usb_source)):
                with self.subTest(package=package, source=name):
                    original = (package / name).read_text(encoding="utf-8")
                    self.assertIn("meshEsp32TinyUsb", patcher(original))

    def test_all_esp32_profiles_inherit_hook(self):
        config = (ROOT / "platformio.ini").read_text()
        base = config[config.index("[esp32_base]"):config.index("[esp32_ota]")]
        self.assertIn("pre:scripts/esp32_usb_session_fix.py", base)


if __name__ == "__main__":
    unittest.main()