262 lines
8.5 KiB
Python
262 lines
8.5 KiB
Python
"""CH347 stream framing, byte for byte.
|
|
|
|
The wire format is the one thing here with no second source: it was reverse
|
|
engineered from the aystarik Linux driver, and a silent change to it would show
|
|
up as unexplained NACKs on real hardware rather than as an exception. So these
|
|
tests assert the exact bytes rather than "a write happened".
|
|
"""
|
|
|
|
import pytest
|
|
|
|
import ch347
|
|
|
|
ADDR = 0x23
|
|
ACK = ch347.ACK
|
|
|
|
|
|
@pytest.fixture
|
|
def bus(usb_device):
|
|
"""A CH347I2C whose set_speed handshake has already been accounted for."""
|
|
b = ch347.CH347I2C()
|
|
usb_device.writes.clear()
|
|
return b
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Construction
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_open_sets_the_requested_speed(usb_device):
|
|
ch347.CH347I2C(ch347.SPEED_400KHZ)
|
|
assert usb_device.writes == [
|
|
bytes([ch347.CMD_STREAM, ch347.CMD_SET | ch347.SPEED_400KHZ,
|
|
ch347.CMD_END])]
|
|
|
|
|
|
def test_open_defaults_to_the_documented_speed(usb_device):
|
|
ch347.CH347I2C()
|
|
expected = ch347.SPEEDS[ch347.DEFAULT_SPEED_KHZ]
|
|
assert usb_device.last_write[1] == ch347.CMD_SET | expected
|
|
|
|
|
|
def test_speeds_table_covers_every_speed_constant():
|
|
assert sorted(ch347.SPEEDS.values()) == [
|
|
ch347.SPEED_20KHZ, ch347.SPEED_100KHZ,
|
|
ch347.SPEED_400KHZ, ch347.SPEED_750KHZ]
|
|
assert ch347.DEFAULT_SPEED_KHZ in ch347.SPEEDS
|
|
|
|
|
|
def test_missing_adapter_names_the_device(monkeypatch):
|
|
import usb.core
|
|
monkeypatch.setattr(usb.core, "find", lambda **kwargs: None)
|
|
with pytest.raises(IOError, match="1a86:55db"):
|
|
ch347.CH347I2C()
|
|
|
|
|
|
def test_claim_failure_points_at_setup(monkeypatch, usb_device):
|
|
import usb.core
|
|
import usb.util
|
|
|
|
def refuse(*args):
|
|
raise usb.core.USBError("Access denied")
|
|
|
|
monkeypatch.setattr(usb.util, "claim_interface", refuse)
|
|
with pytest.raises(IOError, match="setup.sh"):
|
|
ch347.CH347I2C()
|
|
|
|
|
|
def test_set_speed_rejects_an_unknown_rate(bus):
|
|
with pytest.raises(ValueError, match="expected 0-3"):
|
|
bus.set_speed(9)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# write()
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_write_frames_a_register_write(bus, usb_device):
|
|
usb_device.queue([ACK, ACK, ACK])
|
|
bus.write(ADDR, [0x04, 0x64])
|
|
assert usb_device.last_write == bytes([
|
|
ch347.CMD_STREAM,
|
|
ch347.CMD_STA,
|
|
ch347.CMD_OUT | 3, # address plus two payload bytes
|
|
ADDR << 1, # write: LSB clear
|
|
0x04, 0x64,
|
|
ch347.CMD_STO,
|
|
ch347.CMD_END,
|
|
])
|
|
|
|
|
|
def test_write_raises_on_an_address_nack(bus, usb_device):
|
|
usb_device.queue([0, ACK])
|
|
with pytest.raises(IOError, match="No ACK from I2C address 0x23 on write"):
|
|
bus.write(ADDR, [0x04])
|
|
|
|
|
|
def test_write_raises_on_a_data_nack(bus, usb_device):
|
|
usb_device.queue([ACK, 0])
|
|
with pytest.raises(IOError, match="NACKed a data byte"):
|
|
bus.write(ADDR, [0x04])
|
|
|
|
|
|
def test_write_rejects_an_oversized_payload(bus):
|
|
with pytest.raises(ValueError, match="exceeds the CH347 limit"):
|
|
bus.write(ADDR, bytes(ch347.MAX_XFER))
|
|
|
|
|
|
def test_write_accepts_the_largest_legal_payload(bus, usb_device):
|
|
payload = bytes(ch347.MAX_XFER - 1)
|
|
usb_device.queue([ACK] * (len(payload) + 1))
|
|
bus.write(ADDR, payload)
|
|
# The count field is 6 bits and must not have overflowed into the opcode.
|
|
assert usb_device.last_write[2] == ch347.CMD_OUT | ch347.MAX_XFER
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# read()
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_read_acks_all_but_the_last_byte(bus, usb_device):
|
|
usb_device.queue([ACK, 0x11, 0x22, 0x33])
|
|
data, mono, wall = bus.read(ADDR, 3)
|
|
assert data == bytes([0x11, 0x22, 0x33])
|
|
assert usb_device.last_write == bytes([
|
|
ch347.CMD_STREAM,
|
|
ch347.CMD_STA,
|
|
ch347.CMD_OUT | 1,
|
|
(ADDR << 1) | 1, # read: LSB set
|
|
ch347.CMD_IN | 2, # first n-1 bytes, ACKed
|
|
ch347.CMD_IN, # final byte, NACKed to end the read
|
|
ch347.CMD_STO,
|
|
ch347.CMD_END,
|
|
])
|
|
assert wall > mono # epoch seconds against a monotonic count
|
|
|
|
|
|
def test_single_byte_read_omits_the_acked_run(bus, usb_device):
|
|
usb_device.queue([ACK, 0x22])
|
|
assert bus.read(ADDR, 1)[0] == b"\x22"
|
|
assert usb_device.last_write == bytes([
|
|
ch347.CMD_STREAM, ch347.CMD_STA, ch347.CMD_OUT | 1, (ADDR << 1) | 1,
|
|
ch347.CMD_IN, ch347.CMD_STO, ch347.CMD_END])
|
|
|
|
|
|
def test_read_raises_on_an_address_nack(bus, usb_device):
|
|
usb_device.queue([0, 0x00])
|
|
with pytest.raises(IOError, match="No ACK from I2C address 0x23 on read"):
|
|
bus.read(ADDR, 1)
|
|
|
|
|
|
@pytest.mark.parametrize("count", [0, -1, ch347.MAX_XFER + 1])
|
|
def test_read_rejects_an_illegal_length(bus, count):
|
|
with pytest.raises(ValueError, match="outside 1.."):
|
|
bus.read(ADDR, count)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# write_read() -- the repeated START that halves the round trips
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_write_read_uses_a_repeated_start(bus, usb_device):
|
|
usb_device.queue([ACK, ACK, ACK] + [0x00] * 9)
|
|
data, _, _ = bus.write_read(ADDR, [0x24], 9)
|
|
assert len(data) == 9
|
|
assert usb_device.last_write == bytes([
|
|
ch347.CMD_STREAM,
|
|
ch347.CMD_STA,
|
|
ch347.CMD_OUT | 2, # write address plus the register pointer
|
|
ADDR << 1,
|
|
0x24,
|
|
ch347.CMD_STA, # repeated START -- no STOP in between
|
|
ch347.CMD_OUT | 1,
|
|
(ADDR << 1) | 1,
|
|
ch347.CMD_IN | 8,
|
|
ch347.CMD_IN,
|
|
ch347.CMD_STO,
|
|
ch347.CMD_END,
|
|
])
|
|
|
|
|
|
def test_write_read_expects_one_ack_per_clocked_out_byte(bus, usb_device):
|
|
"""Two addresses plus the payload -- getting this wrong misaligns the data."""
|
|
usb_device.queue([ACK, ACK, ACK, ACK, 0xAA])
|
|
data, _, _ = bus.write_read(ADDR, [0x01, 0x02], 1)
|
|
assert data == b"\xaa"
|
|
|
|
|
|
def test_write_read_raises_when_any_phase_nacks(bus, usb_device):
|
|
usb_device.queue([ACK, ACK, 0, 0x00])
|
|
with pytest.raises(IOError, match="NACKed during combined transfer"):
|
|
bus.write_read(ADDR, [0x24], 1)
|
|
|
|
|
|
def test_write_read_requires_something_to_write(bus):
|
|
with pytest.raises(ValueError, match="at least one byte"):
|
|
bus.write_read(ADDR, b"", 1)
|
|
|
|
|
|
def test_write_read_validates_both_lengths(bus):
|
|
with pytest.raises(ValueError, match="exceeds the CH347 limit"):
|
|
bus.write_read(ADDR, bytes(ch347.MAX_XFER), 1)
|
|
with pytest.raises(ValueError, match="outside 1.."):
|
|
bus.write_read(ADDR, [0x24], ch347.MAX_XFER + 1)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# probe() / scan()
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_probe_reports_an_ack(bus, usb_device):
|
|
usb_device.queue([ACK])
|
|
assert bus.probe(ADDR) is True
|
|
assert usb_device.last_write == bytes([
|
|
ch347.CMD_STREAM, ch347.CMD_STA, ch347.CMD_OUT | 1, ADDR << 1,
|
|
ch347.CMD_STO, ch347.CMD_END])
|
|
|
|
|
|
def test_probe_treats_a_nack_as_absence_not_an_error(bus, usb_device):
|
|
usb_device.queue([0])
|
|
assert bus.probe(ADDR) is False
|
|
|
|
|
|
def test_probe_swallows_a_usb_error(bus, monkeypatch):
|
|
def boom(*args, **kwargs):
|
|
raise IOError("pipe error")
|
|
|
|
monkeypatch.setattr(bus, "_xfer", boom)
|
|
assert bus.probe(ADDR) is False
|
|
|
|
|
|
def test_scan_returns_only_responders(bus, usb_device):
|
|
# 0x08..0x0a: only the middle one ACKs.
|
|
usb_device.queue([0], [ACK], [0])
|
|
assert bus.scan(first=0x08, last=0x0a) == [0x09]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Transport errors
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_short_usb_write_is_an_error(bus, usb_device):
|
|
usb_device.short_write_by = 1
|
|
usb_device.queue([ACK, ACK])
|
|
with pytest.raises(IOError, match="Short USB write"):
|
|
bus.write(ADDR, [0x04])
|
|
|
|
|
|
def test_short_usb_read_is_an_error(bus, usb_device):
|
|
usb_device.queue([ACK]) # one byte where two were expected
|
|
with pytest.raises(IOError, match="Short USB read"):
|
|
bus.write(ADDR, [0x04])
|
|
|
|
|
|
def test_close_is_idempotent(bus):
|
|
bus.close()
|
|
bus.close()
|
|
|
|
|
|
def test_context_manager_releases_the_interface(usb_device):
|
|
with ch347.CH347I2C() as b:
|
|
assert b._claimed
|
|
assert not b._claimed
|