grossanzeige/test_esp32.py

"""Tests fuer den MicroPython-Teil -- ohne ESP32, mit Ersatzmodulen.

machine und network gibt es auf dem Mac nicht. Hier werden sie durch Attrappen
ersetzt, damit sich UART-Transport, WLAN-Anbindung, ADC-Auswertung und die
ueberlaufsichere Zeitrechnung auch ohne Geraet pruefen lassen. Getestet wird
damit die Logik, nicht die Hardware -- ob 19200 8E1 auf dem realen UART
ankommen, zeigt erst der Aufbau.

    python3 -m unittest test_esp32
"""

import importlib
import sys
import types
import unittest

import migra
import ticks


# -- Ersatzmodule ----------------------------------------------------------

class FakeUART:
    """machine.UART-Attrappe: merkt sich Parameter und Gesendetes."""

    last = None

    def __init__(self, uart_id, **kwargs):
        self.uart_id = uart_id
        self.kwargs = kwargs
        self.rx = bytearray()       # was die Anzeige "zurueckschickt"
        self.written = []
        self.deinit_called = False
        FakeUART.last = self

    def any(self):
        return len(self.rx)

    def read(self, n=None):
        if n is None:
            n = len(self.rx)
        chunk = bytes(self.rx[:n])
        del self.rx[:n]
        return chunk or None

    def write(self, data):
        self.written.append(bytes(data))
        return len(data)

    def deinit(self):
        self.deinit_called = True


class FakePin:
    def __init__(self, number):
        self.number = number

    def __eq__(self, other):
        return isinstance(other, FakePin) and other.number == self.number

    def __repr__(self):
        return "FakePin(%d)" % self.number


class FakeADC:
    ATTN_11DB = 3

    last = None

    def __init__(self, pin):
        self.pin = pin
        self.value = 0
        self.attenuation = None
        FakeADC.last = self

    def atten(self, value):
        self.attenuation = value

    def read_u16(self):
        return self.value


class FakeWLAN:
    """network.WLAN-Attrappe. connect() gelingt, wenn succeed gesetzt ist."""

    succeed = True
    last = None

    def __init__(self, mode):
        self.mode = mode
        self._active = False
        self._connected = False
        self.credentials = None
        self.disconnect_calls = 0
        FakeWLAN.last = self

    def active(self, value=None):
        if value is not None:
            self._active = value
        return self._active

    def connect(self, ssid, password):
        self.credentials = (ssid, password)
        self._connected = FakeWLAN.succeed

    def isconnected(self):
        return self._connected

    def disconnect(self):
        self.disconnect_calls += 1
        self._connected = False

    def ifconfig(self):
        return ("192.168.1.77", "255.255.255.0", "192.168.1.1", "192.168.1.1")


def _install_machine():
    """Ein Ersatz-machine bereitstellen und transport neu laden."""
    m = types.ModuleType("machine")
    m.UART = FakeUART
    m.Pin = FakePin
    m.ADC = FakeADC
    sys.modules["machine"] = m
    return m


def _install_network():
    n = types.ModuleType("network")
    n.STA_IF = 0
    n.WLAN = FakeWLAN
    sys.modules["network"] = n
    return n


def _remove(*names):
    for name in names:
        sys.modules.pop(name, None)


# -- UART-Transport --------------------------------------------------------

