205 lines
7.6 KiB
Python
205 lines
7.6 KiB
Python
"""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]
|