Files
grossanzeige/test_esp32.py
T
admin 142a621066 Dauerbetrieb: Autostart (main.py) und Watchdog
MicroPython fuehrt main.py nach jedem Reset selbst aus -- nach Stromausfall oder
Watchdog-Neustart laeuft die Anzeige damit ohne Notebook wieder an. Bisher gab es
die Datei bewusst nicht, weil am Aufbau gearbeitet wurde.

**Das Abbruchfenster ist kein Komfort, sondern die Notbremse.** Ohne es waere ein
Geraet, dessen Programm beim Start haengt, nur noch durch Neuflashen erreichbar:
Die Endlosschleife belegt die REPL sofort nach dem Reset. main.py wartet deshalb
erst AUTOSTART_DELAY (3 s), in denen Ctrl-C zurueck an die REPL fuehrt -- dann
laeuft auch kein Watchdog.

Watchdog (WDT_TIMEOUT = 120 s): startet das Geraet neu, wenn die Poll-Schleife
stehenbleibt. Gefuettert wird zu Beginn *jedes* Durchlaufs, auch nach einem
Fehler -- "KEIN LINK" ist kein Stillstand, die Schleife laeuft ja weiter. Der
Wert muss ueber allem liegen, was regulaer blockieren darf (WLAN 20 s,
Montierung 3 s, dazu ein MQTT-Versuch); knapper waere er eine Neustart-Schleife.

Ein gestarteter Watchdog laesst sich nicht mehr abschalten. Deshalb legt ihn nur
main.py an -- im Handbetrieb ueber die REPL startete das Geraet sonst neu,
waehrend man nachdenkt.

Weiteres:
- Nach einem Watchdog-Neustart meldet main.py das beim Start
  (machine.reset_cause()), sonst saehe er aus wie ein Stromausfall.
- Eine Ausnahme faengt main.py ab und startet nach RESTART_DELAY (10 s) neu,
  statt in die REPL zu fallen und die Anzeige einzufrieren. Die Pause verhindert
  eine Neustart-Schleife im Sekundentakt.
- Autostart faehrt mit Helligkeitsregelung (with_ldr=True, Nutzerentscheidung).

Am Geraet bestaetigt: Autostart nach Reset laeuft durch (Watchdog aktiv, WLAN,
MQTT), ein absichtlich herbeigefuehrter Stillstand loest den Watchdog aus und das
Geraet kommt von allein wieder hoch -- mit der Meldung "Vorheriger Lauf wurde vom
Watchdog beendet". Ctrl-C im Startfenster gibt die REPL frei.

Beim Geraetetest aufgefallen: make_watchdog wurde ohne log aufgerufen, im
Startprotokoll stand also nichts vom Watchdog. Behoben und durch einen Test
festgehalten. 178 Tests gruen (neu: test_main.py).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 18:41:33 +02:00

735 lines
25 KiB
Python

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