class TestUartTransport(unittest.TestCase):

    def setUp(self):
        _install_machine()
        import transport
        self.transport = importlib.reload(transport)
        import config
        self.config = config

    def tearDown(self):
        _remove("machine")
        import transport
        importlib.reload(transport)     # zurueck auf UART = None

    def _open(self, timeout=1.0):
        return self.transport.UartTransport(
            uart_id=self.config.UART_ID,
            baudrate=self.config.BAUDRATE,
            parity=self.config.PARITY,
            bytesize=self.config.BYTESIZE,
            stopbits=self.config.STOPBITS,
            timeout=timeout,
            tx=self.config.UART_TX,
            rx=self.config.UART_RX,
        )

    def test_uart_parameter_aus_config(self):
        """19200 8E1 auf UART1, Pins wie im Schaltplan (GPIO21/GPIO20)."""
        self._open()
        u = FakeUART.last
        self.assertEqual(u.uart_id, 1)
        self.assertEqual(u.kwargs["baudrate"], 19200)
        self.assertEqual(u.kwargs["bits"], 8)
        self.assertEqual(u.kwargs["parity"], 0)      # 0 = even
        self.assertEqual(u.kwargs["stop"], 1)
        self.assertEqual(u.kwargs["tx"], FakePin(21))
        self.assertEqual(u.kwargs["rx"], FakePin(20))

    def test_write_leert_erst_den_empfangspuffer(self):
        """Stoerbytes vor dem Senden verwerfen -- wie SerialTransport."""
        t = self._open()
        FakeUART.last.rx.extend(b"\xff\x00Muell")
        t.write(b"\x02\x81\x80\x81\x03")
        self.assertEqual(FakeUART.last.any(), 0)
        self.assertEqual(FakeUART.last.written, [b"\x02\x81\x80\x81\x03"])

    def test_read_response_liest_bis_etx(self):
        t = self._open()
        FakeUART.last.rx.extend(b"\x02\x80\x81\x800\x03")
        frame = t.read_response()
        self.assertEqual(frame, b"\x02\x80\x81\x800\x03")
        self.assertEqual(migra.parse_response(frame), "0")

    def test_read_response_schneidet_nachlauf_ab(self):
        """Nach ETX ist Schluss; Nachlauf wuerde parse_response scheitern lassen."""
        t = self._open()
        FakeUART.last.rx.extend(b"\x02\x80\x81\x800\x03\xff\xfe")
        self.assertEqual(t.read_response(), b"\x02\x80\x81\x800\x03")

    def test_read_response_timeout_liefert_leer(self):
        t = self._open(timeout=0.05)
        self.assertEqual(t.read_response(), b"")

    def test_close_gibt_den_uart_frei(self):
        t = self._open()
        t.close()
        self.assertTrue(FakeUART.last.deinit_called)


class TestOpenDisplayAufMicropython(unittest.TestCase):
    """open_display muss auf dem ESP32 den UART nehmen, nicht pyserial."""

    def setUp(self):
        _install_machine()
        import transport
        importlib.reload(transport)
        import config
        self.config = config
        self._orig = config.MICROPYTHON
        config.MICROPYTHON = True

    def tearDown(self):
        self.config.MICROPYTHON = self._orig
        _remove("machine")
        import transport
        importlib.reload(transport)

    def test_waehlt_uart_transport(self):
        import display
        d = display.open_display()
        self.assertEqual(FakeUART.last.uart_id, self.config.UART_ID)
        d.close()


# -- Zeitrechnung ----------------------------------------------------------

class FakeTicks:
    """time.ticks_*-Attrappe von MicroPython, inklusive Ueberlauf bei 2^30."""

    MASK = (1 << 30) - 1

    def __init__(self, now=0):
        self.now = now

    def ticks_ms(self):
        return self.now

    def ticks_add(self, t, delta):
        return (t + delta) & self.MASK

    def ticks_diff(self, a, b):
        d = (a - b) & self.MASK
        if d > self.MASK // 2:
            d -= self.MASK + 1
        return d


class TestTicks(unittest.TestCase):

    def setUp(self):
        self._orig_time = ticks.time
        self.clock = FakeTicks()
        ticks.time = self.clock

    def tearDown(self):
        ticks.time = self._orig_time

    def test_frist_laeuft_normal_ab(self):
        self.clock.now = 1000
        end = ticks.ticks_deadline(1.5)
        self.assertFalse(ticks.ticks_expired(end))
        self.clock.now = 2499
        self.assertFalse(ticks.ticks_expired(end))
        self.clock.now = 2500
        self.assertTrue(ticks.ticks_expired(end))

    def test_frist_ueberlebt_den_ueberlauf(self):
        """Der Kern: 500 ms vor dem Ueberlauf eine 1-s-Frist setzen.

        Mit "jetzt + timeout" und einfachem Vergleich haenge die Warteschleife
        hier fast 12 Tage, weil der Zaehler zurueckspringt. Mit ticks_diff
        laeuft die Frist wie erwartet nach 1000 ms ab.
        """
        self.clock.now = FakeTicks.MASK - 500
        end = ticks.ticks_deadline(1.0)
        self.assertLess(end, self.clock.now)        # der Zaehler ist umgelaufen

        self.assertFalse(ticks.ticks_expired(end))  # sofort: noch nicht
        self.clock.now = 400                        # 900 ms spaeter
        self.assertFalse(ticks.ticks_expired(end))
        self.clock.now = 499                        # genau 1000 ms
        self.assertTrue(ticks.ticks_expired(end))
        self.clock.now = 600
        self.assertTrue(ticks.ticks_expired(end))


