LAN7800 EEPROM Programmer

"""Offline tests for bulk program mode: sequence math, image stamping, and the
resumable manifest. No hardware."""
import os
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from lan7800.eeprom import Eeprom, EepromError  # noqa
from lan7800 import bulk  # noqa
from tests.make_fixture import build  # noqa


def test_sequence_basic():
    seq = bulk.sequence("00:80:0F:78:00:00", count=3)
    assert [u["mac"] for u in seq] == [
        "00:80:0F:78:00:00", "00:80:0F:78:00:01", "00:80:0F:78:00:02"]
    assert [u["index"] for u in seq] == [0, 1, 2]


def test_sequence_step_and_serial():
    seq = bulk.sequence("00:80:0F:78:00:00", count=3, step=16)
    assert [u["mac"] for u in seq] == [
        "00:80:0F:78:00:00", "00:80:0F:78:00:10", "00:80:0F:78:00:20"]
    # serial tracks MAC 1:1 as uppercase hex, no separators
    assert seq[2]["serial"] == "00800F780020"


def test_sequence_carry_across_octet():
    seq = bulk.sequence("00:80:0F:78:00:FF", count=2)
    assert seq[1]["mac"] == "00:80:0F:78:01:00"


def test_local_admin_sets_bit():
    seq = bulk.sequence("00:11:22:33:44:55", count=1, local_admin=True)
    # 0x00 -> 0x02 (locally administered, unicast)
    assert seq[0]["mac"] == "02:11:22:33:44:55"


def test_multicast_base_refused():
    try:
        bulk.sequence("01:00:5E:00:00:01", count=1)  # low bit set = multicast
    except EepromError:
        return
    raise AssertionError("expected multicast base to be refused")


def test_oui_overflow_refused():
    # only 2 NIC values left in this OUI, ask for 4 -> must refuse
    try:
        bulk.sequence("00:80:0F:FF:FF:FE", count=4)
    except EepromError:
        return
    raise AssertionError("expected OUI-crossing overflow to be refused")


def test_build_image_stamps_mac_and_serial():
    tmpl = Eeprom.from_bytes(build())          # fixture has a 12-char serial slot
    data = bulk.build_image(tmpl, "00:80:0F:78:00:2A", "00800F78002A")
    e = Eeprom.from_bytes(data)
    assert e.has_signature
    assert e.mac == "00:80:0F:78:00:2A"
    assert e.string("serial").value == "00800F78002A"


def test_manifest_resume_and_dedupe():
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "run.csv")
        m = bulk.Manifest(path)
        assert m.next_index() == 0
        m.append(index=0, usb_path="1-1", old_mac="FF:FF:FF:FF:FF:FF",
                 old_serial="", new_mac="00:80:0F:78:00:00",
                 new_serial="00800F780000", verify="PASS")
        m.append(index=1, usb_path="1-1", old_mac="FF:FF:FF:FF:FF:FF",
                 old_serial="", new_mac="00:80:0F:78:00:01",
                 new_serial="00800F780001", verify="PASS")
        # reopen -> resumes at next index, remembers burned MACs
        m2 = bulk.Manifest(path)
        assert m2.next_index() == 2
        assert m2.has_mac("00:80:0F:78:00:00")
        assert os.path.exists(os.path.join(d, "run.json"))
        # refuse to reuse a burned MAC
        try:
            m2.append(index=2, usb_path="1-1", old_mac="x", old_serial="",
                      new_mac="00:80:0F:78:00:00", new_serial="00800F780000",
                      verify="PASS")
        except EepromError:
            return
        raise AssertionError("expected duplicate-MAC append to be refused")


def test_last_burned_mac_for_swap_guard():
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "run.csv")
        m = bulk.Manifest(path)
        assert m.last_burned_mac() is None
        m.append(index=0, usb_path="1-1", old_mac="FF:FF:FF:FF:FF:FF",
                 old_serial="", new_mac="00:80:0F:78:00:00",
                 new_serial="00800F780000", verify="PASS")
        m.append(index=1, usb_path="1-1", old_mac="FF:FF:FF:FF:FF:FF",
                 old_serial="", new_mac="00:80:0F:78:00:01",
                 new_serial="00800F780001", verify="PASS")
        # a DRYRUN row must NOT count as the last burned unit
        m.append(index=2, usb_path="", old_mac="", old_serial="",
                 new_mac="00:80:0F:78:00:99", new_serial="00800F780099",
                 verify="DRYRUN")
        assert m.last_burned_mac() == "00:80:0F:78:00:01"


def test_manifest_dryrun_counts_as_done():
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "run.csv")
        m = bulk.Manifest(path)
        m.append(index=0, usb_path="", old_mac="", old_serial="",
                 new_mac="00:80:0F:78:00:00", new_serial="00800F780000",
                 verify="DRYRUN")
        assert bulk.Manifest(path).next_index() == 1


def _run():
    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
    for fn in fns:
        fn()
        print(f"  ok  {fn.__name__}")
    print(f"\n{len(fns)}/{len(fns)} passed")


if __name__ == "__main__":
    _run()