vogelalbum/tests/test_verteiler.py

"""Der Live-Verteiler.

Der Kern ist nicht der HTTP-Teil, sondern das Versprechen: die Erkennung darf
davon nichts merken. Deshalb pruefen die Tests vor allem, dass ohne Zuschauer
nicht kodiert wird und dass fertige JPEG-Bytes unveraendert durchlaufen.
"""

from __future__ import annotations

import json  # noqa: F401 - Symmetrie zu den anderen Testdateien
import threading
import unittest
import urllib.error
import urllib.request
from io import BytesIO

from PIL import Image

from detector.verteiler import Bildhalter, Handler, Verteiler

BILD = Image.new("RGB", (320, 180), (60, 110, 70))


def jpeg_bytes(bild: Image.Image = BILD) -> bytes:
    puffer = BytesIO()
    bild.save(puffer, format="JPEG")
    return puffer.getvalue()


class BildhalterTests(unittest.TestCase):
    def test_ohne_bild_kommt_nichts(self):
        self.assertIsNone(Bildhalter().jpeg())

    def test_fertige_bytes_laufen_unveraendert_durch(self):
        """Der haeufige Fall: die Quelle liefert MJPEG. Wird hier umkodiert,
        zahlt der Detector Rechenzeit fuer nichts."""
        roh = jpeg_bytes()
        h = Bildhalter()
        h.zeige(BILD, roh)

        nummer, daten = h.jpeg()
        self.assertEqual(nummer, 1)
        self.assertIs(daten, roh)

    def test_ohne_fertige_bytes_wird_kodiert(self):
        """RTSP und Bilderordner liefern kein fertiges JPEG — dann muss der
        Verteiler selbst eines machen."""
        h = Bildhalter()
        h.zeige(BILD, None)

        nummer, daten = h.jpeg()
        self.assertEqual(nummer, 1)
        self.assertEqual(daten[:2], b"\xff\xd8")
        self.assertEqual(Image.open(BytesIO(daten)).size, BILD.size)

    def test_kodiert_nur_einmal_je_bild(self):
        """Sonst zahlt jeder Zuschauer seine eigene Kodierung."""
        h = Bildhalter()
        h.zeige(BILD, None)

        erst = h.jpeg()
        nochmal = h.jpeg()
        self.assertIs(erst[1], nochmal[1])

    def test_neues_bild_wird_neu_kodiert(self):
        h = Bildhalter()
        h.zeige(BILD, None)
        erst = h.jpeg()

        h.zeige(Image.new("RGB", (320, 180), (200, 30, 30)), None)
        neu = h.jpeg()

        self.assertEqual(neu[0], erst[0] + 1)
        self.assertIsNot(neu[1], erst[1])

    def test_zeige_kodiert_nicht(self):
        """Das Versprechen an die Erkennung: anmelden kostet Zeigerarbeit.
        Solange niemand abholt, entsteht kein JPEG."""
        h = Bildhalter()
        for _ in range(50):
            h.zeige(BILD, None)
        self.assertIsNone(h._kodiert)
        self.assertEqual(h.nummer, 50)

    def test_grosse_bilder_werden_verkleinert(self):
        h = Bildhalter(max_breite=160)
        h.zeige(BILD, None)
        self.assertEqual(Image.open(BytesIO(h.jpeg()[1])).width, 160)

    def test_warte_laeuft_ohne_bild_ab(self):
        h = Bildhalter()
        self.assertEqual(h.warte(0, timeout=0.05), 0)

    def test_warte_kehrt_mit_neuem_bild_zurueck(self):
        h = Bildhalter()
        threading.Timer(0.05, lambda: h.zeige(BILD, None)).start()
        self.assertEqual(h.warte(0, timeout=5.0), 1)

    def test_warte_kehrt_sofort_zurueck_wenn_schon_neu(self):
        h = Bildhalter()
        h.zeige(BILD, None)
        self.assertEqual(h.warte(0, timeout=5.0), 1)


class VerteilerTests(unittest.TestCase):
    def setUp(self):
        Handler._zuschauer = 0
        self.v = Verteiler(port=0, bind="127.0.0.1")
        self.v.start()
        self.addCleanup(self.v.stop)

    def url(self, pfad: str) -> str:
        return f"http://127.0.0.1:{self.v.port}{pfad}"

    def hole(self, pfad: str, timeout: float = 5.0):
        return urllib.request.urlopen(self.url(pfad), timeout=timeout)

    def test_seite_kommt(self):
        with self.hole("/") as r:
            self.assertEqual(r.status, 200)
            self.assertIn(b"/live", r.read())

    def test_einzelbild_erst_mit_bild(self):
        with self.assertRaises(urllib.error.HTTPError) as fehler:
            self.hole("/bild.jpg")
        self.assertEqual(fehler.exception.code, 503)

        self.v.zeige(BILD, jpeg_bytes())
        with self.hole("/bild.jpg") as r:
            self.assertEqual(r.status, 200)
            self.assertEqual(r.headers["Content-Type"], "image/jpeg")
            self.assertEqual(r.read()[:2], b"\xff\xd8")

    def test_unbekannter_pfad_ist_404(self):
        with self.assertRaises(urllib.error.HTTPError) as fehler:
            self.hole("/gibtsnicht")
        self.assertEqual(fehler.exception.code, 404)

    def test_stream_liefert_multipart(self):
        stop = threading.Event()

        def speisen():
            while not stop.is_set():
                self.v.zeige(BILD, jpeg_bytes())
                stop.wait(0.01)

        threading.Thread(target=speisen, daemon=True).start()
        self.addCleanup(stop.set)

        strom = self.hole("/live")
        self.addCleanup(strom.close)
        self.assertIn("multipart/x-mixed-replace", strom.headers["Content-Type"])
        anfang = strom.read(400)
        self.assertIn(b"--BILDGRENZE", anfang)
        self.assertIn(b"\xff\xd8", anfang)

    def test_zwei_zuschauer_sind_erlaubt(self):
        """Der ganze Zweck: an der Kamera geht nur einer, hier beliebig viele."""
        stop = threading.Event()

        def speisen():
            while not stop.is_set():
                self.v.zeige(BILD, jpeg_bytes())
                stop.wait(0.01)

        threading.Thread(target=speisen, daemon=True).start()
        self.addCleanup(stop.set)

        erster = self.hole("/live")
        self.addCleanup(erster.close)
        erster.read(200)

        zweiter = self.hole("/live")
        self.addCleanup(zweiter.close)
        self.assertEqual(zweiter.status, 200)
        zweiter.read(200)

        self.assertEqual(self.v.zuschauer, 2)

    def test_obergrenze_weist_ab(self):
        stop = threading.Event()

        def speisen():
            while not stop.is_set():
                self.v.zeige(BILD, jpeg_bytes())
                stop.wait(0.01)

        threading.Thread(target=speisen, daemon=True).start()
        self.addCleanup(stop.set)

        Handler.max_zuschauer = 1
        self.addCleanup(setattr, Handler, "max_zuschauer", 8)

        erster = self.hole("/live")
        self.addCleanup(erster.close)
        erster.read(200)

        with self.assertRaises(urllib.error.HTTPError) as fehler:
            self.hole("/live")
        self.assertEqual(fehler.exception.code, 503)


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