class TestTicksCPython(unittest.TestCase):
    """Eigene Klasse: hier laeuft die echte Uhr, nicht die Attrappe."""

    def test_monotonic_variante(self):
        self.assertTrue(ticks.monotonic_expired(ticks.monotonic_deadline(0.0)))
        self.assertFalse(ticks.monotonic_expired(ticks.monotonic_deadline(30.0)))

    def test_auf_dem_mac_ist_monotonic_ausgewaehlt(self):
        self.assertIs(ticks.deadline, ticks.monotonic_deadline)
        self.assertIs(ticks.expired, ticks.monotonic_expired)


# -- Helligkeit ------------------------------------------------------------

class TestLdrKennlinie(unittest.TestCase):
    """Der Teiler auf der Platine dreht die Kennlinie um (siehe config)."""

    def test_invertiert_spiegelt_den_rohwert(self):
        import brightness
        self.assertEqual(brightness.raw_to_ldr(0, inverted=True, full_scale=4095), 4095)
        self.assertEqual(brightness.raw_to_ldr(4095, inverted=True, full_scale=4095), 0)
        self.assertEqual(brightness.raw_to_ldr(1000, inverted=True, full_scale=4095), 3095)

    def test_nicht_invertiert_laesst_durch(self):
        import brightness
        self.assertEqual(brightness.raw_to_ldr(1234, inverted=False), 1234)

    def test_vorgabe_kommt_aus_config(self):
        import brightness
        import config
        self.assertEqual(brightness.raw_to_ldr(0),
                         config.LDR_MAX if config.LDR_INVERTED else 0)

    def test_helle_umgebung_ergibt_hohe_stufe(self):
        """Viel Licht -> niedrige ADC-Spannung -> nach Spiegelung hoher Wert."""
        import brightness
        hell = brightness.raw_to_ldr(100, inverted=True, full_scale=4095)
        dunkel = brightness.raw_to_ldr(4000, inverted=True, full_scale=4095)
        c = brightness.BrightnessController()
        self.assertEqual(c.update(hell)[0], len(brightness.PERCENTS) - 1)
        c = brightness.BrightnessController()
        self.assertEqual(c.update(dunkel)[0], 0)


class TestLdrLeser(unittest.TestCase):
    """run_esp32.make_ldr_reader: ADC lesen, auf 12 bit bringen, spiegeln."""

    def setUp(self):
        _install_machine()
        import transport
        importlib.reload(transport)
        _remove("run_esp32")
        import run_esp32
        self.run_esp32 = run_esp32

    def tearDown(self):
        _remove("machine", "run_esp32")
        import transport
        importlib.reload(transport)

    def test_liest_den_konfigurierten_pin(self):
        import config
        self.run_esp32.make_ldr_reader()
        self.assertEqual(FakeADC.last.pin, FakePin(config.LDR_PIN))
        self.assertEqual(FakeADC.last.attenuation, FakeADC.ATTN_11DB)

    def test_skaliert_und_spiegelt(self):
        import config
        read = self.run_esp32.make_ldr_reader()
        FakeADC.last.value = 0              # dunkelste ADC-Spannung
        self.assertEqual(read(), config.LDR_MAX)
        FakeADC.last.value = 65535          # voller Ausschlag
        self.assertEqual(read(), 0)


# -- WLAN ------------------------------------------------------------------

