"""Fakes standing in for the hardware, so the whole suite runs unplugged. Three levels are faked, matching the three seams in the code: FakeUsbDevice a CH347 at the libusb boundary -- ch347.py's own framing is then the thing under test, byte for byte. FakeBus an I2C master at the rm3100.py boundary, holding a register map, so the driver's register sequences are under test. FakeSensor an RM3100 at the logger.py boundary, replaying a scripted timeline of DRDY transitions, so the miss-counting maths is under test without waiting on real time. """ import sys from pathlib import Path import pytest # The modules under test sit at the repo root, next to this directory. sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) class FakeUsbDevice: """Records bulk writes and replays queued bulk reads.""" def __init__(self): self.writes = [] # bytes sent to EP_OUT, in order self.replies = [] # bytes to hand back, one per read() self.short_write_by = 0 # subtract this from the reported write length def queue(self, *replies): """Queue one reply per expected read, as bytes or an iterable of ints.""" self.replies.extend(bytes(r) for r in replies) return self def write(self, endpoint, data, timeout): self.writes.append(bytes(data)) return len(data) - self.short_write_by def read(self, endpoint, length, timeout): if not self.replies: raise AssertionError( f"unexpected read of {length} bytes: nothing queued") return bytearray(self.replies.pop(0)) @property def last_write(self): return self.writes[-1] @pytest.fixture def usb_device(monkeypatch): """A CH347 that never touches USB. Yields the device; build the bus with it.""" import usb.core import usb.util device = FakeUsbDevice() monkeypatch.setattr(usb.core, "find", lambda **kwargs: device) monkeypatch.setattr(usb.util, "claim_interface", lambda *a: None) monkeypatch.setattr(usb.util, "release_interface", lambda *a: None) monkeypatch.setattr(usb.util, "dispose_resources", lambda *a: None) return device class FakeBus: """An I2C bus holding one device's register map. Implements write/read/write_read with the same (data, mono, wall) contract as ch347.CH347I2C, so rm3100.RM3100 cannot tell the difference. """ def __init__(self, registers=None, address=0x23, combined=True): self.address = address self.registers = dict(registers or {}) self.writes = [] # (addr, bytes) of every write self.pointer = 0 self.clock = 1000.0 if not combined: # Some buses cannot do a repeated START; the driver has a fallback # path for them and it needs exercising too. read_reg() looks the # method up with getattr(..., None), so hiding it this way is the # same to the driver as never having had one. self.write_read = None def _stamp(self): self.clock += 0.001 return self.clock, self.clock + 1_700_000_000.0 def write(self, addr, data): data = bytes(data) self.writes.append((addr, data)) self.pointer = data[0] for offset, value in enumerate(data[1:]): self.registers[data[0] + offset] = value def read(self, addr, count): mono, wall = self._stamp() data = bytes(self.registers.get(self.pointer + i, 0) for i in range(count)) return data, mono, wall def write_read(self, addr, data, count): self.write(addr, data) return self.read(addr, count) class FakeSensor: """An RM3100 replaying a scripted DRDY timeline against a fake clock. `events` is a list of (monotonic_time, counts) pairs: the instants at which a measurement completes and what it reads. poll_ready() reports DRDY set once the clock has passed the next event and the previous result was consumed. Every poll and read advances the clock by a fixed cost, so a test states its timeline in seconds and gets deterministic brackets out. """ POLL_COST = 0.0003 READ_COST = 0.0005 class Exhausted(Exception): """Raised once max_polls is reached, to stop an unbounded run.""" def __init__(self, events, poll_cost=POLL_COST, read_cost=READ_COST, stalls=None, max_polls=None): self.events = list(events) self.poll_cost = poll_cost self.read_cost = read_cost # [(at_time, extra_seconds), ...] -- the first poll starting at or after # at_time pays extra_seconds before returning. A host stall is the only # thing that actually loses a measurement on this rig, and stating it in # seconds rather than in poll counts keeps a test independent of how # many times the loop happens to poll. self.stalls = sorted(stalls or []) # A run with no --duration never returns on its own, so a test of that # path needs the sensor to be the thing that stops it. self.max_polls = max_polls self.now = 0.0 self.pending = None # counts of a completed, unread measurement self.next_event = 0 self.polls = 0 self.reads = [] # -- clock plumbing ------------------------------------------------- def monotonic(self): return self.now def wall(self): return self.now + 1_700_000_000.0 def _advance(self, dt): self.now += dt while (self.next_event < len(self.events) and self.events[self.next_event][0] <= self.now): # A completion overwrites any unread one: the chip does not queue, # which is exactly why an unnoticed miss loses a measurement. self.pending = self.events[self.next_event][1] self.next_event += 1 # -- the RM3100 surface logger.py uses ------------------------------ def poll_ready(self): self.polls += 1 if self.max_polls is not None and self.polls > self.max_polls: raise self.Exhausted(f"stopped after {self.max_polls} polls") stall = 0.0 if self.stalls and self.now >= self.stalls[0][0]: stall = self.stalls.pop(0)[1] self._advance(self.poll_cost + stall) return self.pending is not None, self.now, self.wall() def read_raw(self): self._advance(self.read_cost) counts = self.pending if self.pending is not None else (0, 0, 0) self.pending = None self.reads.append(counts) return counts, self.now, self.wall() @pytest.fixture def fake_clock(monkeypatch): """Point time.monotonic and time.time at a FakeSensor's clock. sample_loop() binds `monotonic = time.monotonic` when it runs, not at import, so patching the module attribute is enough to control it. """ import time def install(sensor): monkeypatch.setattr(time, "monotonic", sensor.monotonic) monkeypatch.setattr(time, "time", sensor.wall) return sensor return install def grid_events(period, count, start=None, counts=(100, 200, 300), skip=()): """Measurement completions on a uniform grid, optionally dropping some. A skipped index is a completion the chip still makes -- the grid never pauses -- so it is present here; a test forces a *miss* by making the host too slow to read it, not by removing it. """ start = period if start is None else start return [(start + i * period, (counts[0] + i, counts[1], counts[2])) for i in range(count) if i not in skip]