test_serve_snapshot_webhook.py
Rama main · réplica de lectura del repositorio original.
"""Le webhook de snapshot (issue GitHub #9) a sa propre authentification,
independante de hote_autorise()/jeton_valide() (le modele du navigateur ne
s'applique pas a un appelant externe) : un secret persistant compare en
temps constant. Couvre aussi resolve_snapshot() (meme garde-fou anti-
traversee que resolve_media()), lister_snapshots() et la suppression."""
from __future__ import annotations
import io
import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest import mock
os.environ["BLINK_BOOTSTRAP"] = "none"
_TEST_HOME = tempfile.TemporaryDirectory(prefix="blink-snapshot-webhook-")
os.environ["BLINK_HOME"] = _TEST_HOME.name
import runtime # noqa: E402 - environnement isole avant import
import serve # noqa: E402
class JetonWebhookTests(unittest.TestCase):
def setUp(self):
self.dossier = tempfile.TemporaryDirectory(prefix="blink-jeton-webhook-")
self.addCleanup(self.dossier.cleanup)
self._ancien = os.environ.get("BLINK_HOME")
os.environ["BLINK_HOME"] = self.dossier.name
def tearDown(self):
if self._ancien is None:
os.environ.pop("BLINK_HOME", None)
else:
os.environ["BLINK_HOME"] = self._ancien
def test_genere_et_persiste_au_premier_appel(self):
jeton = runtime.lire_jeton_webhook()
self.assertGreaterEqual(len(jeton), 32)
self.assertEqual(runtime.lire_jeton_webhook(), jeton, "doit rester stable entre deux lectures")
def test_regenerer_invalide_l_ancien(self):
ancien = runtime.lire_jeton_webhook()
nouveau = runtime.regenerer_jeton_webhook()
self.assertNotEqual(ancien, nouveau)
self.assertEqual(runtime.lire_jeton_webhook(), nouveau)
class ResolveSnapshotTests(unittest.TestCase):
def setUp(self):
self.handler = serve.Handler.__new__(serve.Handler)
self.racine = Path(tempfile.mkdtemp(prefix="blink-resolve-snapshot-"))
self.addCleanup(lambda: __import__("shutil").rmtree(self.racine, ignore_errors=True))
self.handler.paths = {"snapshots": self.racine}
(self.racine / "Jardin").mkdir()
(self.racine / "Jardin" / "2026-09-14_10-00-00Z_abcd1234.jpg").write_bytes(b"photo")
def test_fichier_existant_resolu(self):
chemin = self.handler.resolve_snapshot("Jardin/2026-09-14_10-00-00Z_abcd1234.jpg")
self.assertIsNotNone(chemin)
self.assertTrue(chemin.is_file())
def test_traversee_de_chemin_refusee(self):
for tentative in ("../secret.jpg", "Jardin/../../secret.jpg",
"Jardin/../autre/x.jpg"):
with self.subTest(tentative=tentative):
self.assertIsNone(self.handler.resolve_snapshot(tentative))
def test_mauvaise_extension_refusee(self):
self.assertIsNone(self.handler.resolve_snapshot("Jardin/fichier.mp4"))
self.assertIsNone(self.handler.resolve_snapshot("Jardin/fichier.txt"))
def test_fichier_absent_renvoie_none(self):
self.assertIsNone(self.handler.resolve_snapshot("Jardin/absent.jpg"))
def test_forme_sans_sous_dossier_refusee(self):
self.assertIsNone(self.handler.resolve_snapshot("fichier.jpg"))
class ListerSnapshotsTests(unittest.TestCase):
def setUp(self):
self.handler = serve.Handler.__new__(serve.Handler)
self.racine = Path(tempfile.mkdtemp(prefix="blink-lister-snapshot-"))
self.addCleanup(lambda: __import__("shutil").rmtree(self.racine, ignore_errors=True))
self.handler.paths = {"snapshots": self.racine}
def _creer(self, camera: str, horodatage: str) -> None:
dossier = self.racine / camera
dossier.mkdir(parents=True, exist_ok=True)
(dossier / f"{horodatage}.jpg").write_bytes(b"photo")
def test_dossier_absent_renvoie_liste_vide(self):
self.handler.paths = {"snapshots": self.racine / "n-existe-pas"}
self.assertEqual(self.handler.lister_snapshots(), [])
def test_tri_du_plus_recent_au_plus_ancien_toutes_cameras_confondues(self):
self._creer("Jardin", "2026-09-14_08-00-00Z_aaaa")
self._creer("Salon", "2026-09-14_10-00-00Z_bbbb")
self._creer("Jardin", "2026-09-14_09-00-00Z_cccc")
horodatages = [entree["horodatage"] for entree in self.handler.lister_snapshots()]
self.assertEqual(horodatages, sorted(horodatages, reverse=True))
self.assertEqual(horodatages[0], "2026-09-14_10-00-00Z_bbbb")
def test_chemin_relatif_pret_pour_l_url(self):
self._creer("Jardin", "2026-09-14_08-00-00Z_aaaa")
entree = self.handler.lister_snapshots()[0]
self.assertEqual(entree["fichier"], "Jardin/2026-09-14_08-00-00Z_aaaa.jpg")
self.assertEqual(entree["camera"], "Jardin")
class WebhookAuthTests(unittest.TestCase):
def setUp(self):
self.dossier = tempfile.TemporaryDirectory(prefix="blink-webhook-auth-")
self.addCleanup(self.dossier.cleanup)
self._ancien = os.environ.get("BLINK_HOME")
os.environ["BLINK_HOME"] = self.dossier.name
self.jeton = runtime.regenerer_jeton_webhook()
self.handler = serve.Handler.__new__(serve.Handler)
self.handler.paths = {"snapshots": Path(self.dossier.name) / "Blink_Snapshots"}
self.handler.send_json = mock.Mock()
self.handler.send_error = mock.Mock()
self.declenche = mock.Mock(
return_value=self.handler.paths["snapshots"] / "Jardin" / "x.jpg")
self.handler.declencher_snapshot = self.declenche
def tearDown(self):
if self._ancien is None:
os.environ.pop("BLINK_HOME", None)
else:
os.environ["BLINK_HOME"] = self._ancien
def appeler(self, query: str):
self.handler.path = f"{serve.WEBHOOK_SNAPSHOT_ROUTE}?{query}"
self.handler.gerer_webhook_snapshot()
def test_jeton_correct_declenche_la_photo(self):
(self.handler.paths["snapshots"] / "Jardin").mkdir(parents=True)
(self.handler.paths["snapshots"] / "Jardin" / "x.jpg").write_bytes(b"photo")
self.appeler(f"camera=Jardin&token={self.jeton}")
self.declenche.assert_called_once_with("Jardin")
self.handler.send_error.assert_not_called()
self.assertTrue(self.handler.send_json.call_args.args[0]["ok"])
def test_jeton_absent_refuse_sans_toucher_a_blink(self):
self.appeler("camera=Jardin")
self.handler.send_error.assert_called_once_with(403)
self.declenche.assert_not_called()
def test_jeton_incorrect_refuse(self):
self.appeler("camera=Jardin&token=mauvais-jeton-devine")
self.handler.send_error.assert_called_once_with(403)
self.declenche.assert_not_called()
def test_jeton_correct_mais_camera_manquante(self):
self.appeler(f"token={self.jeton}")
self.declenche.assert_not_called()
self.assertEqual(self.handler.send_json.call_args.args[1], 400)
def test_camera_occupee_renvoie_409(self):
self.declenche.side_effect = serve.blink_engine.BusyError("occupe")
self.appeler(f"camera=Jardin&token={self.jeton}")
self.assertEqual(self.handler.send_json.call_args.args[1], 409)
def test_ne_depend_pas_de_hote_autorise_ni_jeton_session(self):
# Le point du webhook : un appelant qui echouerait les deux controles
# de session doit quand meme reussir avec le bon secret.
self.handler.hote_autorise = mock.Mock(return_value=False)
self.handler.jeton_valide = mock.Mock(return_value=False)
(self.handler.paths["snapshots"] / "Jardin").mkdir(parents=True)
(self.handler.paths["snapshots"] / "Jardin" / "x.jpg").write_bytes(b"photo")
self.appeler(f"camera=Jardin&token={self.jeton}")
self.handler.hote_autorise.assert_not_called()
self.handler.jeton_valide.assert_not_called()
self.handler.send_error.assert_not_called()
def test_do_get_route_vers_le_webhook_avant_hote_autorise(self):
self.handler.path = f"{serve.WEBHOOK_SNAPSHOT_ROUTE}?camera=Jardin&token={self.jeton}"
self.handler.hote_autorise = mock.Mock(return_value=False)
self.handler.gerer_webhook_snapshot = mock.Mock()
serve.Handler.do_GET(self.handler)
self.handler.gerer_webhook_snapshot.assert_called_once()
self.handler.hote_autorise.assert_not_called()
class SupprimerSnapshotTests(unittest.TestCase):
def setUp(self):
self.handler = serve.Handler.__new__(serve.Handler)
self.racine = Path(tempfile.mkdtemp(prefix="blink-supprimer-snapshot-"))
self.addCleanup(lambda: __import__("shutil").rmtree(self.racine, ignore_errors=True))
self.handler.paths = {"snapshots": self.racine}
(self.racine / "Jardin").mkdir()
self.fichier = self.racine / "Jardin" / "2026-09-14_10-00-00Z_abcd1234.jpg"
self.fichier.write_bytes(b"photo")
self.handler.hote_autorise = lambda: True
self.handler.jeton_valide = lambda: True
self.handler.send_json = mock.Mock()
self.handler.send_error = mock.Mock()
def poster(self, fichier: str):
corps = json.dumps({"fichier": fichier}).encode()
self.handler.path = "/api/snapshot-supprimer"
self.handler.headers = {"Content-Length": str(len(corps))}
self.handler.rfile = io.BytesIO(corps)
self.handler.do_POST()
def test_supprime_le_fichier_designe(self):
self.poster("Jardin/2026-09-14_10-00-00Z_abcd1234.jpg")
self.assertFalse(self.fichier.exists())
self.handler.send_json.assert_called_once_with({"ok": True})
def test_traversee_de_chemin_refusee_en_404(self):
self.poster("../../../../etc/passwd")
self.assertTrue(self.fichier.exists(), "le fichier legitime ne doit pas bouger")
self.handler.send_error.assert_called_once_with(404)
if __name__ == "__main__":
unittest.main()