class TestWifi(unittest.TestCase):

    def setUp(self):
        _install_network()
        FakeWLAN.succeed = True
        _remove("wifi", "wifi_config")
        import wifi
        self.wifi = wifi
        # Zugangsdaten unterschieben, damit der Test ohne wifi_config.py laeuft
        cfg = types.ModuleType("wifi_config")
        cfg.SSID = "Sternwarte"
        cfg.PASSWORD = "geheim"
        sys.modules["wifi_config"] = cfg

    def tearDown(self):
        _remove("network", "wifi", "wifi_config")

    def _quiet(self, *args):
        pass

    def test_connect_meldet_sich_mit_den_zugangsdaten_an(self):
        cfg = self.wifi.connect(log=self._quiet)
        self.assertEqual(FakeWLAN.last.credentials, ("Sternwarte", "geheim"))
        self.assertTrue(FakeWLAN.last.active())
        self.assertEqual(cfg[0], "192.168.1.77")

    def test_connect_wirft_wifierror_bei_zeitablauf(self):
        FakeWLAN.succeed = False
        with self.assertRaises(self.wifi.WifiError):
            self.wifi.connect(timeout=0.2, log=self._quiet)
        # Der haengende Versuch muss abgebrochen werden, sonst kollidiert er
        # mit dem naechsten connect().
        self.assertEqual(FakeWLAN.last.disconnect_calls, 1)

    def test_wifierror_ist_ein_linkerror(self):
        """Damit die Poll-Schleife den Ausfall selbst abfaengt."""
        import displayloop
        self.assertTrue(issubclass(self.wifi.WifiError, displayloop.LinkError))

    def test_ensure_verbindet_nur_wenn_noetig(self):
        self.wifi.connect(log=self._quiet)
        wlan = FakeWLAN.last
        wlan.credentials = None
        self.wifi.ensure(log=self._quiet)
        self.assertIsNone(wlan.credentials)     # kein zweiter connect()

    def test_fehlende_zugangsdaten_melden_sich_deutlich(self):
        # None in sys.modules laesst "import wifi_config" mit ImportError
        # scheitern -- unabhaengig davon, ob auf diesem Rechner schon eine
        # echte wifi_config.py liegt. Ein blosses _remove() wuerde die Datei
        # einfach neu laden und der Test haenge am Zustand des Entwicklerrechners.
        sys.modules["wifi_config"] = None
        with self.assertRaises(self.wifi.WifiError) as ctx:
            self.wifi.credentials()
        self.assertIn("wifi_config", str(ctx.exception))



# -- Vertraeglichkeit mit MicroPython --------------------------------------

# Die Dateien, die auf dem Geraet landen -- muss zu deploy.sh passen.
DEPLOYED = (
    "ticks.py", "migra.py", "config.py", "transport.py", "display.py",
    "coords.py", "lx200.py", "mount_client.py", "brightness.py",
    "displayloop.py", "wifi.py", "run_esp32.py",
)


class TestMicropythonVertraeglichkeit(unittest.TestCase):
    """Statische Pruefung dessen, was CPython kann und MicroPython nicht.

    Solche Fehler entgehen den ueblichen Tests, weil sie hier auf CPython
    laufen: str.ljust gibt es auf dem Mac, auf dem ESP32 nicht -- aufgefallen
    ist das erst am Geraet. Diese Liste haelt fest, was dort nachgeprueft
    wurde (MicroPython 1.28, ESP32_GENERIC_C3).
    """

    # Auf dem Geraet mit hasattr geprueft: fehlen.
    FEHLENDE_STRINGMETHODEN = (".ljust(", ".rjust(", ".center(",
                               ".zfill(", ".casefold(", ".expandtabs(")

    # Module, die es unter MicroPython nicht gibt. Sie duerfen in den
    # deployten Dateien hoechstens innerhalb einer Funktion importiert werden
    # (wie in brightness.main), nicht auf Modulebene.
    FEHLENDE_MODULE = ("argparse", "threading", "subprocess", "unittest")

    def _quelle(self, name):
        with open(name, encoding="utf-8") as f:
            return f.read()

    def test_keine_fehlenden_stringmethoden(self):
        for name in DEPLOYED:
            src = self._quelle(name)
            for methode in self.FEHLENDE_STRINGMETHODEN:
                self.assertNotIn(
                    methode, src,
                    "%s benutzt %s -- das gibt es unter MicroPython nicht" % (name, methode))

    def test_keine_fehlenden_module_auf_modulebene(self):
        import ast
        for name in DEPLOYED:
            baum = ast.parse(self._quelle(name), filename=name)
            for knoten in baum.body:          # nur die oberste Ebene
                if isinstance(knoten, ast.Import):
                    namen = [a.name.split(".")[0] for a in knoten.names]
                elif isinstance(knoten, ast.ImportFrom):
                    namen = [(knoten.module or "").split(".")[0]]
                else:
                    continue
                for modul in namen:
                    self.assertNotIn(
                        modul, self.FEHLENDE_MODULE,
                        "%s importiert %s auf Modulebene -- unter MicroPython "
                        "scheitert damit schon der Import" % (name, modul))

    def test_deploy_sh_kopiert_genau_diese_dateien(self):
        """Sonst laeuft die Liste hier und die im Deploy-Skript auseinander."""
        with open("deploy.sh", encoding="utf-8") as f:
            skript = f.read()
        block = skript.split('FILES="', 1)[1].split('"', 1)[0]
        im_skript = tuple(z.strip() for z in block.split() if z.strip())
        self.assertEqual(sorted(im_skript), sorted(DEPLOYED))


# -- Gesamtkette -----------------------------------------------------------

