"""Einstellungen aus der Web-App. Der Kern ist nicht der Abruf, sondern das Verhalten im Fehlerfall: eine unerreichbare Web-App darf die Erkennung nicht anhalten und nicht auf Vorgaben zurückwerfen. Wer zwei Wochen auf den ersten Vogel wartet, will ihn nicht verpassen, weil ein Webserver neu gestartet wurde. """ from __future__ import annotations import json import threading import unittest from http import server from detector.fernkonfig import Fernkonfig, arten_aus_json ANTWORT = { "kamera": "vogelhaus", "stream_url": "http://10.0.0.9:8080/stream", "habitats": ["garten"], "detect_threshold": 0.35, "min_bird_score": 0.07, "region": "de", "arten": [ {"scientific": "Parus major", "german": "Kohlmeise", "english": "Great Tit", "rarity": "haeufig", "lebensraum": "garten"}, {"scientific": "Turdus merula", "german": "Amsel", "english": "Blackbird", "rarity": "haeufig", "lebensraum": "garten"}, ], } class _Handler(server.BaseHTTPRequestHandler): antwort = ANTWORT status = 200 gesehene_token: list[str] = [] def log_message(self, fmt, *args): # noqa: A003 pass def do_GET(self): # noqa: N802 _Handler.gesehene_token.append(self.headers.get("Authorization", "")) koerper = json.dumps(self.antwort).encode("utf-8") self.send_response(self.status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(koerper))) self.end_headers() self.wfile.write(koerper) class ArtenTests(unittest.TestCase): def test_uebersetzt_die_felder(self): arten = arten_aus_json(ANTWORT["arten"]) self.assertEqual([a.scientific for a in arten], ["Parus major", "Turdus merula"]) self.assertEqual(arten[0].german, "Kohlmeise") self.assertEqual(arten[0].lebensraum, "garten") def test_ohne_wissenschaftlichen_namen_faellt_weg(self): """Der wissenschaftliche Name ist der Schlüssel ins Modell — ohne ihn trifft der Eintrag später leise nichts.""" arten = arten_aus_json([{"german": "Namenlos"}, {"scientific": " "}]) self.assertEqual(arten, []) def test_fehlende_beschriftung_wird_ersetzt(self): arten = arten_aus_json([{"scientific": "Pica pica"}]) self.assertEqual(arten[0].german, "Pica pica") self.assertEqual(arten[0].rarity, "haeufig") class AbrufTests(unittest.TestCase): def setUp(self): _Handler.antwort = ANTWORT _Handler.status = 200 _Handler.gesehene_token = [] self.srv = server.HTTPServer(("127.0.0.1", 0), _Handler) threading.Thread(target=self.srv.serve_forever, daemon=True).start() self.addCleanup(self.srv.server_close) self.addCleanup(self.srv.shutdown) self.api = f"http://127.0.0.1:{self.srv.server_address[1]}" def test_holt_und_merkt_sich(self): f = Fernkonfig(self.api, "geheim") daten = f.hole() self.assertEqual(daten["kamera"], "vogelhaus") self.assertEqual(f.letzte, daten) self.assertEqual(_Handler.gesehene_token, ["Bearer geheim"]) def test_faellig_erst_nach_dem_intervall(self): f = Fernkonfig(self.api, "geheim", intervall=300) self.assertTrue(f.faellig(1000.0)) f.hole(1000.0) self.assertFalse(f.faellig(1299.0)) self.assertTrue(f.faellig(1300.0)) def test_abgelehnter_token_gibt_none(self): _Handler.status = 401 f = Fernkonfig(self.api, "falsch") self.assertIsNone(f.hole()) def test_unerreichbar_gibt_none_und_behaelt_den_stand(self): """Das eigentliche Versprechen: ein Ausfall ändert nichts.""" f = Fernkonfig(self.api, "geheim") vorher = f.hole() self.assertIsNotNone(vorher) self.srv.shutdown() self.srv.server_close() self.assertIsNone(f.hole()) self.assertEqual(f.letzte, vorher) def test_unlesbare_antwort_gibt_none(self): _Handler.antwort = ["kein", "objekt"] f = Fernkonfig(self.api, "geheim") self.assertIsNone(f.hole()) if __name__ == "__main__": unittest.main()