"""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 FakeWDT: """machine.WDT-Attrappe: merkt sich Timeout und feed-Aufrufe.""" last = None def __init__(self, timeout=None): self.timeout = timeout self.feeds = 0 FakeWDT.last = self def feed(self): self.feeds += 1 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 m.WDT = FakeWDT m.reset_cause = lambda: 1 m.PWRON_RESET, m.WDT_RESET, m.SOFT_RESET, m.DEEPSLEEP_RESET = 1, 3, 5, 4 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_die_helle_stufe(self): """Viel Licht -> niedrige ADC-Spannung -> nach Spiegelung hoher Wert.""" import brightness # Feste Werte statt settings.json: der Test soll nicht davon abhaengen, # was auf einem Geraet gerade konfiguriert ist. werte = {"dunkel_prozent": 5, "hell_prozent": 50, "schwelle": 1800, "hysterese": 150} 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(werte=werte) self.assertEqual(c.update(hell)[0], brightness.HELL) c = brightness.BrightnessController(werte=werte) self.assertEqual(c.update(dunkel)[0], brightness.DUNKEL) class TestWatchdog(unittest.TestCase): """Dauerbetrieb: Der Watchdog holt das Geraet aus einem Stillstand zurueck.""" 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_timeout_kommt_aus_config_und_ist_in_millisekunden(self): import config self.run_esp32.make_watchdog() self.assertEqual(FakeWDT.last.timeout, int(config.WDT_TIMEOUT * 1000)) def test_gelieferte_funktion_fuettert(self): feed = self.run_esp32.make_watchdog() feed() feed() self.assertEqual(FakeWDT.last.feeds, 2) def test_timeout_liegt_ueber_den_regulaeren_wartezeiten(self): """Sonst startet das Geraet mitten im normalen Betrieb neu. Der WLAN-Aufbau darf allein WIFI_TIMEOUT Sekunden dauern; kommt die Montierung dazu, waere ein knapper Watchdog eine Neustart-Schleife. """ import config self.assertGreater(config.WDT_TIMEOUT, config.WIFI_TIMEOUT + config.MOUNT_TIMEOUT + 10) def test_ohne_machine_gibt_es_keinen_watchdog(self): """Auf dem Mac laeuft dieselbe Schleife -- ohne Watchdog, ohne Fehler.""" _remove("machine", "run_esp32") import run_esp32 meldungen = [] self.assertIsNone(run_esp32.make_watchdog(log=meldungen.append)) self.assertTrue(any("WDT" in m for m in meldungen)) class TestWatchdogInDerSchleife(unittest.TestCase): """poll_loop muss den Watchdog jede Runde fuettern -- auch bei Stoerung.""" def _lauf(self, client, once=True): import displayloop from display import Display from transport import FakeTransport gefuettert = [] d = Display(FakeTransport(echo=False), response=True) displayloop.poll_loop(d, client, once=once, feed=lambda: gefuettert.append(1), log=lambda *a: None) return gefuettert def test_normaler_durchlauf_fuettert(self): class Client: connected = True def connect(self): pass def close(self): pass def get_coordinates(self): return "18:36:56", "+38*47:01" self.assertEqual(len(self._lauf(Client())), 1) def test_auch_bei_ausgefallener_montierung_wird_gefuettert(self): """Der wichtige Fall: "KEIN LINK" ist kein Stillstand. Die Schleife laeuft weiter und zeigt den Hinweis -- der Watchdog darf deshalb nicht zuschlagen, sonst startet das Geraet neu, solange die Montierung aus ist. """ from mount_client import MountError class Client: connected = False def connect(self): raise MountError("keine Verbindung") def close(self): pass def get_coordinates(self): raise MountError("keine Verbindung") self.assertEqual(len(self._lauf(Client())), 1) 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", "settings.py", "transport.py", "display.py", "coords.py", "lx200.py", "mount_client.py", "brightness.py", "displayloop.py", "wifi.py", "mqtt.py", "run_esp32.py", "probe.py", "main.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): """Der Mock sendet 18:36:56; bei Niedrigpraezision (config-Vorgabe) wird daraus die Zehntelform mit Beschriftung.""" self._run(self.run_esp32.main, fake_display=True, once=True, with_mqtt=False) text = "\n".join(self.log) self.assertIn("RA 18h36.9m", text) self.assertIn("DE +38", text) def test_main_once_baut_ein_gueltiges_telegramm(self): hexdump = self._run(self.run_esp32.main, fake_display=True, once=True, with_mqtt=False) self.assertIn("TX: 02 81 80 81", hexdump) # STX, DA, SA, FC self.assertTrue(hexdump.strip().endswith("03")) # ETX def test_gespeicherte_adresse_gilt_nach_dem_neustart(self): """Der Sinn des Speicherns: per MQTT gesetzt, Stromausfall, laeuft weiter. main() ohne host nimmt, was in settings.json steht -- hier der Mock. """ import contextlib import io import os import settings pfad = "test_esp32_settings_tmp.json" alt = settings.PFAD settings.PFAD = pfad try: settings.update({"mount_host": self.mock.host}) with contextlib.redirect_stdout(io.StringIO()): self.run_esp32.main(port=self.mock.port, fake_display=True, once=True, with_mqtt=False, log=self.log.append) finally: settings.PFAD = alt try: os.remove(pfad) except OSError: pass text = "\n".join(self.log) self.assertIn("Montierung: %s:%d" % (self.mock.host, self.mock.port), text) self.assertIn("RA 18h36.9m", text) # der Mock hat wirklich geantwortet 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, with_mqtt=False, 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. # Seit jedes Zeichen einzeln positioniert wird, steht der Text nicht mehr # am Stueck im Telegramm; deshalb ueber die Cursor-Sequenzen zurueckrechnen. from test_display import zeile as _zeile frames = [bytes(int(b, 16) for b in z.split(":", 1)[1].split()) for z in out.getvalue().splitlines() if z.startswith("TX:")] for i, erwartet in enumerate(config.NOLINK_LINES): self.assertEqual(_zeile(frames, i).strip(), erwartet) 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, with_mqtt=False, 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()