test_watch_identity.py
Rama main · réplica de lectura del repositorio original.
"""Surveillance et préférences : deux homonymes ne sont pas une caméra."""
import asyncio
import contextlib
import datetime as dt
import io
import json
import os
from pathlib import Path
import tempfile
from types import SimpleNamespace
import unittest
from unittest import mock
os.environ["BLINK_BOOTSTRAP"] = "none"
_TEST_HOME = tempfile.TemporaryDirectory(prefix="blink-watch-identite-import-")
os.environ["BLINK_HOME"] = _TEST_HOME.name
import runtime
import serve
import watch
class TestIdentitesSurveillance(unittest.TestCase):
def setUp(self):
temporaire = tempfile.TemporaryDirectory(prefix="blink-watch-identite-")
self.addCleanup(temporaire.cleanup)
self.base = Path(temporaire.name)
self.patches = contextlib.ExitStack()
self.addCleanup(self.patches.close)
self.patches.enter_context(mock.patch.object(watch, "BASE_DIR", self.base))
self.patches.enter_context(
mock.patch.object(watch, "CLIPS", self.base / "Blink_Clips"))
self.patches.enter_context(mock.patch.object(runtime, "lire_langue", return_value="fr"))
self.blink = SimpleNamespace(
refresh=mock.AsyncMock(),
homescreen={"cameras": [
{"name": "Jardin", "network_id": 111, "id": 1, "status": "offline"},
{"name": "Jardin", "network_id": 222, "id": 2, "status": "online"},
]},
sync={
"Maison": SimpleNamespace(arm=True, network_id=111, cameras={
"Jardin": SimpleNamespace(motion_enabled=True, attributes={"battery": "low"})}),
"Annexe": SimpleNamespace(arm=True, network_id=222, cameras={
"Jardin": SimpleNamespace(motion_enabled=True, attributes={"battery": "ok"})}),
})
def lire(self):
@contextlib.asynccontextmanager
async def session():
yield object()
with mock.patch.object(watch.blink_auth, "session_http_temporaire", new=session), \
mock.patch.object(watch.blink_auth, "connect_saved",
new=mock.AsyncMock(return_value=self.blink)):
return asyncio.run(watch.read_state(dt.timezone.utc))
@staticmethod
def libelle(etat, appareil):
return next(nom for nom, camera in etat["cameras"].items()
if camera["device_id"] == str(appareil))
def test_homonymes_reseaux_distincts_alertent_et_ne_dependent_pas_de_l_ordre(self):
courant = self.lire()
self.assertEqual(len(courant["cameras"]), 2)
nom = self.libelle(courant, 1)
self.assertIn("111", nom)
self.assertFalse(courant["cameras"][nom]["online"])
alertes, retours = watch.compare({}, courant, dt.timezone.utc, set())
self.assertEqual(len(alertes), 2)
self.assertTrue(all(nom in alerte for alerte in alertes))
self.assertEqual(retours, [])
self.blink.homescreen["cameras"].reverse()
self.blink.sync = dict(reversed(list(self.blink.sync.items())))
suivant = self.lire()
self.assertEqual(courant["cameras"], suivant["cameras"])
self.assertEqual(watch.compare(courant, suivant, dt.timezone.utc, set()), ([], []))
def test_homonymes_meme_reseau_meme_si_blinkpy_a_deja_ecrase_un_objet(self):
self.blink.homescreen["cameras"][1]["network_id"] = 111
for info in self.blink.homescreen["cameras"]:
info["enabled"] = True
info["battery"] = "low" if info["id"] == 1 else "ok"
self.blink.sync.pop("Annexe")
camera = self.blink.sync["Maison"].cameras["Jardin"]
camera.camera_id = 2
camera.attributes["battery"] = "ok"
courant = self.lire()
self.assertEqual(len(courant["cameras"]), 2)
self.assertNotEqual(self.libelle(courant, 1), self.libelle(courant, 2))
alertes, _ = watch.compare({}, courant, dt.timezone.utc, set())
self.assertEqual(len(alertes), 2)
self.assertTrue(all(self.libelle(courant, 1) in a for a in alertes))
def test_nom_unique_inchange_et_suivi_par_identifiant_quand_homonyme_disparait(self):
ancien = self.lire()
self.blink.homescreen["cameras"].pop()
self.blink.sync.pop("Annexe")
courant = self.lire()
self.assertEqual(list(courant["cameras"]), ["Jardin"])
self.assertEqual(watch.compare(ancien, courant, dt.timezone.utc, set()), ([], []))
self.assertEqual(watch.normaliser_sourdines(
{self.libelle(ancien, 1)}, courant["cameras"], ancien["cameras"]), {"Jardin"})
def test_ancien_etat_ambigu_ne_masque_pas_la_premiere_alerte_reelle(self):
courant = self.lire()
ancien = {"cameras": {"Jardin": {
"online": False, "armed": True, "battery": "low", "system_armed": True}}}
alertes, retours = watch.compare(ancien, courant, dt.timezone.utc, set())
self.assertEqual(len(alertes), 2)
self.assertEqual(retours, [])
def test_ancien_etat_unique_reste_compatible(self):
self.blink.homescreen["cameras"].pop()
self.blink.sync.pop("Annexe")
courant = self.lire()
ancien = {"cameras": {"Jardin": {
"online": False, "armed": True, "battery": "low", "system_armed": True}}}
self.assertEqual(watch.compare(ancien, courant, dt.timezone.utc, set()), ([], []))
def test_ancien_nom_en_sourdine_se_migre_en_deux_choix_independants(self):
courant = self.lire()
ignores = watch.normaliser_sourdines({"Jardin"}, courant["cameras"])
self.assertEqual(ignores, set(courant["cameras"]))
ignores -= {self.libelle(courant, 1)}
alertes, _ = watch.compare({}, courant, dt.timezone.utc, ignores)
self.assertEqual(len(alertes), 2)
self.assertEqual(watch.compare({}, courant, dt.timezone.utc, {"Jardin"}), ([], []))
def test_migration_persistante_puis_reactivation_d_un_seul_homonyme(self):
courant = self.lire()
fichier = self.base / "watch_state.json"
fichier.write_text(json.dumps({"cameras": {"Jardin": {}},
"ignored": ["Jardin"]}), encoding="utf-8")
args = SimpleNamespace(dry_run=False, timezone="UTC", loop=None, test=False,
ignore=[], unignore=[self.libelle(courant, 1)],
ignore_module=[], unignore_module=[])
with mock.patch.object(watch, "WATCH_STATE", fichier), \
mock.patch.object(watch, "read_state", new=mock.AsyncMock(return_value=courant)), \
mock.patch.object(runtime, "verrou"), \
mock.patch.object(watch, "journal"), \
mock.patch.object(watch, "popup"), \
mock.patch.object(watch, "toast"), \
mock.patch.object(watch, "parse_args", return_value=args), \
mock.patch.object(runtime, "repeter", return_value=0), \
contextlib.redirect_stdout(io.StringIO()):
watch._controler(args, dt.timezone.utc)
self.assertEqual(json.loads(fichier.read_text(encoding="utf-8"))["ignored"],
sorted(courant["cameras"]))
self.assertEqual(watch.main(), 0)
enregistre = json.loads(fichier.read_text(encoding="utf-8"))
self.assertEqual(enregistre["ignored"], [self.libelle(courant, 2)])
self.assertEqual(enregistre["cameras"], courant["cameras"])
def test_activite_d_un_homonyme_ne_masque_pas_le_silence_de_l_autre(self):
self.blink.homescreen["cameras"][0]["status"] = "online"
self.blink.sync["Maison"].cameras["Jardin"].attributes["battery"] = "ok"
maintenant = dt.datetime.now(dt.timezone.utc)
vieux = (maintenant - dt.timedelta(days=5)).isoformat()
recent = (maintenant - dt.timedelta(hours=1)).isoformat()
registre = self.base / "Blink_Clips" / watch.md.DOWNLOAD_STATE
registre.parent.mkdir()
registre.write_text(json.dumps({"clips": {
"premier": {"camera": "Jardin", "network_id": "111", "device_id": "1",
"created_at": vieux, "excluded": True},
"second": {"camera": "Jardin", "network_id": "222", "device_id": "2",
"created_at": recent},
"ancien_sans_identite": {"camera": "Jardin", "created_at": recent},
}}), encoding="utf-8")
courant = self.lire()
nom = self.libelle(courant, 1)
self.assertEqual(courant["last_clip"][nom], vieux)
self.assertEqual(courant["last_clip"][self.libelle(courant, 2)], recent)
alertes, _ = watch.compare({}, courant, dt.timezone.utc, set())
self.assertEqual(len(alertes), 1)
self.assertIn(nom, alertes[0])
self.assertIn("aucun clip", alertes[0])
def test_nom_deja_suffixe_par_utilisateur_ne_peut_ecraser_un_homonyme(self):
nom = self.libelle(self.lire(), 1)
etats = list(self.lire()["cameras"].values())
etats.append({**etats[1], "name": nom, "device_id": "3"})
cameras = watch._libelles_cameras(etats)
self.assertEqual(len(cameras), 3)
self.assertEqual(cameras[nom]["device_id"], "3")
class TestAPIIdentitesSurveillance(unittest.TestCase):
def setUp(self):
self.cameras = watch._libelles_cameras([
{"name": "Jardin", "network_id": "111", "device_id": "1"},
{"name": "Jardin", "network_id": "222", "device_id": "2"},
])
self.noms = list(self.cameras)
self.entrees = {
"premier": {"camera": "Jardin", "network_id": "111", "device_id": "1"},
"second": {"camera": "Jardin", "network_id": "222", "device_id": "2"},
"usb": {"camera": "Jardin", "network_id": "111", "sync_id": "10"},
"ambigu": {"camera": "Jardin"},
}
self.cles = {cle: serve.blink_registre.camera_setting_key_from_entry(entree)
for cle, entree in self.entrees.items()}
self.actives = {self.cles["premier"], self.cles["second"], self.cles["usb"]}
self.etat = {"cameras": self.cameras, "ignored": ["Jardin"]}
self.patches = contextlib.ExitStack()
self.addCleanup(self.patches.close)
self.patches.enter_context(mock.patch.object(serve, "read_entries", return_value=self.entrees))
self.patches.enter_context(mock.patch.object(watch.md, "load_json", return_value=self.etat))
self.patches.enter_context(mock.patch.object(runtime, "lire_suppression_auto",
side_effect=lambda: set(self.actives)))
self.patches.enter_context(mock.patch.object(runtime, "ecrire_suppression_auto",
side_effect=self.enregistrer))
self.patches.enter_context(mock.patch.object(runtime, "verrou_configuration"))
def enregistrer(self, cles):
self.actives = set(cles)
def requete(self, route, payload=None):
handler = serve.Handler.__new__(serve.Handler)
handler.path, handler.paths = route, {}
handler.headers = {"Host": "127.0.0.1", "X-Blink-Token": serve.TOKEN}
reponses = []
handler.send_json = lambda valeur, code=200: reponses.append((code, valeur))
if payload is None:
handler.do_GET()
else:
corps = json.dumps(payload).encode("utf-8")
handler.headers["Content-Length"] = str(len(corps))
handler.rfile = io.BytesIO(corps)
handler.do_POST()
self.assertEqual(len(reponses), 1)
return reponses[0]
def test_sourdine_affiche_deux_appareils_sans_troisieme_nom_ambigu(self):
code, etat = self.requete("/api/sourdine")
self.assertEqual(code, 200)
self.assertEqual(etat["cameras"], self.noms)
self.assertEqual(etat["ignored"], self.noms)
def test_suppression_auto_desactive_un_seul_homonyme_cloud_et_usb(self):
code, _ = self.requete("/api/suppression-auto",
{"camera": self.noms[0], "actif": False})
self.assertEqual(code, 200)
self.assertEqual(self.actives, {self.cles["second"]})
_, etat = self.requete("/api/suppression-auto")
self.assertEqual(etat["cameras"], [{"name": nom} for nom in self.noms])
self.assertEqual(etat["actives"], [self.noms[1]])
def test_activation_ne_touche_pas_l_homonyme_ni_les_anciens_clips_ambigus(self):
self.actives = set()
code, _ = self.requete("/api/suppression-auto",
{"camera": self.noms[0], "actif": True})
self.assertEqual(code, 200)
self.assertEqual(self.actives, {self.cles["premier"], self.cles["usb"]})
def test_ancien_nom_ambigu_ne_peut_activer_les_deux_appareils(self):
self.actives = set()
code, _ = self.requete("/api/suppression-auto", {"camera": "Jardin", "actif": True})
self.assertEqual(code, 400)
self.assertEqual(self.actives, set())
def test_get_suppression_auto_resout_le_registre_une_seule_fois(self):
with mock.patch.object(watch, "_cameras_correspondantes",
wraps=watch._cameras_correspondantes) as rapprocher:
code, etat = self.requete("/api/suppression-auto")
self.assertEqual(code, 200)
self.assertEqual(etat["actives"], self.noms)
self.assertEqual(rapprocher.call_count, len(self.entrees))
def test_get_suppression_auto_reflete_le_registre_courant(self):
_, etat = self.requete("/api/suppression-auto")
self.assertEqual(etat["actives"], self.noms)
del self.entrees["second"]
_, suivant = self.requete("/api/suppression-auto")
self.assertEqual(suivant["cameras"], etat["cameras"])
self.assertEqual(suivant["actives"], [self.noms[0]])
if __name__ == "__main__":
unittest.main()