fds-os/rust/dasungd/tests/pty_smoke.py

#!/usr/bin/env python3
"""Exercise the real daemon/IPC/reconnect/persistence using a fake serial monitor.

No root access, physical display changes, USB claims or Python runtime dependency
in the shipped daemon. Run after cargo build --release.
"""
import errno
import json
import os
from pathlib import Path
import pty
import select
import signal
import socket
import subprocess
import tempfile
import threading
import time

ROOT = Path(__file__).resolve().parents[1]
BINARY = Path(os.environ.get("DASUNGD_BIN", ROOT.parents[1] / "target/x86_64-unknown-linux-gnu/release/dasungd"))


class Monitor:
    def __init__(self, link):
        self.master, slave = pty.openpty()
        self.values = {1: 4, 2: 1, 7: 2, 8: 0, 9: 40, 0x10: 0x31, 0x13: 2}
        self.keepalives = 0
        self.running = True
        self.respond = True
        link.unlink(missing_ok=True)
        link.symlink_to(os.ttyname(slave))
        os.close(slave)
        self.thread = threading.Thread(target=self.loop)
        self.thread.start()

    def loop(self):
        buffer = b""
        while self.running:
            try:
                if not select.select([self.master], [], [], 0.1)[0]:
                    continue
                data = os.read(self.master, 4096)
                buffer += data
                while b"5FF5" in buffer and len(buffer) >= 24:
                    buffer = buffer[buffer.index(b"5FF5"):]
                    if len(buffer) < 24:
                        break
                    frame, buffer = buffer[:24], buffer[24:]
                    if frame[-4:] != b"A0FA":
                        continue
                    cmd, opt = int(frame[4:6], 16), int(frame[6:8], 16)
                    if cmd == 0x20:
                        self.keepalives += opt == 1
                    elif cmd == 0x0A and self.respond:
                        reply = f"5FF5F00A{opt:02X}{self.values.get(opt,0):02X}000000A0FA".encode()
                        # Fragment actual 22-byte query responses across reads.
                        os.write(self.master, reply[:9])
                        time.sleep(0.005)
                        os.write(self.master, reply[9:])
                    elif cmd in self.values:
                        self.values[cmd] = opt
                        if self.respond:
                            os.write(self.master, frame)
            except OSError as exc:
                if exc.errno in (errno.EIO, errno.EBADF):
                    time.sleep(0.05)
                else:
                    raise

    def close(self):
        self.running = False
        self.thread.join(timeout=2)
        os.close(self.master)


def wait_for(predicate, timeout=12):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            value = predicate()
            if value:
                return value
        except (FileNotFoundError, ConnectionRefusedError, json.JSONDecodeError):
            pass
        time.sleep(0.1)
    raise AssertionError("condition did not become true")


with tempfile.TemporaryDirectory(prefix="dasungd-smoke-") as directory:
    d = Path(directory)
    sock = d / "control.sock"
    link = d / "uart"
    cfg = d / "config.toml"
    cfg.write_text(f'''monitor_serial = "TEST-NOT-A-PHYSICAL-MONITOR"
socket = "{sock}"
state_file = "{d / 'state.json'}"
transport = "serial"
serial_device = "{link}"
require_display = false
keepalive_ms = 500
reconnect_ms = 500
watchdog_ms = 20000
[display]
enabled = false
''')

    def request(value):
        with socket.socket(socket.AF_UNIX) as connection:
            connection.settimeout(5)
            connection.connect(str(sock))
            connection.sendall(json.dumps(value).encode() + b"\n")
            with connection.makefile("r") as reader:
                return json.loads(reader.readline())

    def status():
        return request({"op": "status"})["status"]

    log = (d / "daemon.log").open("w+")
    process = None
    monitor = None
    try:
        # Start with the monitor absent, then plug it in.
        process = subprocess.Popen([str(BINARY), "--config", str(cfg), "daemon"], stderr=log)
        wait_for(lambda: sock.exists())
        assert not status()["connected"]
        monitor = Monitor(link)
        wait_for(lambda: status()["parameters"].get("0x02") == 1)
        wait_for(lambda: monitor.keepalives >= 2)
        assert not request({"op": "set", "parameter": "mode", "value": 255})["ok"]
        assert request({"op": "set", "parameter": "contrast", "value": 5, "save": True})["ok"]
        wait_for(lambda: monitor.values[1] == 5)
        # USB/serial unplug, node replacement, automatic reinitialization.
        monitor.close()
        monitor = None
        wait_for(lambda: not status()["connected"])
        monitor = Monitor(link)
        wait_for(lambda: monitor.values[1] == 5)
        assert status()["reconnects"] >= 2
        wait_for(lambda: monitor.keepalives >= 2)
        # Only one daemon can own a given socket; the first remains reachable.
        duplicate = subprocess.run([str(BINARY), "--config", str(cfg), "daemon"], capture_output=True)
        assert duplicate.returncode != 0
        assert status()["connected"]
        # A control board that stops replying gets reopened by the watchdog.
        old_reconnects = status()["reconnects"]
        monitor.respond = False
        wait_for(lambda: status()["reconnects"] > old_reconnects, timeout=25)
        monitor.respond = True
        wait_for(lambda: status()["parameters"].get("0x02") == 1)
        # Restart the process and verify saved settings survive.
        process.send_signal(signal.SIGTERM)
        assert process.wait(timeout=5) == 0
        monitor.values[1] = 4
        process = subprocess.Popen([str(BINARY), "--config", str(cfg), "daemon"], stderr=log)
        wait_for(lambda: monitor.values[1] == 5)
        assert request({"op": "forget", "parameter": "contrast"})["ok"]
        assert json.loads((d / "state.json").read_text()) == {}
        print("PASS: late attach, keepalive, fragmented replies, IPC validation, saved settings, unplug/replug, duplicate exclusion, reply watchdog, restart and forget")
    finally:
        if process is not None and process.poll() is None:
            process.terminate()
            process.wait(timeout=5)
        if monitor is not None:
            monitor.close()
        log.flush()
        log.seek(0)
        print(log.read())