velxio/test/backend/unit/test_i2c_slaves.py

"""
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)