test_serve_camera_wakeup.py
Rama main · réplica de lectura del repositorio original.
"""Le reveil partage le module avec le direct et confirme la commande Blink."""
from __future__ import annotations
import asyncio
import io
import json
import os
import re
from pathlib import Path
import tempfile
import threading
import unittest
from types import SimpleNamespace
from unittest import mock
os.environ["BLINK_BOOTSTRAP"] = "none"
_TEST_HOME = tempfile.TemporaryDirectory(prefix="blink-camera-wakeup-")
os.environ["BLINK_HOME"] = _TEST_HOME.name
import serve # noqa: E402 - environnement isole avant import
from blinkpy import api # noqa: E402
class ReveilCameraTests(unittest.TestCase):
def setUp(self):
self.handler = serve.Handler.__new__(serve.Handler)
self.slot = threading.BoundedSemaphore(1)
self.info = {}
self.blink = object()
self.camera = SimpleNamespace(
network_id=42,
snap_picture=mock.AsyncMock(return_value={"id": 123, "network_id": 42}),
)
self.status = mock.AsyncMock(return_value={"status_code": 908, "complete": True})
self.disk = tempfile.TemporaryDirectory(prefix="blink-wakeup-lock-")
self.addCleanup(self.disk.cleanup)
self.root = Path(self.disk.name)
def reserver(_owner, **_kwargs):
return serve.runtime.verrou("hub", "reveil", racine=self.root)
def appeler(factory, timeout):
return asyncio.run(factory(self.blink))
self.call = mock.Mock(side_effect=appeler)
self.hub_lock = mock.Mock(side_effect=reserver)
for patch in (
mock.patch.object(serve, "MODULE_SLOT", self.slot),
mock.patch.object(serve, "MODULE_SLOT_INFO", self.info),
mock.patch.object(serve.blink_engine, "hub_lock", self.hub_lock),
mock.patch.object(serve.BLINK, "call", self.call),
mock.patch.object(serve.BLINK, "find_camera", return_value=(None, self.camera)),
mock.patch.object(api, "request_command_status", self.status),
):
patch.start()
self.addCleanup(patch.stop)
def verifier_liberation(self):
self.assertEqual(self.info, {})
self.assertTrue(self.slot.acquire(blocking=False), "slot memoire encore occupe")
self.slot.release()
with serve.runtime.verrou("hub", "verification", racine=self.root):
pass
def test_direct_actif_refuse_le_reveil_sans_appeler_blink(self):
self.slot.acquire()
self.info.update({"quoi": "direct WebRTC", "camera": "Salon"})
try:
with self.assertRaisesRegex(serve.blink_engine.BusyError, "Salon"):
self.handler.reveiller_camera("Jardin")
self.call.assert_not_called()
self.hub_lock.assert_not_called()
self.assertEqual(self.info["camera"], "Salon")
self.assertFalse(self.slot.acquire(blocking=False))
finally:
self.slot.release()
def test_telechargement_actif_refuse_le_reveil_et_rend_le_slot(self):
with serve.runtime.verrou("hub", "telechargement", racine=self.root):
with self.assertRaises(serve.blink_engine.BusyError):
self.handler.reveiller_camera("Jardin")
self.call.assert_not_called()
self.verifier_liberation()
def test_une_photo_et_confirmation_gardent_les_deux_verrous(self):
async def confirmer(*_args):
self.assertFalse(self.slot.acquire(blocking=False))
self.assertEqual(self.info["camera"], "Salon")
with self.assertRaises(serve.runtime.BusyError):
with serve.runtime.verrou("hub", "telechargement", racine=self.root):
self.fail("Le telechargement a pu prendre le module pendant le reveil")
return {"status_code": 908, "complete": True}
self.status.side_effect = confirmer
self.handler.reveiller_camera("Salon")
self.camera.snap_picture.assert_awaited_once_with()
self.status.assert_awaited_once_with(self.blink, 42, 123)
self.hub_lock.assert_called_once_with("reveil", attente=serve.ATTENTE_HUB_MAX_SECONDS)
self.verifier_liberation()
def test_reponse_refusee_ne_devient_pas_un_succes(self):
for reponse in (None, False, {}, {"message": "System is busy"}):
with self.subTest(reponse=reponse):
self.camera.snap_picture.return_value = reponse
with self.assertRaisesRegex(RuntimeError, re.escape(serve.msg("reveil_refuse"))):
self.handler.reveiller_camera("Salon")
self.verifier_liberation()
self.status.assert_not_awaited()
def test_commande_creee_mais_non_confirmee_ne_devient_pas_un_succes(self):
for statut in (None, {}, {"status_code": 908, "complete": False},
{"status_code": 101, "complete": True}):
with self.subTest(statut=statut):
self.status.return_value = statut
with self.assertRaisesRegex(RuntimeError,
re.escape(serve.msg("reveil_non_confirme"))):
self.handler.reveiller_camera("Salon")
self.verifier_liberation()
def test_exception_ou_annulation_rend_les_deux_verrous(self):
for erreur in (OSError("reseau coupe"), asyncio.CancelledError()):
with self.subTest(erreur=type(erreur).__name__):
self.camera.snap_picture.side_effect = erreur
with self.assertRaises(type(erreur)):
self.handler.reveiller_camera("Salon")
self.verifier_liberation()
def test_erreur_liberation_disque_ne_fait_pas_fuir_le_slot(self):
verrou = mock.MagicMock()
verrou.__exit__.side_effect = OSError("verrou illisible")
self.hub_lock.side_effect = None
self.hub_lock.return_value = verrou
with self.assertRaisesRegex(OSError, "verrou illisible"):
self.handler.reveiller_camera("Salon")
self.verifier_liberation()
def test_route_distingue_module_occupe_et_refus_blink(self):
for erreur, code in ((serve.blink_engine.BusyError("occupe"), 409),
(RuntimeError("refus Blink"), 503)):
with self.subTest(code=code):
body = json.dumps({"name": "Salon"}).encode()
self.handler.path = "/api/reveiller"
self.handler.headers = {"Content-Length": str(len(body))}
self.handler.rfile = io.BytesIO(body)
self.handler.hote_autorise = lambda: True
self.handler.jeton_valide = lambda: True
self.handler.reveiller_camera = mock.Mock(side_effect=erreur)
self.handler.system_state = mock.Mock()
self.handler.send_json = mock.Mock()
self.handler.do_POST()
self.assertEqual(self.handler.send_json.call_args.args[1], code)
self.handler.system_state.assert_not_called()
if __name__ == "__main__":
unittest.main()