512 lines
23 KiB
Python
512 lines
23 KiB
Python
"""
|
||
Tests for ESP32 I2C slave state machines.
|
||
|
||
Covers BMP280Slave, DS1307Slave, DS3231Slave, I2CWriteSink, and MPU6050Slave
|
||
from app/services/esp32_i2c_slaves.py.
|
||
|
||
Correct picsimlab I2C event encoding (from hw/i2c/picsimlab_i2c.c + QEMU i2c.h):
|
||
event & 0xFF = operation type:
|
||
0x00 = I2C_START_RECV — firmware called requestFrom (read direction START)
|
||
0x01 = I2C_START_SEND — firmware called beginTransmission (write direction START)
|
||
0x03 = I2C_FINISH — end of transaction (STOP or RSTART between write+read)
|
||
0x05 = WRITE byte — (event >> 8) & 0xFF is the data byte
|
||
0x06 = READ byte — return value is the data byte to deliver to firmware
|
||
|
||
ACK convention: return 0 = ACK (success), non-zero = NACK.
|
||
For READ events: return value is the data byte sent to the firmware.
|
||
|
||
Run from the backend/ directory:
|
||
python test_esp32_i2c_slaves.py
|
||
"""
|
||
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
# Ensure backend/ is importable (for direct execution; pytest uses conftest.py)
|
||
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / 'backend'))
|
||
|
||
from app.services.esp32_i2c_slaves import (
|
||
BMP280Slave,
|
||
DS1307Slave,
|
||
DS3231Slave,
|
||
I2CWriteSink,
|
||
MPU6050Slave,
|
||
I2C_START_RECV,
|
||
I2C_START_SEND,
|
||
I2C_FINISH,
|
||
I2C_WRITE,
|
||
I2C_READ,
|
||
)
|
||
|
||
|
||
# ── I2C protocol helpers ──────────────────────────────────────────────────────
|
||
|
||
def i2c_write(byte: int) -> int:
|
||
"""WRITE event: data in high byte, op 0x05 in low byte."""
|
||
return ((byte & 0xFF) << 8) | I2C_WRITE
|
||
|
||
|
||
def i2c_read_seq(slave, reg: int, n: int) -> list[int]:
|
||
"""Simulate write-then-read: write register address, then read n bytes.
|
||
|
||
Models the Adafruit BusIO write_then_read pattern:
|
||
beginTransmission → write(reg) → endTransmission(false) → requestFrom → read()*n
|
||
"""
|
||
slave.handle_event(I2C_START_SEND) # write direction START
|
||
slave.handle_event(i2c_write(reg)) # set register pointer
|
||
slave.handle_event(I2C_FINISH) # RSTART (repeated start before read phase)
|
||
slave.handle_event(I2C_START_RECV) # read direction START
|
||
data = [slave.handle_event(I2C_READ) for _ in range(n)]
|
||
slave.handle_event(I2C_FINISH) # STOP
|
||
return data
|
||
|
||
|
||
def read_u16_le(regs: bytearray, addr: int) -> int:
|
||
"""Read unsigned 16-bit little-endian from register array."""
|
||
return regs[addr] | (regs[addr + 1] << 8)
|
||
|
||
|
||
def bcd_valid(value: int) -> bool:
|
||
"""Return True if both nibbles of a BCD byte are in 0–9."""
|
||
return (value >> 4) <= 9 and (value & 0xF) <= 9
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# BMP280 Slave Tests
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestBMP280Slave(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.slave = BMP280Slave()
|
||
|
||
# ── I2C protocol ───────────────────────────────────────────────────────────
|
||
|
||
def test_ack_on_start_send(self):
|
||
result = self.slave.handle_event(I2C_START_SEND)
|
||
self.assertEqual(result, 0, 'START_SEND must return 0 (ACK)')
|
||
|
||
def test_ack_on_start_recv(self):
|
||
result = self.slave.handle_event(I2C_START_RECV)
|
||
self.assertEqual(result, 0, 'START_RECV must return 0 (ACK)')
|
||
|
||
def test_ack_on_write(self):
|
||
self.slave.handle_event(I2C_START_SEND)
|
||
result = self.slave.handle_event(i2c_write(0xD0))
|
||
self.assertEqual(result, 0, 'WRITE must return 0 (ACK)')
|
||
|
||
def test_finish_returns_zero(self):
|
||
self.slave.handle_event(I2C_START_SEND)
|
||
result = self.slave.handle_event(I2C_FINISH)
|
||
self.assertEqual(result, 0)
|
||
|
||
# ── Chip identity ──────────────────────────────────────────────────────────
|
||
|
||
def test_chip_id_register_0xd0(self):
|
||
chip_id = i2c_read_seq(self.slave, 0xD0, 1)[0]
|
||
self.assertEqual(chip_id, 0x58, 'chip_id must be 0x58 for BMP280')
|
||
|
||
# ── Calibration registers ──────────────────────────────────────────────────
|
||
|
||
def test_calibration_t1_little_endian(self):
|
||
# DIG_T1=27504 stored at 0x88 (LSB) and 0x89 (MSB)
|
||
t1 = read_u16_le(self.slave.regs, 0x88)
|
||
self.assertEqual(t1, BMP280Slave.DIG_T1)
|
||
|
||
def test_calibration_p1_little_endian(self):
|
||
p1 = read_u16_le(self.slave.regs, 0x8E)
|
||
self.assertEqual(p1, BMP280Slave.DIG_P1)
|
||
|
||
# ── Default measurement values ─────────────────────────────────────────────
|
||
|
||
def test_default_temp_25c(self):
|
||
"""Default state: 25°C → compensated centidegrees ≈ 2500."""
|
||
adc_t_bytes = i2c_read_seq(self.slave, 0xFA, 3)
|
||
adc_t = ((adc_t_bytes[0] << 12) | (adc_t_bytes[1] << 4) | (adc_t_bytes[2] >> 4))
|
||
compensated = self.slave._compensate_t(adc_t)
|
||
# Allow ±1 centidegree tolerance (binary-search rounding)
|
||
self.assertAlmostEqual(compensated, 2500, delta=1)
|
||
|
||
def test_default_pressure_1013hpa(self):
|
||
"""Default state: 1013.25 hPa → compensated Pa within 100 Pa tolerance."""
|
||
adc_t_bytes = i2c_read_seq(self.slave, 0xFA, 3)
|
||
adc_t = ((adc_t_bytes[0] << 12) | (adc_t_bytes[1] << 4) | (adc_t_bytes[2] >> 4))
|
||
adc_p_bytes = i2c_read_seq(self.slave, 0xF7, 3)
|
||
adc_p = ((adc_p_bytes[0] << 12) | (adc_p_bytes[1] << 4) | (adc_p_bytes[2] >> 4))
|
||
compensated_pa = self.slave._compensate_p(adc_p, adc_t)
|
||
target_pa = 1013.25 * 100.0
|
||
self.assertAlmostEqual(compensated_pa, target_pa, delta=100) # ±1 hPa
|
||
|
||
# ── update() changes ADC registers ────────────────────────────────────────
|
||
|
||
def test_update_changes_temp_regs(self):
|
||
before = list(self.slave.regs[0xFA:0xFD])
|
||
self.slave.update(30.0, 1013.25)
|
||
after = list(self.slave.regs[0xFA:0xFD])
|
||
self.assertNotEqual(before, after, 'Temp ADC regs must change after update(30.0,...)')
|
||
|
||
def test_update_changes_pressure_regs(self):
|
||
before = list(self.slave.regs[0xF7:0xFA])
|
||
self.slave.update(25.0, 900.0)
|
||
after = list(self.slave.regs[0xF7:0xFA])
|
||
self.assertNotEqual(before, after, 'Pressure ADC regs must change after update(...,900.0)')
|
||
|
||
def test_update_temp_compensates_correctly(self):
|
||
self.slave.update(40.0, 1013.25)
|
||
adc_t_bytes = i2c_read_seq(self.slave, 0xFA, 3)
|
||
adc_t = ((adc_t_bytes[0] << 12) | (adc_t_bytes[1] << 4) | (adc_t_bytes[2] >> 4))
|
||
compensated = self.slave._compensate_t(adc_t)
|
||
self.assertAlmostEqual(compensated, 4000, delta=1)
|
||
|
||
# ── Sequential register reads ──────────────────────────────────────────────
|
||
|
||
def test_sequential_read_advances_pointer(self):
|
||
"""Reading 3 bytes from 0xF7 must yield distinct pressure MSB/LSB/XLSB."""
|
||
bytes_ = i2c_read_seq(self.slave, 0xF7, 3)
|
||
self.assertEqual(len(bytes_), 3)
|
||
# At least two of the three bytes must differ (non-trivial measurement)
|
||
self.assertFalse(bytes_[0] == bytes_[1] == bytes_[2],
|
||
'All three pressure ADC bytes should not be identical')
|
||
|
||
# ── State machine ─────────────────────────────────────────────────────────
|
||
|
||
def test_write_sets_register_ptr(self):
|
||
"""Write-then-read: write reg address 0xD0, then read returns chip_id."""
|
||
self.slave.handle_event(I2C_START_SEND)
|
||
self.slave.handle_event(i2c_write(0xD0)) # set ptr to chip_id reg
|
||
self.slave.handle_event(I2C_FINISH)
|
||
self.slave.handle_event(I2C_START_RECV)
|
||
val = self.slave.handle_event(I2C_READ)
|
||
self.assertEqual(val, 0x58)
|
||
|
||
def test_finish_resets_first_byte_flag(self):
|
||
"""After FINISH, the next transaction's first WRITE must set reg_ptr."""
|
||
self.slave.handle_event(I2C_START_SEND)
|
||
self.slave.handle_event(i2c_write(0xD0))
|
||
self.slave.handle_event(I2C_FINISH)
|
||
# New transaction: WRITE 0xD0 again → should set reg_ptr, not write data
|
||
self.slave.handle_event(I2C_START_SEND)
|
||
self.slave.handle_event(i2c_write(0xD0))
|
||
self.slave.handle_event(I2C_FINISH)
|
||
self.slave.handle_event(I2C_START_RECV)
|
||
val = self.slave.handle_event(I2C_READ)
|
||
self.assertEqual(val, 0x58, 'chip_id should still be 0x58 after FINISH + new transaction')
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# DS1307 Slave Tests
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestDS1307Slave(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.slave = DS1307Slave()
|
||
|
||
def test_ack_on_start_send(self):
|
||
self.assertEqual(self.slave.handle_event(I2C_START_SEND), 0)
|
||
|
||
def test_ack_on_write(self):
|
||
self.slave.handle_event(I2C_START_SEND)
|
||
self.assertEqual(self.slave.handle_event(i2c_write(0x00)), 0)
|
||
|
||
def test_seconds_is_valid_bcd(self):
|
||
seconds = i2c_read_seq(self.slave, 0x00, 1)[0]
|
||
tens = (seconds >> 4) & 0xF
|
||
units = seconds & 0xF
|
||
self.assertLessEqual(tens, 5, 'Seconds tens digit must be ≤ 5')
|
||
self.assertLessEqual(units, 9, 'Seconds units digit must be ≤ 9')
|
||
|
||
def test_minutes_is_valid_bcd(self):
|
||
minutes = i2c_read_seq(self.slave, 0x01, 1)[0]
|
||
self.assertLessEqual((minutes >> 4) & 0xF, 5)
|
||
self.assertLessEqual( minutes & 0xF, 9)
|
||
|
||
def test_hours_is_valid_bcd(self):
|
||
hours = i2c_read_seq(self.slave, 0x02, 1)[0]
|
||
# BCD 00–23
|
||
tens = (hours >> 4) & 0xF
|
||
units = hours & 0xF
|
||
value = tens * 10 + units
|
||
self.assertGreaterEqual(value, 0)
|
||
self.assertLessEqual(value, 23)
|
||
|
||
def test_year_is_valid_bcd(self):
|
||
year = i2c_read_seq(self.slave, 0x06, 1)[0]
|
||
self.assertTrue(bcd_valid(year), f'Year 0x{year:02X} is not valid BCD')
|
||
|
||
def test_sequential_read_7_regs(self):
|
||
"""Read all 7 time registers in one transaction — all must be valid BCD."""
|
||
data = i2c_read_seq(self.slave, 0x00, 7)
|
||
self.assertEqual(len(data), 7)
|
||
for i, b in enumerate(data):
|
||
self.assertTrue(bcd_valid(b), f'Register {i} value 0x{b:02X} is not valid BCD')
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# DS3231 Slave Tests
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestDS3231Slave(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.slave = DS3231Slave()
|
||
|
||
def test_inherits_time_registers(self):
|
||
"""DS3231 regs 0–6 must return the same valid BCD as DS1307."""
|
||
data = i2c_read_seq(self.slave, 0x00, 7)
|
||
self.assertEqual(len(data), 7)
|
||
for i, b in enumerate(data):
|
||
self.assertTrue(bcd_valid(b), f'Register {i} value 0x{b:02X} is not valid BCD')
|
||
|
||
def test_control_register_0x0e(self):
|
||
val = self.slave._read_reg(0x0E)
|
||
self.assertEqual(val, 0x00)
|
||
|
||
def test_status_register_0x0f(self):
|
||
val = self.slave._read_reg(0x0F)
|
||
self.assertEqual(val, 0x00)
|
||
|
||
def test_temperature_msb_default_25c(self):
|
||
val = self.slave._read_reg(0x11)
|
||
self.assertEqual(val, 25, 'Temp MSB must be 25 (integer °C) at default 25.0°C')
|
||
|
||
def test_temperature_lsb_zero_frac(self):
|
||
val = self.slave._read_reg(0x12)
|
||
self.assertEqual(val, 0, 'Temp LSB must be 0 when fractional part is .00')
|
||
|
||
def test_temperature_update_quarter_degree(self):
|
||
"""25.25°C → LSB bits 7:6 == 0b01 (0.25°C step)."""
|
||
self.slave.temperatureC = 25.25
|
||
msb = self.slave._read_reg(0x11)
|
||
lsb = self.slave._read_reg(0x12)
|
||
self.assertEqual(msb, 25)
|
||
frac_bits = (lsb >> 6) & 0x03
|
||
self.assertEqual(frac_bits, 1, '0.25°C → bits 7:6 must be 0b01')
|
||
|
||
def test_temperature_update_half_degree(self):
|
||
"""25.5°C → LSB bits 7:6 == 0b10 (0.50°C step)."""
|
||
self.slave.temperatureC = 25.5
|
||
lsb = self.slave._read_reg(0x12)
|
||
frac_bits = (lsb >> 6) & 0x03
|
||
self.assertEqual(frac_bits, 2, '0.50°C → bits 7:6 must be 0b10')
|
||
|
||
def test_temperature_read_via_i2c(self):
|
||
"""Temperature registers are accessible via I2C READ from 0x11."""
|
||
self.slave.temperatureC = 30.0
|
||
data = i2c_read_seq(self.slave, 0x11, 2)
|
||
self.assertEqual(data[0], 30) # MSB = integer °C
|
||
self.assertEqual(data[1], 0) # LSB = 0 (no fractional part)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# I2CWriteSink Tests
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestI2CWriteSink(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.emitted: list[dict] = []
|
||
self.sink = I2CWriteSink(addr=0x3C, emit_fn=self.emitted.append)
|
||
|
||
def test_ack_on_start_send(self):
|
||
self.assertEqual(self.sink.handle_event(I2C_START_SEND), 0)
|
||
|
||
def test_ack_on_start_recv(self):
|
||
self.assertEqual(self.sink.handle_event(I2C_START_RECV), 0)
|
||
|
||
def test_ack_on_write(self):
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
for byte in [0x00, 0x21, 0xAB]:
|
||
result = self.sink.handle_event(i2c_write(byte))
|
||
self.assertEqual(result, 0, f'WRITE 0x{byte:02X} must return 0 (ACK)')
|
||
|
||
def test_read_returns_0xff(self):
|
||
"""Write-only device: READ must return 0xFF (no data to send)."""
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
result = self.sink.handle_event(I2C_READ)
|
||
self.assertEqual(result, 0xFF)
|
||
|
||
def test_emits_on_finish(self):
|
||
"""After 3 writes + FINISH, emit_fn must be called exactly once."""
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
for b in [0x00, 0x21, 0x7F]:
|
||
self.sink.handle_event(i2c_write(b))
|
||
self.sink.handle_event(I2C_FINISH)
|
||
self.assertEqual(len(self.emitted), 1, 'emit_fn must be called once on FINISH')
|
||
|
||
def test_emit_payload_addr(self):
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
self.sink.handle_event(i2c_write(0xAB))
|
||
self.sink.handle_event(I2C_FINISH)
|
||
self.assertEqual(self.emitted[0]['addr'], 0x3C)
|
||
|
||
def test_emit_payload_type(self):
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
self.sink.handle_event(i2c_write(0xAB))
|
||
self.sink.handle_event(I2C_FINISH)
|
||
self.assertEqual(self.emitted[0]['type'], 'i2c_transaction')
|
||
|
||
def test_emit_payload_data(self):
|
||
bytes_ = [0x00, 0x21, 0x7F]
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
for b in bytes_:
|
||
self.sink.handle_event(i2c_write(b))
|
||
self.sink.handle_event(I2C_FINISH)
|
||
self.assertEqual(self.emitted[0]['data'], bytes_)
|
||
|
||
def test_no_emit_for_empty_buffer(self):
|
||
"""START then immediate FINISH with no writes must NOT emit."""
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
self.sink.handle_event(I2C_FINISH)
|
||
self.assertEqual(len(self.emitted), 0, 'Empty buffer must not trigger emit_fn')
|
||
|
||
def test_resets_buffer_after_emit(self):
|
||
"""Second transaction accumulates fresh bytes, not leftover from first."""
|
||
# First transaction: writes [0xAA]
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
self.sink.handle_event(i2c_write(0xAA))
|
||
self.sink.handle_event(I2C_FINISH)
|
||
# Second transaction: writes [0xBB]
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
self.sink.handle_event(i2c_write(0xBB))
|
||
self.sink.handle_event(I2C_FINISH)
|
||
self.assertEqual(len(self.emitted), 2)
|
||
self.assertEqual(self.emitted[0]['data'], [0xAA])
|
||
self.assertEqual(self.emitted[1]['data'], [0xBB])
|
||
|
||
def test_custom_addr_forwarded(self):
|
||
"""Sink created with addr=0x27 emits with that addr."""
|
||
sink = I2CWriteSink(addr=0x27, emit_fn=self.emitted.append)
|
||
sink.handle_event(I2C_START_SEND)
|
||
sink.handle_event(i2c_write(0x38))
|
||
sink.handle_event(I2C_FINISH)
|
||
self.assertEqual(self.emitted[0]['addr'], 0x27)
|
||
|
||
def test_finish_return_value(self):
|
||
"""FINISH always returns 0."""
|
||
self.sink.handle_event(I2C_START_SEND)
|
||
result = self.sink.handle_event(I2C_FINISH)
|
||
self.assertEqual(result, 0)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# MPU6050 Slave Tests
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestMPU6050Slave(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.mpu = MPU6050Slave()
|
||
|
||
def test_who_am_i_default_addr(self):
|
||
"""WHO_AM_I register (0x75) must return 0x68."""
|
||
result = i2c_read_seq(self.mpu, 0x75, 1)
|
||
self.assertEqual(result[0], 0x68)
|
||
|
||
def test_start_send_ack(self):
|
||
"""START_SEND event must return 0 (ACK)."""
|
||
self.assertEqual(self.mpu.handle_event(I2C_START_SEND), 0)
|
||
|
||
def test_start_recv_ack(self):
|
||
"""START_RECV event must return 0 (ACK)."""
|
||
self.assertEqual(self.mpu.handle_event(I2C_START_RECV), 0)
|
||
|
||
def test_write_then_read_who_am_i(self):
|
||
"""Full write-then-read: write reg 0x75, then read returns 0x68."""
|
||
m = MPU6050Slave()
|
||
m.handle_event(I2C_START_SEND) # beginTransmission
|
||
m.handle_event(i2c_write(0x75)) # write register address
|
||
m.handle_event(I2C_FINISH) # RSTART
|
||
m.handle_event(I2C_START_RECV) # requestFrom
|
||
result = m.handle_event(I2C_READ) # read byte
|
||
self.assertEqual(result, 0x68, f"WHO_AM_I must be 0x68, got 0x{result:02x}")
|
||
|
||
def test_detected_pattern(self):
|
||
"""detected() pattern: START_SEND then FINISH (no data bytes) returns ACK."""
|
||
m = MPU6050Slave()
|
||
# beginTransmission + endTransmission (no data)
|
||
r1 = m.handle_event(I2C_START_SEND)
|
||
self.assertEqual(r1, 0, "START_SEND must return 0 (ACK) for detected()")
|
||
r2 = m.handle_event(I2C_FINISH)
|
||
self.assertEqual(r2, 0, "FINISH must return 0")
|
||
|
||
def test_accel_z_default_1g(self):
|
||
"""ACCEL_Z should default to +1g = 0x4000 (MSB=0x40, LSB=0x00)."""
|
||
result = i2c_read_seq(self.mpu, 0x3F, 2)
|
||
accel_z = (result[0] << 8) | result[1]
|
||
self.assertEqual(accel_z, 0x4000, f"Expected ACCEL_Z=0x4000, got 0x{accel_z:04x}")
|
||
|
||
def test_update_accel(self):
|
||
"""update() must reflect new accel values in register reads."""
|
||
self.mpu.update(accel_x=1.0, accel_y=0.0, accel_z=0.0)
|
||
result = i2c_read_seq(self.mpu, 0x3B, 2)
|
||
accel_x = (result[0] << 8) | result[1]
|
||
# 1g at ±2g full-scale = 16384 = 0x4000
|
||
self.assertEqual(accel_x, 0x4000)
|
||
|
||
def test_sequential_read_14_bytes(self):
|
||
"""getEvent() reads 14 bytes from 0x3B — all must come from correct regs."""
|
||
result = i2c_read_seq(self.mpu, 0x3B, 14)
|
||
self.assertEqual(len(result), 14)
|
||
expected = list(self.mpu.regs[0x3B:0x3B + 14])
|
||
self.assertEqual(result, expected, "14-byte read from 0x3B must match register map")
|
||
|
||
def test_device_reset_bit_auto_cleared(self):
|
||
"""Writing 0x80 to PWR_MGMT_1 (0x6B) must auto-clear bit 7 immediately."""
|
||
self.mpu.handle_event(I2C_START_SEND)
|
||
self.mpu.handle_event(i2c_write(0x6B)) # register address
|
||
self.mpu.handle_event(i2c_write(0x80)) # write 0x80 (DEVICE_RESET bit)
|
||
self.mpu.handle_event(I2C_FINISH)
|
||
# Read back: bit 7 should be 0 (auto-cleared)
|
||
result = i2c_read_seq(self.mpu, 0x6B, 1)
|
||
self.assertEqual(result[0] & 0x80, 0, "DEVICE_RESET bit must auto-clear")
|
||
|
||
def test_alternate_address(self):
|
||
"""MPU6050 at address 0x69 (AD0=HIGH) must still return WHO_AM_I=0x68."""
|
||
mpu69 = MPU6050Slave(addr=0x69)
|
||
result = i2c_read_seq(mpu69, 0x75, 1)
|
||
self.assertEqual(result[0], 0x68)
|
||
|
||
def test_reg_ptr_preserved_across_rstart(self):
|
||
"""Write sets reg_ptr; FINISH then START_RECV should NOT reset reg_ptr."""
|
||
m = MPU6050Slave()
|
||
m.handle_event(I2C_START_SEND)
|
||
m.handle_event(i2c_write(0x75)) # set reg_ptr = 0x75
|
||
m.handle_event(I2C_FINISH) # RSTART — must NOT reset reg_ptr
|
||
m.handle_event(I2C_START_RECV) # read direction START
|
||
val = m.handle_event(I2C_READ)
|
||
self.assertEqual(val, 0x68, "reg_ptr must be preserved across RSTART")
|
||
|
||
def test_write_to_reg(self):
|
||
"""Writing two bytes sets reg address then reg value."""
|
||
m = MPU6050Slave()
|
||
m.handle_event(I2C_START_SEND)
|
||
m.handle_event(i2c_write(0x6B)) # reg address
|
||
m.handle_event(i2c_write(0x01)) # value
|
||
m.handle_event(I2C_FINISH)
|
||
result = i2c_read_seq(m, 0x6B, 1)
|
||
self.assertEqual(result[0], 0x01)
|
||
|
||
|
||
# ── Runner ────────────────────────────────────────────────────────────────────
|
||
|
||
if __name__ == '__main__':
|
||
print('=' * 60)
|
||
print('ESP32 I2C Slave Tests')
|
||
print('=' * 60)
|
||
loader = unittest.TestLoader()
|
||
suite = unittest.TestSuite()
|
||
for cls in [
|
||
TestBMP280Slave,
|
||
TestDS1307Slave,
|
||
TestDS3231Slave,
|
||
TestI2CWriteSink,
|
||
TestMPU6050Slave,
|
||
]:
|
||
suite.addTests(loader.loadTestsFromTestCase(cls))
|
||
|
||
runner = unittest.TextTestRunner(verbosity=2)
|
||
result = runner.run(suite)
|
||
sys.exit(0 if result.wasSuccessful() else 1)
|