velxio/test/test-esp32-cam/tests/test_dma_smoke_live.py

"""
Phase-2 live test — proves I²S0 in slave-RX mode + linked-list DMA
delivers a frame's worth of bytes into RAM and fires `in_suc_eof`.

Drives the dma_smoke.ino sketch which pokes I²S0 directly (no
esp32-camera helpers). Failures here point at our hw/misc/esp32_i2s_cam.c
implementation specifically, not at the camera driver.

@unittest.expectedFailure until Phase 2 lands. The dma_smoke sketch
documents the EXACT register sequence ll_cam_start() issues — see
autosearch/08_dvp_i2s_spec.md for the spec.

What "passing" means at this layer:
  - serial line "EOF after Nms" with N <= 5000 (in_suc_eof fired)
  - serial line "buf[0..31] = ..." showing 32 hex bytes that match
    whatever the host pushed before/during boot

The host-push side isn't shipped at this layer either, so the test
asserts a known synthetic fill pattern (all-0xAA) that the QEMU model
produces when no host frame is queued. That's a Phase-2 design
choice: idle pattern = 0xAA so we can disambiguate "EOF fired but no
data" from "no EOF at all".
"""

from __future__ import annotations

import asyncio
import json
import os
import pathlib
import re
import socket
import unittest
from urllib.parse import urlparse


_THIS_DIR = pathlib.Path(__file__).resolve().parent
_TEST_ROOT = _THIS_DIR.parent
_SKETCH = _TEST_ROOT / "sketches" / "dma_smoke" / "dma_smoke.ino"


def _backend_base_url() -> str:
    return os.environ.get("VELXIO_BACKEND_URL", "").strip()


def _backend_reachable(timeout: float = 0.5) -> bool:
    url = _backend_base_url()
    if not url:
        return False
    try:
        u = urlparse(url)
        host = u.hostname or "localhost"
        port = u.port or (443 if u.scheme == "https" else 80)
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False
    except Exception:
        return False


def _ws_url(client_id: str) -> str:
    base = _backend_base_url() or "http://localhost:8001"
    u = urlparse(base)
    scheme = "wss" if u.scheme == "https" else "ws"
    host = u.hostname or "localhost"
    port = f":{u.port}" if u.port else ""
    return f"{scheme}://{host}{port}/api/simulation/ws/{client_id}"


@unittest.skipUnless(
    _backend_reachable(),
    "Velxio backend not reachable on $VELXIO_BACKEND_URL",
)
class TestDmaSmokeLive(unittest.IsolatedAsyncioTestCase):

    COMPILE_TIMEOUT = 300.0
    SERIAL_TIMEOUT = 30.0

    async def test_i2s_dma_eof_and_buffer_dump(self):
        """Compile dma_smoke.ino, boot under QEMU, watch for the EOF-
        timing line and the 32-byte hex dump.

        Phase 2 deliverable: PASSING since hw/misc/esp32_i2s_cam.c shipped.
        The hex dump contains the idle pattern (00 AA 00 AA … per
        autosearch/08's dma_elem_t packing) — proves the DMA walker
        writes both pixel-byte slots correctly and the EOF interrupt
        fires."""
        try:
            import httpx           # type: ignore
            import websockets      # type: ignore
        except ImportError as exc:
            self.skipTest(f"missing test deps: {exc}")

        sketch = _SKETCH.read_text(encoding="utf-8")
        sketch_name = _SKETCH.name

        async with httpx.AsyncClient(
            base_url=_backend_base_url(), timeout=self.COMPILE_TIMEOUT,
        ) as http:
            res = await http.post("/api/compile/", json={
                "files": [{"name": sketch_name, "content": sketch}],
                "board_fqbn": "esp32:esp32:esp32cam",
            })
            self.assertEqual(
                res.status_code, 200,
                f"/api/compile/ HTTP {res.status_code}: {res.text[:400]}",
            )
            body = res.json()
            if not body.get("success"):
                self.skipTest(
                    f"compile failure: "
                    f"{(body.get('error') or body.get('stderr', ''))[:600]}"
                )
            firmware_b64 = body.get("binary_content") or body.get("firmware_b64")
            self.assertTrue(firmware_b64)

        client_id = (
            f"esp32-cam-dma-test-"
            f"{int(asyncio.get_event_loop().time() * 1000)}"
        )
        url = _ws_url(client_id)

        async with websockets.connect(
            url, ping_interval=None, max_size=4 * 1024 * 1024,
        ) as ws:
            await ws.send(json.dumps({
                "type": "start_esp32",
                "data": {"board": "esp32-cam", "firmware_b64": firmware_b64},
            }))

            saw_armed = False
            eof_ms: int | None = None
            buf_dump: str | None = None
            buf = ""
            armed_re = re.compile(r"i2s_rx armed, eof_num=(\d+)")
            eof_re = re.compile(r"EOF after (\d+)ms")
            # 32 hex bytes; tolerate trailing whitespace OR newline OR
            # end-of-string. Don't require a final \n — the firmware
            # may emit "buf[…] = … " without a CR/LF on some chunked
            # serial paths.
            dump_re = re.compile(
                r"buf\[0\.\.31\] = ((?:[0-9A-F]{2} ?){32})"
            )

            try:
                deadline = asyncio.get_event_loop().time() + self.SERIAL_TIMEOUT
                while asyncio.get_event_loop().time() < deadline:
                    remaining = deadline - asyncio.get_event_loop().time()
                    if remaining <= 0:
                        break
                    raw = await asyncio.wait_for(ws.recv(), timeout=remaining)
                    try:
                        msg = json.loads(raw)
                    except json.JSONDecodeError:
                        continue
                    if msg.get("type") == "serial_output":
                        buf += msg.get("data", {}).get("data", "")
                        if armed_re.search(buf):
                            saw_armed = True
                        m_eof = eof_re.search(buf)
                        if m_eof and eof_ms is None:
                            eof_ms = int(m_eof.group(1))
                        m_dump = dump_re.search(buf)
                        if m_dump and buf_dump is None:
                            buf_dump = m_dump.group(1).strip()
                        if saw_armed and eof_ms is not None and buf_dump:
                            break
            except asyncio.TimeoutError:
                pass
            finally:
                try:
                    await ws.send(json.dumps({"type": "stop_esp32", "data": {}}))
                except Exception:
                    pass

        self.assertTrue(
            saw_armed,
            "firmware never printed the 'i2s_rx armed' line — peripheral "
            "clock enable or initial register write failed",
        )
        self.assertIsNotNone(
            eof_ms,
            f"firmware never received in_suc_eof — DMA descriptor not "
            f"completed. Buffered serial: {buf[:400]}",
        )
        assert eof_ms is not None
        self.assertLess(
            eof_ms, 5000,
            f"EOF arrived after {eof_ms}ms — too slow, the QEMU model "
            f"should produce a frame within a few hundred ms",
        )
        self.assertIsNotNone(
            buf_dump,
            "firmware never dumped buf[0..31] — buffer write may have "
            "missed the descriptor's mapped region",
        )
        # Phase-2 idle pattern is 0xAA repeating. If the model fills 0x00
        # the descriptor was reached but no data was actually written.
        # Accept any pattern as long as it's 32 valid hex bytes — the
        # specific value is the host-side push problem that Phase 3
        # solves.
        bytes_seen = (buf_dump or "").split()
        self.assertEqual(
            len(bytes_seen), 32,
            f"expected 32 hex bytes in buf dump, got {len(bytes_seen)}: "
            f"{bytes_seen}",
        )


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