class TestEsp32GegenMock(unittest.TestCase):
    """Der ESP32-Pfad gegen einen echten Mock, ueber eine echte TCP-Verbindung.

    Alles ausser der Hardware ist hier zusammengeschaltet: WLAN-Attrappe,
    mount_client, displayloop, Telegrammbau. Das ist derselbe Ablauf, der auf
    dem Geraet gegen den Mock auf dem Mac laufen soll -- nur eben auf einem
    Rechner und mit Attrappe statt Funk.
    """

    def setUp(self):
        _install_machine()
        _install_network()
        FakeWLAN.succeed = True
        _remove("wifi", "wifi_config", "run_esp32")
        cfg = types.ModuleType("wifi_config")
        cfg.SSID = "Sternwarte"
        cfg.PASSWORD = "geheim"
        sys.modules["wifi_config"] = cfg

        import transport
        importlib.reload(transport)

        from mount_mock import MountMock
        self.mock = MountMock(host="127.0.0.1", port=0, move=True)
        self.mock.start()

        import run_esp32
        self.run_esp32 = run_esp32
        self.log = []

    def tearDown(self):
        self.mock.stop()
        _remove("machine", "network", "wifi", "wifi_config", "run_esp32")
        import transport
        importlib.reload(transport)

    def _run(self, fn, **kwargs):
        """Aufrufen und dabei die Hex-Ausgabe des FakeTransport wegfangen."""
        import contextlib
        import io
        with contextlib.redirect_stdout(io.StringIO()) as out:
            fn(host=self.mock.host, port=self.mock.port,
               log=self.log.append, **kwargs)
        return out.getvalue()

    def test_selftest_geht_durch_alle_stufen(self):
        self._run(self.run_esp32.selftest)
        text = "\n".join(self.log)
        self.assertIn("WLAN", text)
        self.assertIn("192.168.1.77", text)         # IP der WLAN-Attrappe
        self.assertIn("18:36:56", text)             # RA vom Mock
        self.assertIn("Selbsttest bestanden", text)

    def test_main_once_zeigt_die_koordinaten(self):
        self._run(self.run_esp32.main, fake_display=True, once=True)
        text = "\n".join(self.log)
        self.assertIn("18h36m56s", text)            # RA, fuer die Anzeige formatiert
        self.assertIn("+38", text)                  # DEC

    def test_main_once_baut_ein_gueltiges_telegramm(self):
        hexdump = self._run(self.run_esp32.main, fake_display=True, once=True)
        self.assertIn("TX: 02 81 80 81", hexdump)   # STX, DA, SA, FC
        self.assertTrue(hexdump.strip().endswith("03"))     # ETX

    def test_ohne_mock_zeigt_die_anzeige_kein_link(self):
        """Faellt die Gegenstelle aus, laeuft die Schleife weiter."""
        import config
        import contextlib
        import io
        self.mock.stop()
        with contextlib.redirect_stdout(io.StringIO()) as out:
            self.run_esp32.main(host="127.0.0.1", port=self.mock.port,
                                fake_display=True, once=True, log=self.log.append)
        self.assertTrue(any("Montierung" in line for line in self.log))
        # Die Hinweiszeilen stehen wirklich im Telegramm -- ohne den Text hier
        # zu wiederholen, sonst bricht der Test bei jeder Umformulierung.
        gesendet = "".join(chr(int(b, 16)) if 32 <= int(b, 16) < 127 else "."
                           for zeile in out.getvalue().splitlines()
                           if zeile.startswith("TX:")
                           for b in zeile.split(":", 1)[1].split())
        for zeile in config.NOLINK_LINES:
            self.assertIn(zeile, gesendet)

    def test_ohne_wlan_zeigt_die_anzeige_kein_wlan(self):
        import config
        FakeWLAN.succeed = False
        # Den WLAN-Timeout kurz drehen, sonst wartet der Test zweimal die
        # vollen config.WIFI_TIMEOUT Sekunden ab (vorab-connect und ensure).
        orig = config.WIFI_TIMEOUT
        config.WIFI_TIMEOUT = 0.2
        try:
            import contextlib
            import io
            with contextlib.redirect_stdout(io.StringIO()):
                self.run_esp32.main(host=self.mock.host, port=self.mock.port,
                                    fake_display=True, once=True, log=self.log.append)
        finally:
            config.WIFI_TIMEOUT = orig
        text = "\n".join(self.log)
        self.assertIn("WLAN", text)
        self.assertIn("Netz", text)     # LinkError wurde in der Schleife gefangen


if __name__ == "__main__":
    unittest.main()