"""Tests fuer den MJPEG-Server der Kameraseite (``pi/vogelkamera.py``). ./venv/bin/python -m unittest discover tests Laeuft ohne Raspberry Pi und ohne Kamera: ``rpicam-vid`` wird nie gestartet, der Bildpuffer von Hand befuellt. Geprueft wird die HTTP-Schicht -- und die ist der Teil, der schon Schaden angerichtet hat. Dass ein zweiter Zuschauer abgewiesen wird, ist keine Kosmetik: ein stiller zweiter Client halbiert die Bildrate und der Detector faellt in Timeouts, ohne dass irgendwo ein Fehler steht. ⚠️ **Diese Datei war vom 25.09. bis zum 09.10.2026 rot** -- 19 Fehler. Commit b57c1b3 ersetzte die picamera2-Fassung durch die rpicam-vid-Fassung, die wirklich auf dem Geraet laeuft; die Tests blieben auf den alten Namen stehen (``Bildpuffer`` statt ``FrameBuffer``, ``Handler`` statt ``MJPEGHandler``) und pruefen Code, den es nicht mehr gab. 🔑 Beim Nachziehen kam heraus, dass ``pi/README.md`` die 503-Absage fuer den zweiten Zuschauer die ganze Zeit **versprochen** hat, waehrend die Fassung vom Geraet ihn nur zaehlte. Die Tests haben also nicht nur den Namen verfehlt, sondern eine fehlende Schutzfunktion verdeckt. """ from __future__ import annotations import importlib.util import json import threading import time import unittest import urllib.error import urllib.request from pathlib import Path _pfad = Path(__file__).resolve().parents[1] / "pi" / "vogelkamera.py" _spec = importlib.util.spec_from_file_location("vogelkamera", _pfad) vogelkamera = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(vogelkamera) # Ein winziges, aber gueltiges JPEG waere hier Overkill -- der Server reicht # Bytes durch, ohne sie zu deuten. BILD = b"\xff\xd8" + b"x" * 64 + b"\xff\xd9" class ServerTests(unittest.TestCase): def setUp(self): # Der Server haelt seinen Zustand in Modulvariablen, nicht in einer # Instanz. Jeder Test setzt sie zurueck, sonst schleppt er den Stand # des vorigen mit -- der Zuschauerplatz ganz besonders. vogelkamera.frame_buffer = vogelkamera.FrameBuffer() vogelkamera.zuschauer_platz = threading.Semaphore(1) vogelkamera.bilder_gesamt = 0 vogelkamera.zuschauer = 0 vogelkamera.START_ZEIT = time.time() self.server = vogelkamera.ThreadedHTTPServer( ("127.0.0.1", 0), vogelkamera.MJPEGHandler ) self.port = self.server.server_address[1] self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() self.addCleanup(self._stop) def _stop(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) def url(self, pfad: str) -> str: return f"http://127.0.0.1:{self.port}{pfad}" def hole(self, pfad: str, timeout: float = 5.0): return urllib.request.urlopen(self.url(pfad), timeout=timeout) def schiebe(self, bild: bytes = BILD) -> None: """Ein Bild einspeisen, so wie es ``capture_loop`` tut -- inklusive Zaehlwerk, denn /status liest beides.""" vogelkamera.frame_buffer.update(bild) with vogelkamera._zaehler_sperre: vogelkamera.bilder_gesamt += 1 def speisen(self) -> threading.Event: """Bilder nachschieben, bis der Test fertig ist.""" stop = threading.Event() def lauf(): while not stop.is_set(): self.schiebe() stop.wait(0.01) threading.Thread(target=lauf, daemon=True).start() self.addCleanup(stop.set) return stop # -- die einfachen Endpunkte ------------------------------------------ def test_startseite_kommt(self): with self.hole("/") as r: self.assertEqual(r.status, 200) self.assertIn(b"/stream", r.read()) def test_temperatur_endpunkt_antwortet_immer(self): """Auf einem Nicht-Pi gibt es die Datei nicht -- dann 'N/A', aber kein Fehler. Der Endpunkt ist die einzige Gesundheitspruefung, die den Stream nicht anfasst.""" with self.hole("/temp") as r: self.assertEqual(r.status, 200) self.assertTrue(r.read().strip()) def test_gesund_meldet_erst_mit_bild(self): with self.assertRaises(urllib.error.HTTPError) as fehler: self.hole("/gesund") self.assertEqual(fehler.exception.code, 503) self.schiebe() with self.hole("/gesund") as r: self.assertEqual(r.status, 200) def test_gesund_ist_auch_unter_health_erreichbar(self): """Der englische Name steht in fremden Werkzeugen (Uptime-Kuma und Konsorten) -- beide Pfade muessen dasselbe tun.""" self.schiebe() with self.hole("/health") as r: self.assertEqual(r.status, 200) def test_gesund_merkt_ein_altes_bild(self): """Der Server antwortet noch, die Aufnahme haengt. Genau das ist der Fall, den ein Ping nie sieht.""" vogelkamera.frame_buffer.update(BILD) with vogelkamera.frame_buffer._lock: vogelkamera.frame_buffer._timestamp = time.time() - 30 with self.assertRaises(urllib.error.HTTPError) as fehler: self.hole("/gesund") self.assertEqual(fehler.exception.code, 503) def test_schnappschuss_gibt_das_bild_her(self): """Ein einzelnes Bild statt eines Streams -- der schonende Weg, einmal nachzusehen, was die Kamera sieht.""" self.schiebe() with self.hole("/snapshot") as r: self.assertEqual(r.status, 200) self.assertEqual(r.headers["Content-Type"], "image/jpeg") self.assertEqual(r.read(), BILD) def test_schnappschuss_ohne_bild_ist_503(self): with self.assertRaises(urllib.error.HTTPError) as fehler: self.hole("/snapshot") self.assertEqual(fehler.exception.code, 503) def test_unbekannter_pfad_ist_404(self): with self.assertRaises(urllib.error.HTTPError) as fehler: self.hole("/gibtsnicht") self.assertEqual(fehler.exception.code, 404) # -- der Stream -------------------------------------------------------- def test_stream_liefert_multipart_mit_bildern(self): self.speisen() with self.hole("/stream") as r: self.assertEqual(r.status, 200) self.assertIn("multipart/x-mixed-replace", r.headers["Content-Type"]) anfang = r.read(len(BILD) + 200) self.assertIn(b"--frame", anfang) self.assertIn(b"\xff\xd8", anfang) def test_zweiter_zuschauer_wird_abgewiesen(self): """Der eigentliche Punkt: lieber eine klare Absage als zwei ruckelnde Bilder und ein Detector in Timeouts. ⚠️ Dieser Test war der Grund, die Datei wieder anzufassen -- er stand auf der alten Fassung und hat deshalb nicht gemeldet, dass die Absage im Code fehlte, obwohl das README sie zusagte.""" self.speisen() erster = self.hole("/stream") self.addCleanup(erster.close) erster.read(len(BILD)) # sicherstellen, dass er wirklich haengt with self.assertRaises(urllib.error.HTTPError) as fehler: self.hole("/stream") self.assertEqual(fehler.exception.code, 503) self.assertIn("Zuschauer", fehler.exception.read().decode("utf-8")) def test_nach_dem_trennen_darf_der_naechste_wieder(self): """Ohne das waere die Kamera nach einem Browserfenster dauerhaft dicht.""" self.speisen() erster = self.hole("/stream") erster.read(len(BILD)) erster.close() # Der Server merkt das Trennen erst beim naechsten Schreibversuch. for _ in range(250): if vogelkamera.zuschauer_platz.acquire(blocking=False): vogelkamera.zuschauer_platz.release() break time.sleep(0.02) else: self.fail("Der Zuschauerplatz wurde nach dem Trennen nie frei") zweiter = self.hole("/stream") self.addCleanup(zweiter.close) self.assertEqual(zweiter.status, 200) def test_stehende_kamera_gibt_den_platz_wieder_frei(self): """🔑 Die Falle, die der Zuschauerplatz selbst aufstellt. Ohne neues Bild schreibt der Server nichts, merkt also auch nicht, dass der Zuschauer weg ist -- und haelt den einen Platz fuer immer. Der Detector bekaeme bei jedem Versuch 503, die Kamera waere dauerhaft stumm. Deshalb endet ein Stream ohne Bild nach ``STREAM_LEERLAUF``.""" vogelkamera.STREAM_LEERLAUF = 0.3 self.addCleanup(setattr, vogelkamera, "STREAM_LEERLAUF", 15.0) self.schiebe() # ein Bild, danach bleibt die Kamera stehen erster = self.hole("/stream") self.addCleanup(erster.close) erster.read(len(BILD)) for _ in range(100): if vogelkamera.zuschauer_platz.acquire(blocking=False): vogelkamera.zuschauer_platz.release() break time.sleep(0.02) else: self.fail("Der Platz blieb an einer stehenden Kamera haengen") zweiter = self.hole("/stream") self.addCleanup(zweiter.close) self.assertEqual(zweiter.status, 200) # -- /status ---------------------------------------------------------- def test_status_antwortet_ohne_bild(self): """Genau dann braucht man ihn: die Kamera liefert nichts und man will wissen, warum -- ohne den Stream anzufassen.""" with self.hole("/status") as r: self.assertEqual(r.status, 200) self.assertEqual(r.headers["Content-Type"], "application/json") d = json.loads(r.read()) self.assertEqual(d["kamera"], "fehlt") self.assertEqual(d["fehler"], "0x1") self.assertEqual(d["bilder"], 0) self.assertEqual(d["letztes_bild_s"], -1.0) self.assertFalse(d["zuschauer"]) def test_status_zaehlt_bilder_auch_ohne_zuschauer(self): """Die Kamera liefert ins Leere, wenn niemand zuschaut -- /status muss trotzdem sagen koennen, dass sie arbeitet.""" for _ in range(3): self.schiebe() with self.hole("/status") as r: d = json.loads(r.read()) self.assertEqual(d["kamera"], "bereit") self.assertEqual(d["fehler"], "0x0") self.assertEqual(d["bilder"], 3) self.assertGreaterEqual(d["letztes_bild_s"], 0.0) self.assertFalse(d["zuschauer"]) def test_status_hat_die_felder_der_esp32_seite(self): """Gleiches Schema auf beiden Kameraseiten, sonst braucht jedes Werkzeug zwei Faelle. ``psram_frei`` fehlt absichtlich -- das gibt es nur auf dem ESP32.""" self.schiebe() with self.hole("/status") as r: d = json.loads(r.read()) for feld in ("kamera", "fehler", "laufzeit_s", "bilder", "fps", "letztes_bild_s", "zuschauer", "rssi", "temperatur_c", "speicher_frei"): self.assertIn(feld, d) def test_status_sieht_den_zuschauer(self): self.speisen() strom = self.hole("/stream") self.addCleanup(strom.close) strom.read(len(BILD)) with self.hole("/status") as r: self.assertTrue(json.loads(r.read())["zuschauer"]) def test_status_geht_waehrend_der_stream_laeuft(self): """Der Stream haelt seinen Socket dauerhaft. Waere der Server dann blockiert, waere /status nutzlos -- gefragt ist er ja im Betrieb.""" self.speisen() strom = self.hole("/stream") self.addCleanup(strom.close) strom.read(len(BILD)) with self.hole("/status", timeout=5.0) as r: self.assertEqual(r.status, 200) class MesswerteTests(unittest.TestCase): """Die drei Ablesungen aus /proc und /sys. Auf einem Nicht-Pi fehlen sie teils -- dann ist None richtig und ein Absturz falsch.""" def test_temperatur_ist_zahl_oder_none(self): c = vogelkamera.read_cpu_temp() self.assertTrue(c is None or isinstance(c, float)) def test_funkstaerke_ist_zahl_oder_none(self): r = vogelkamera.funkstaerke() self.assertTrue(r is None or isinstance(r, int)) def test_speicher_ist_zahl_oder_none(self): m = vogelkamera.speicher_frei() self.assertTrue(m is None or m > 0) def test_status_json_ueberlebt_fehlende_messwerte(self): """Auf einem Nicht-Pi fehlt alles, was aus /sys kommt. Trotzdem muss das JSON vollstaendig und serialisierbar sein -- die Bruecke nach Home Assistant liest es blind.""" vogelkamera.frame_buffer = vogelkamera.FrameBuffer() d = vogelkamera.status_json() json.dumps(d) # wirft, wenn ein Wert nicht serialisierbar ist self.assertEqual(d["kamera"], "fehlt") class FrameBufferTests(unittest.TestCase): def test_haelt_nur_das_neueste_bild(self): """Kein Rueckstau: wer sich spaet einklinkt, will das aktuelle Bild und nicht die Warteschlange von vor zehn Sekunden.""" p = vogelkamera.FrameBuffer() p.update(b"alt") p.update(b"neu") self.assertEqual(p.get()[0], b"neu") def test_leerer_puffer_hat_keinen_zeitstempel(self): """Die Null ist der Wert, an dem /status 'fehlt' erkennt -- ein geratener Zeitstempel wuerde eine stehende Kamera gesund aussehen lassen.""" p = vogelkamera.FrameBuffer() frame, ts = p.get() self.assertIsNone(frame) self.assertEqual(ts, 0) def test_datiert_jedes_bild(self): p = vogelkamera.FrameBuffer() p.update(b"eins") erster = p.get()[1] self.assertGreater(erster, 0) time.sleep(0.02) p.update(b"zwei") self.assertGreater(p.get()[1], erster) if __name__ == "__main__": unittest.main()