test_diagnostic_xr.py
Rama main · réplica de lectura del repositorio original.
"""Rapport XR transmissible : lecture seule, résultat utile et données privées absentes."""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
import diagnostic_xr
class FauxSession:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return None
class DiagnosticXRTests(unittest.IsolatedAsyncioTestCase):
async def test_manifest_accessible_produit_un_rapport_anonyme(self):
storage = {"compatible": True, "enabled": True, "status": True}
sync = SimpleNamespace(
network_id=7,
sync_id=900,
serial="SECRET-SERIAL",
name="Jardin privé",
_local_storage=storage,
local_storage=True,
)
blink = SimpleNamespace(
homescreen={
"networks": [{"id": 7, "name": "Jardin privé"}],
"sync_modules": [{"id": 900, "serial": "SECRET-SERIAL"}],
"cameras": [],
"owls": [],
"doorbells": [{"name": "Porte secrète"}],
},
sync={"Porte secrète": SimpleNamespace()},
)
manifest_probe = mock.AsyncMock(return_value={
"clips": 2,
"schema_clips": 2,
"mapped_clips": 2,
"cameras": 1,
})
code, lines = await diagnostic_xr.inspect_blink(
blink,
select_modules=lambda _blink, _name: [("Jardin privé", sync)],
manifest_probe=manifest_probe,
)
report = "\n".join(lines)
self.assertEqual(code, 0)
self.assertIn("manifest API: OK", report)
self.assertIn("clips reported by API: 2", report)
self.assertIn("Overall result: PASS_WITH_CLIPS", report)
for private in ("Jardin privé", "Porte secrète", "SECRET-SERIAL"):
self.assertNotIn(private, report)
self.assertNotIn("900", report)
self.assertNotIn("network identifier: 7", report)
manifest_probe.assert_awaited_once_with(sync)
async def test_manifeste_vide_valide_la_route_sans_inventer_de_clips(self):
sync = SimpleNamespace(
network_id=7,
sync_id=900,
_local_storage={"compatible": True, "enabled": True},
local_storage=True,
)
code, lines = await diagnostic_xr.inspect_blink(
SimpleNamespace(homescreen={}, sync={}),
select_modules=lambda _blink, _name: [("secret", sync)],
manifest_probe=mock.AsyncMock(return_value={
"clips": 0,
"schema_clips": 0,
"mapped_clips": 0,
"cameras": 0,
}),
)
self.assertEqual(code, 0)
self.assertIn("PASS_EMPTY", "\n".join(lines))
async def test_reporteur_recoit_le_vrai_module_homescreen_pas_la_doorbell(self):
from test_xr_local_storage import compte_avec_stockage_local
blink, doorbell = compte_avec_stockage_local()
manifest_probe = mock.AsyncMock(return_value={
"clips": 0,
"schema_clips": 0,
"mapped_clips": 0,
"cameras": 0,
})
with mock.patch.object(
diagnostic_xr, "probe_local_manifest", new=manifest_probe,
):
code, lines = await diagnostic_xr.inspect_blink(blink)
selected = manifest_probe.await_args.args[0]
self.assertEqual(code, 0)
self.assertIsNot(selected, doorbell)
self.assertEqual(str(selected.sync_id), "900")
report = "\n".join(lines)
self.assertNotIn("Jardin", report)
self.assertNotIn("Porte d'entrée", report)
self.assertNotIn("900", report)
async def test_aucun_module_est_un_echec_lisible(self):
blink = SimpleNamespace(homescreen={}, sync={})
code, lines = await diagnostic_xr.inspect_blink(
blink,
select_modules=lambda _blink, _name: [],
manifest_probe=mock.AsyncMock(),
)
self.assertEqual(code, 1)
self.assertIn("no Sync Module", "\n".join(lines))
async def test_session_absente_ne_demande_jamais_des_identifiants(self):
async def connector_with_noisy_output(_session):
print("private@example.test SECRET_AUTH_TOKEN_123456789")
return None
connector = mock.AsyncMock(side_effect=connector_with_noisy_output)
code, report = await diagnostic_xr.collect_report(
session_factory=FauxSession,
connector=connector,
)
self.assertEqual(code, 2)
self.assertIn("LOGIN REQUIRED", report)
self.assertNotIn("private@example.test", report)
self.assertNotIn("SECRET_AUTH_TOKEN", report)
connector.assert_awaited_once()
async def test_erreur_est_redigee_avant_le_rapport(self):
message = (
"failure for jane@example.com at C:\\Users\\Jane\\blink_auth.json "
"https://example.test/path?token=secret "
"abcdefghijklmnopqrstuvwxyz0123456789"
)
blink = SimpleNamespace(homescreen={}, sync={})
code, lines = await diagnostic_xr.inspect_blink(
blink,
select_modules=mock.Mock(side_effect=RuntimeError(message)),
manifest_probe=mock.AsyncMock(),
)
report = "\n".join(lines)
self.assertEqual(code, 1)
self.assertIn("reference", report)
self.assertNotIn("jane@example.com", report)
self.assertNotIn("blink_auth.json", report)
self.assertNotIn("example.test", report)
self.assertNotIn("abcdefghijklmnopqrstuvwxyz", report)
async def test_xr_actif_est_sonde_meme_si_compatible_vaut_faux(self):
poll = mock.AsyncMock(side_effect=[
{"id": 51},
{
"manifest_id": 61,
"clips": [{
"id": 71,
"camera_name": "CameraAnonyme",
"created_at": "2026-08-31T10:00:00+00:00",
"size": "128",
}],
},
])
sync = SimpleNamespace(
network_id=7,
sync_id=900,
_local_storage={
"compatible": False,
"enabled": True,
"status": True,
},
local_storage=True,
_names_table={"CameraAnonyme": "Nom privé"},
poll_local_storage_manifest=poll,
)
code, lines = await diagnostic_xr.inspect_blink(
SimpleNamespace(homescreen={}, sync={}),
select_modules=lambda _blink, _name: [("secret", sync)],
)
report = "\n".join(lines)
self.assertEqual(code, 0)
self.assertEqual(
poll.await_args_list,
[mock.call(), mock.call(51)],
)
self.assertIn("compatible=no, enabled=yes, active=yes", report)
self.assertIn("compatibility flag: informational", report)
self.assertIn("clips matching expected schema: 1/1", report)
self.assertIn("clips mapped to known cameras: 1/1", report)
self.assertIn("PASS_WITH_CLIPS", report)
for private in ("CameraAnonyme", "Nom privé", "900", "51", "61", "71"):
self.assertNotIn(private, report)
async def test_manifeste_accessible_mais_camera_inconnue_est_partiel(self):
sync = SimpleNamespace(
network_id=7,
sync_id=900,
_local_storage={"compatible": False, "enabled": True, "status": True},
local_storage=True,
)
probe = mock.AsyncMock(return_value={
"clips": 2,
"schema_clips": 2,
"mapped_clips": 0,
"cameras": 1,
})
code, lines = await diagnostic_xr.inspect_blink(
SimpleNamespace(homescreen={}, sync={}),
select_modules=lambda _blink, _name: [("secret", sync)],
manifest_probe=probe,
)
self.assertEqual(code, 1)
self.assertIn("Overall result: PARTIAL", "\n".join(lines))
async def test_reponse_manifeste_invalide_reste_anonyme(self):
sync = SimpleNamespace(
poll_local_storage_manifest=mock.AsyncMock(
side_effect=[{"id": 51}, {"manifest_id": 61, "private": "SECRET"}],
),
_names_table={},
)
with self.assertRaisesRegex(RuntimeError, "manifest response"):
await diagnostic_xr.probe_local_manifest(sync)
async def test_ecriture_atomique_remplace_le_rapport(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / diagnostic_xr.REPORT_NAME
path.write_text("old", encoding="utf-8")
diagnostic_xr.write_report(path, "new report\n")
self.assertEqual(path.read_text(encoding="utf-8"), "new report\n")
self.assertEqual(list(path.parent.glob(".*.tmp")), [])
async def test_dossier_installation_non_inscriptible_se_replie_dans_temp(self):
with tempfile.TemporaryDirectory() as directory:
fallback = Path(directory) / diagnostic_xr.REPORT_NAME
with mock.patch.object(
diagnostic_xr, "write_report", side_effect=[PermissionError(), None],
) as write, mock.patch.object(
diagnostic_xr.tempfile, "gettempdir", return_value=directory,
):
selected = diagnostic_xr.write_report_with_fallback(
Path("C:/Program Files/blink") / diagnostic_xr.REPORT_NAME,
"report",
)
self.assertEqual(selected, fallback)
self.assertEqual(write.call_count, 2)
def test_main_fonctionne_sans_stdout_dans_le_bundle(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / diagnostic_xr.REPORT_NAME
with mock.patch.object(diagnostic_xr.runtime, "frozen", return_value=True), \
mock.patch.object(
diagnostic_xr, "collect_report",
new=mock.AsyncMock(return_value=(0, "safe report\n")),
), mock.patch.object(
diagnostic_xr, "report_path", return_value=path,
), mock.patch.object(
diagnostic_xr, "_show_message",
), mock.patch.object(
diagnostic_xr, "_open_report",
) as open_report, mock.patch.object(
diagnostic_xr.os, "environ", {},
), mock.patch.object(
diagnostic_xr.sys, "stdout", None,
), mock.patch.object(diagnostic_xr.sys, "stderr", None):
code = diagnostic_xr.main()
self.assertEqual(code, 0)
self.assertEqual(path.read_text(encoding="utf-8"), "safe report\n")
open_report.assert_called_once_with(path)
if __name__ == "__main__":
unittest.main()