rdm-card-reader/tests/test_e2e_led.py

"""End-to-end: a real card read must drive the LED on and off.

Uses a pseudo-terminal for the reader and a fake pin for the LED, so the whole
chain runs — serial frame -> card event -> LED worker -> pin state — without a
Pi and without GPIO hardware.
"""

import os
import pty
import struct
import subprocess
import sys
import threading
import time

HERE = os.path.dirname(os.path.abspath(__file__))
PROJECT = os.path.dirname(HERE)
VENV_PY = os.path.join(PROJECT, ".venv", "bin", "python")
PYTHON = VENV_PY if os.path.exists(VENV_PY) else sys.executable

CARD_VALUE = 0x00C4A0B5

# Runs the app's own CLI with gpiozero's LED replaced by a recorder, so we can
# observe what the app would have done to the pin.
DRIVER = r'''
import os, sys, threading, time, json

sys.path.insert(0, os.environ["PROJECT_SRC"])

from rdm_card_reader import led as led_module

TRANSITIONS = []
_LOCK = threading.Lock()


class RecordingLED:
    def __init__(self, pin=None, active_high=True):
        self.pin = pin
        self._state = False

    def _record(self, value):
        self._state = value
        with _LOCK:
            TRANSITIONS.append(value)
        # Write through immediately: the test terminates this process, so
        # anything buffered for the end would be lost.
        sys.stderr.write("TRANSITION=" + value + "\n")
        sys.stderr.flush()

    def on(self):
        self._record("on")

    def off(self):
        self._record("off")

    def close(self):
        pass


class FakeGPIOLED(led_module.GPIOLED):
    def __init__(self, pin, active_high=True):
        self.pin = pin
        self._led = RecordingLED(pin, active_high)
        self._is_on = False

    def on(self):
        self._led.on()
        self._is_on = True

    def off(self):
        self._led.off()
        self._is_on = False

    def close(self):
        self._led.close()


led_module.GPIOLED = FakeGPIOLED

from rdm_card_reader.cli import main

code = main(["--port", os.environ["SERIAL_PORT"], "--heartbeat", "0.3",
             "--led-mode", os.environ.get("LED_MODE", "error"), "--json"])
sys.exit(code)
'''


def em4100_frame(value, card_type=0x0F):
    id_bytes = struct.pack(">I", value)
    checksum = card_type
    for b in id_bytes:
        checksum ^= b
    frame = (bytes([card_type]) + id_bytes + bytes([checksum])).hex().upper()
    assert len(frame) == 12, frame
    return frame.encode("ascii")


def pump(stream, sink):
    """Read a stream on its own thread.

    Text-mode pipes buffer, so select() would not report this data.
    """
    for line in stream:
        sink.append(line.rstrip())


def run_case(mode):
    """Drive a card through the app and return (events, led transitions)."""
    master, slave = pty.openpty()
    serial_port = os.ttyname(slave)

    env = dict(os.environ)
    env["SERIAL_PORT"] = serial_port
    env["PROJECT_SRC"] = os.path.join(PROJECT, "src")
    env["LED_MODE"] = mode
    env.pop("PYTHONPATH", None)

    proc = subprocess.Popen(
        [PYTHON, "-u", "-c", DRIVER],
        cwd=PROJECT, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        text=True, bufsize=1,
    )

    stdout, stderr = [], []
    t1 = threading.Thread(target=pump, args=(proc.stdout, stdout), daemon=True)
    t2 = threading.Thread(target=pump, args=(proc.stderr, stderr), daemon=True)
    t1.start()
    t2.start()

    frame = em4100_frame(CARD_VALUE)
    try:
        time.sleep(2.0)                      # let the reader open the port
        for _ in range(6):                   # card present
            os.write(master, b"\x02" + frame + b"\x03")
            time.sleep(0.1)
        time.sleep(1.2)                      # stop writing -> card removed
    finally:
        try:
            proc.terminate()
        except Exception:
            pass
        try:
            proc.wait(timeout=5)
        except Exception:
            proc.kill()
        for t in (t1, t2):
            t.join(timeout=3)
        try:
            os.close(master)
            os.close(slave)
        except Exception:
            pass

    transitions = [
        line.split("=", 1)[1]
        for line in stderr
        if line.startswith("TRANSITION=")
    ]
    return stdout, stderr, transitions


def main():
    failures = 0

    # --- error mode: lit while running, dark while a card is read ----------
    stdout, stderr, transitions = run_case("error")
    print("======== error mode ========")
    for line in stdout[:2]:
        print("   " + line)
    print("   events:      {}".format(
        [l.split('"event": "')[1].split('"')[0]
         for l in stdout if '"event"' in l]))
    print("   transitions: {}".format(transitions))

    checks = {
        "card was read":     any("card_inserted" in l for l in stdout),
        "card removal seen": any("card_removed" in l for l in stdout),
        "lit at startup":    transitions[:1] == ["on"],
        "dark while reading": "off" in transitions,
        "lit again after":   transitions[-1:] == ["on"],
    }

    # --- presence mode: dark until a card arrives -------------------------
    stdout2, stderr2, transitions2 = run_case("presence")
    print("\n======== presence mode ========")
    print("   transitions: {}".format(transitions2))
    checks.update({
        "presence: read still works":
            any("card_inserted" in l for l in stdout2),
        "presence: dark at startup": transitions2[:1] == ["on"] or not transitions2,
        "presence: lit on the card": "on" in transitions2,
        "presence: dark after":     transitions2[-1:] == ["off"],
    })

    print()
    ok = True
    for name, passed in checks.items():
        print("  [{}] {}".format("PASS" if passed else "FAIL", name))
        ok = ok and bool(passed)

    if not ok:
        failures = 1
        print("\n--- driver stderr (error mode) ---")
        print("\n".join(stderr)[:1200])
    return failures


if __name__ == "__main__":
    sys.exit(main())