aima_phone_catcher/tests/fakes.py

from __future__ import annotations

import threading
import time

from lingkong_motor_service.protocol import CanFrame


class FakeTransport:
    def __init__(self) -> None:
        self.frames: list[CanFrame] = []
        self.frame_times_ns: list[int] = []
        self.tx_queue_timeouts_s: list[float | None] = []
        self.is_open = True
        self.fail_next = False
        self.received_frames: list[CanFrame] = []
        self.lock = threading.Lock()

    def send(self, frame: CanFrame, *, tx_queue_timeout_s: float | None = None) -> None:
        with self.lock:
            if self.fail_next:
                self.fail_next = False
                from lingkong_motor_service.transport import CanTransportError

                raise CanTransportError("injected CAN failure")
            self.frames.append(frame)
            self.frame_times_ns.append(time.monotonic_ns())
            self.tx_queue_timeouts_s.append(tx_queue_timeout_s)

    def receive(self, *, timeout_s: float = 0.0) -> CanFrame | None:
        del timeout_s
        with self.lock:
            if not self.received_frames:
                return None
            return self.received_frames.pop(0)

    def queue_received(self, frame: CanFrame) -> None:
        with self.lock:
            self.received_frames.append(frame)

    def shutdown(self) -> None:
        with self.lock:
            self.is_open = False