admin/grossanzeige
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
grossanzeige/test_esp32.py
T
admin 4461035777 Adresse der Montierung per MQTT umstellbar und gespeichert 
Zwischen Mock und echter GM4000 umschalten, ohne zu deployen und ohne Neustart:

    mosquitto_pub -t grossanzeige/set/mount_host -m 192.168.1.115
    mosquitto_pub -t grossanzeige/set/mount_host -m 192.168.178.26
    mosquitto_pub -t grossanzeige/set/mount_host -m ''

Die Adresse liegt in settings.json auf dem Geraet und uebersteht damit
Stromausfall und Watchdog-Neustart. deploy.sh kopiert die Datei bewusst nicht
mit, die Einstellung ueberlebt also auch ein Deployen.

**Leer heisst: config.MOUNT_HOST gilt.** mount_client.resolve_host loest auf in
der Reihenfolge: ausdruecklich mitgegebener Host (REPL, run_display --host) ->
mount_host aus settings.json -> config.MOUNT_HOST. Ohne diesen Rueckweg wuerde
eine einmal per MQTT gesetzte Adresse die Zeile in config.py fuer immer
verdecken, auch nach einem Deployen mit geaendertem Code.

Pruefung: settings._pruefe unterscheidet jetzt Zahl und Text am Typ der Vorgabe
in DEFAULTS -- eine neue Einstellung wird weiterhin an genau einer Stelle
eingetragen. Fuer mount_host sind nur Buchstaben, Ziffern, Punkt und
Bindestrich erlaubt, hoechstens 63 Zeichen: Der Wert geht ungeprueft in
socket.getaddrinfo, und von MQTT kommt alles Moegliche. Ungueltiges wird
verworfen, settings.json bleibt unberuehrt.

Sofort wirksam: Der Rueckruf traegt die neue Adresse in den MountClient ein und
trennt nur die bestehende Verbindung -- neu verbindet die Poll-Schleife im
naechsten Durchlauf, mit der gewohnten Fehlerbehandlung. Antwortet dort niemand,
steht "KEIN LINK", und eine weitere Nachricht holt das Geraet zurueck.
grossanzeige/status/mount_host meldet, welche Adresse gerade abgefragt wird --
die Rueckmeldung steht direkt hinter bridge.check() und nicht im Erfolgspfad,
kommt also auch dann, wenn die Montierung schweigt.

Der on_change-Rueckruf ist als run_esp32.make_on_change aus main() herausgeloest.
test_mqtt.py prueft ihn bisher als Abschrift ("Dieselbe Logik wie in
run_esp32.main"); jetzt laeuft der echte Code durch die Tests.

201 Tests gruen. Neu, unter anderem:
- TestUmschaltenImLaufendenBetrieb faehrt die ganze Kette an echten Sockets:
  erst "KEIN LINK" an einer toten Adresse, nach dem set/mount_host die
  Koordinaten vom Mock.
- TestAdresswahl (test_lx200) haelt die Reihenfolge von resolve_host fest.
- test_esp32: main() ohne host nimmt nach einem Neustart, was gespeichert ist.

Nicht angefasst: run_display.py auf dem Mac behaelt config.MOUNT_HOST als
Vorgabe fuer --host. Dort gibt es keine MQTT-Bruecke, und settings.json ist
Zustand des Geraets, nicht des Macs.

Am Geraet bestaetigt (2026-07-29), am Broker mitgeschnitten. Umschalten im
laufenden Betrieb, mit dem Mock auf dem Mac als Ausgangslage:

    0.0s  status/mount_host  192.168.178.26   Ausgangslage, link = 1
    ->  set/mount_host = 192.168.178.99
    5.1s  status/mount_host  192.168.178.99   Rueckmeldung
    5.9s  status/link        0                dort antwortet niemand
    ->  set/mount_host = ''
   17.8s  status/mount_host  192.168.178.26   Vorgabe aus config.py
   18.6s  status/link        1                Koordinaten laufen wieder

Und ueber den Neustart hinweg -- der eigentliche Zweck des Speicherns:

    ->  set/mount_host = 192.168.178.99
    1.6s  status/mount_host  192.168.178.99
    settings.json auf dem Geraet:
      {"hysterese": 150, "mount_host": "192.168.178.99", "schwelle": 1800, ...}
    ->  Hardware-Reset
    8.6s  status/online      0                Last Will
   15.5s  status/online      1                Autostart durch
   15.5s  status/mount_host  192.168.178.99   aus der Datei, nicht aus config.py

Die alte settings.json ohne mount_host wird beim Laden einfach um die Vorgabe
ergaenzt; die eingestellten Helligkeiten bleiben unberuehrt. Eine Migration war
nicht noetig.

set/mount_host wird ohne Retain-Flag geschickt: Das ist ein Befehl, kein
Zustand. Retained laege er im Broker und wuerde bei jedem Reconnect erneut
zugestellt.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-29 19:34:09 +02:00

765 lines
26 KiB
Python
Raw Permalink Blame History

"""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()
Reference in New Issue View Git Blame Copy Permalink