diff --git a/blink_credentials.json b/blink_credentials.json new file mode 100644 index 00000000..985610ee --- /dev/null +++ b/blink_credentials.json @@ -0,0 +1,17 @@ +{ + "username": "nirgal@live.com", + "password": "yHuJ91)3", + "uid": "BlinkCamera_0286240e-db42-cd59-b7a5-2f64fc13bde8", + "device_id": "Blinkpy", + "token": "eyJhbGciOiJSUzI1NiIsImprdSI6Ii9vYXV0aC9pbnRlcm5hbC9qd2tzIiwia2lkIjoiNjQ1NWZmNDEiLCJ0eXAiOiJKV1QifQ.eyJhcHBfaWQiOiJpb3MiLCJjaWQiOiJpb3MiLCJleHAiOjE3NzQ2Mzk2MzgsImhhcmR3YXJlX2lkIjoiNkVBQzU1OEYtMEY4MS00NDM0LTgyMTYtRTU2MzM3RjU3NzBCIiwiaWF0IjoxNzc0NjI1MjM4LCJpc3MiOiJCbGlua09hdXRoU2VydmljZS1wcm9kOnVzLWVhc3QtMTo1YmM2OTg4ZCIsIm9pYXQiOjE3NzQ2MjUyMzgsInJuZCI6Ikl0VV9OdG1kWGIiLCJzY29wZXMiOlsiY2xpZW50Il0sInNlc3Npb25faWQiOiJibGluay1zZXNzaW9uLTA1MmQ0MDA5LTc2MjgtNDVlZC1hOGJkLWMzMmRiYzY3NWI4NiIsInVzZXJfaWQiOjEyNjk3NDExMH0.VOGUbc44jryHbFC_Trmgq46d0qo0PBBevsoRcgFa8lTtwpgH3zbjQFZatN8Xbs5MvpKNTr-Z7ED2BBL2aj10C6rI61ioBdef0Jae2GNPxgAzMnJwASKAQB2H1ZLPid6TaTZzujSvYnigckMNekYjc23b_rweJ_rXZwHOrsLT7ws1_Rc1rjxY5k8N9irPJy-bWBh_S6mv50_bFMSSu1tLFxGvWItsp5sDni7Bz64RhEJFO4AXFzqYcZ4RDW3qEknt6Zeh0z0cxcB7ZDuWxuMw3WF5CBcsjbvPLRHhniYMRPImT2fEcZnpWZMKS7d9nuB1a6A-QcirJ91rOA5GNK6XmA", + "expires_in": 14400, + "expiration_date": 1774639638.2754378, + "refresh_token": "eyJhbGciOiJSUzI1NiIsImprdSI6Ii9vYXV0aC9pbnRlcm5hbC9qd2tzIiwia2lkIjoiNjQ1NWZmNDEiLCJ0eXAiOiJKV1QifQ.eyJpYXQiOjE3NzQ2MjUyMzgsImlzcyI6IkJsaW5rT2F1dGhTZXJ2aWNlLXByb2Q6dXMtZWFzdC0xOjViYzY5ODhkIiwib2lhdCI6MTc3NDYyNTIzOCwicmVmcmVzaF9jaWQiOiJpb3MiLCJyZWZyZXNoX3Njb3BlcyI6WyJjbGllbnQiXSwicmVmcmVzaF91c2VyX2lkIjoxMjY5NzQxMTAsInJuZCI6IlJOZDA4dDRJazYiLCJzZXNzaW9uX2lkIjoiYmxpbmstc2Vzc2lvbi0wNTJkNDAwOS03NjI4LTQ1ZWQtYThiZC1jMzJkYmM2NzViODYiLCJ0eXBlIjoicmVmcmVzaC10b2tlbiJ9.kSRBILMQHxrZhWhbUm78xJDo011plQhotvOiyxoqVgbXZ0cueVvYw8hnaMfdocCpiPUrcibtiuBQYOP9zgzO7e-n98zMaCF_t1l-mVPl5s2j4qcWALoZFCs2fL_Szx-UuYjUxLm6iZrRX7CfDFz2BVaHwcYHgWaNR4V6IBXf_UeaSYrcXWjfdFrx24RnrWFpZQX9j6z2U64T22Z5Tc1ePrChnG7G_4-ELBZZuiw-FNNyZGr_oVsaxabmCI5_zBWZESEiuWBInKPGoBLcF9sENVJd-Bl6hS9eiaGDNCn4vdbN3hpZqlr4_2vCRY6oxJL5cj92jOOyoUk3m1eH_vqc7g", + "host": "prde.immedia-semi.com", + "region_id": "prde", + "client_id": "2340966", + "account_id": 230552, + "user_id": "230551", + "hardware_id": "6EAC558F-0F81-4434-8216-E56337F5770B", + "2fa_code": "190161" +} \ No newline at end of file diff --git a/blink_list_devices_json.py b/blink_list_devices_json.py new file mode 100644 index 00000000..4de708f0 --- /dev/null +++ b/blink_list_devices_json.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +import asyncio +import builtins +import json +from pathlib import Path + +from aiohttp import ClientSession +from blinkpy.auth import Auth +from blinkpy.auth import BlinkTwoFARequiredError +from blinkpy.blinkpy import Blink + + +CRED_FILE = Path("blink_credentials.json") + + +def load_credentials(path: Path) -> dict: + if not path.exists(): + return {} + raw = path.read_text(encoding="utf-8").strip() + if not raw: + return {} + try: + data = json.loads(raw) + if not isinstance(data, dict): + return {} + return data + except json.JSONDecodeError: + return {} + + +def save_credentials(path: Path, data: dict) -> None: + path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + +def prompt_if_missing(creds: dict, key: str, prompt_text: str) -> str: + value = (creds.get(key) or "").strip() + if not value: + value = input(prompt_text).strip() + creds[key] = value + return value + + +async def close_session_if_needed(session_obj) -> None: + if session_obj is None: + return + if getattr(session_obj, "closed", False): + return + close_fn = getattr(session_obj, "close", None) + if close_fn is None: + return + result = close_fn() + if asyncio.iscoroutine(result): + await result + + +async def prompt_2fa_with_optional_stored_code(blink: Blink, stored_code: str) -> str: + code = (stored_code or "").strip() + if not code: + code = input("Enter Blink 2FA code: ").strip() + + # blinkpy 0.25.x exposes prompt_2fa(), but not send_auth_key(). + # Feed the stored code into the prompt automatically. + original_input = builtins.input + try: + builtins.input = lambda _prompt="": code + await blink.prompt_2fa() + return code + finally: + builtins.input = original_input + + +async def main(): + creds = load_credentials(CRED_FILE) + + username = prompt_if_missing(creds, "username", "Blink username/email: ") + password = prompt_if_missing(creds, "password", "Blink password: ") + creds.setdefault("2fa_code", "") + + session = ClientSession() + blink = None + try: + blink = Blink(session=session) + auth_payload = dict(creds) + auth_payload["username"] = username + auth_payload["password"] = password + auth_payload.pop("2fa_code", None) + auth = Auth(auth_payload, no_prompt=True) + blink.auth = auth + + try: + await blink.start() + except BlinkTwoFARequiredError: + try: + used_code = await prompt_2fa_with_optional_stored_code( + blink, creds.get("2fa_code", "") + ) + if used_code: + creds["2fa_code"] = used_code + except Exception: + code = input("Stored 2FA code failed. Enter new 2FA code: ").strip() + creds["2fa_code"] = code + await prompt_2fa_with_optional_stored_code(blink, code) + + # Persist Blink auth cache/tokens so future runs avoid repeated 2FA. + await blink.save(str(CRED_FILE)) + saved = load_credentials(CRED_FILE) + saved["username"] = username + saved["password"] = password + saved["2fa_code"] = creds.get("2fa_code", "") + save_credentials(CRED_FILE, saved) + + print("\n=== Cameras (including doorbells) ===") + if blink.cameras: + for name, camera in blink.cameras.items(): + cam_type = getattr(camera, "camera_type", "unknown") + print(f"- {name} [{cam_type}]") + else: + print("No cameras found.") + + print("\n=== Doorbells ===") + doorbells = getattr(blink, "doorbells", None) + if doorbells: + for name, bell in doorbells.items(): + bell_type = getattr(bell, "camera_type", "doorbell") + print(f"- {name} [{bell_type}]") + else: + derived = [ + (name, cam) + for name, cam in blink.cameras.items() + if "doorbell" in str(getattr(cam, "camera_type", "")).lower() + ] + if derived: + for name, bell in derived: + print(f"- {name} [{getattr(bell, 'camera_type', 'doorbell')}]") + else: + print("No doorbells found.") + finally: + # Some blinkpy flows can leave extra aiohttp sessions open. + # Close all known session handles explicitly. + seen = set() + session_candidates = [ + getattr(blink, "session", None) if blink else None, + getattr(getattr(blink, "auth", None), "session", None) if blink else None, + session, + ] + for item in session_candidates: + if item is None or id(item) in seen: + continue + seen.add(id(item)) + await close_session_if_needed(item) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/blinkpy-changes.patch b/blinkpy-changes.patch new file mode 100644 index 00000000..15a75369 --- /dev/null +++ b/blinkpy-changes.patch @@ -0,0 +1,98 @@ +diff --git a/blinkpy/blinkpy.py b/share/blinkpy-upstream-0.25.5/blinkpy/blinkpy.py +index c970420..375f9a6 100644 +--- a/blinkpy/blinkpy.py ++++ b/share/blinkpy-upstream-0.25.5/blinkpy/blinkpy.py +@@ -75,6 +75,51 @@ class Blink: + self.homescreen = {} + self.no_owls = no_owls + ++ def _iter_device_dicts(self, payload): ++ """Yield device-like dictionaries from potentially nested payloads.""" ++ if isinstance(payload, list): ++ for item in payload: ++ yield from self._iter_device_dicts(item) ++ return ++ if isinstance(payload, dict): ++ # Treat this as a device entry when it has the minimum fields. ++ if "name" in payload and "network_id" in payload: ++ yield payload ++ for value in payload.values(): ++ yield from self._iter_device_dicts(value) ++ ++ def get_homescreen_devices(self, kind): ++ """Return homescreen devices for a given kind.""" ++ key_candidates = { ++ "mini": ["owls", "mini_cameras", "minis"], ++ "doorbell": ["doorbells", "lotus", "doorbell_cameras"], ++ } ++ devices = [] ++ seen = set() ++ for key in key_candidates.get(kind, []): ++ payload = self.homescreen.get(key) ++ for device in self._iter_device_dicts(payload): ++ signature = ( ++ str(device.get("id")), ++ str(device.get("network_id")), ++ str(device.get("name")), ++ ) ++ if signature in seen: ++ continue ++ seen.add(signature) ++ devices.append(device) ++ return devices ++ ++ def has_sync_module_for_network(self, network_id): ++ """Check whether homescreen reports a real sync module for network.""" ++ try: ++ for sync in self.homescreen.get("sync_modules", []): ++ if str(sync.get("network_id")) == str(network_id): ++ return True ++ except AttributeError: ++ return False ++ return False ++ + @property + def client_id(self): + """Return the client id.""" +@@ -220,10 +265,12 @@ class Blink: + network_list = [] + camera_list = [] + try: +- for owl in self.homescreen["owls"]: ++ for owl in self.get_homescreen_devices("mini"): + name = owl["name"] + network_id = str(owl["network_id"]) +- if network_id in self.network_ids: ++ if network_id in self.network_ids and self.has_sync_module_for_network( ++ network_id ++ ): + camera_list.append( + {network_id: {"name": name, "id": network_id, "type": "mini"}} + ) +@@ -244,10 +291,12 @@ class Blink: + network_list = [] + camera_list = [] + try: +- for lotus in self.homescreen["doorbells"]: ++ for lotus in self.get_homescreen_devices("doorbell"): + name = lotus["name"] + network_id = str(lotus["network_id"]) +- if network_id in self.network_ids: ++ if network_id in self.network_ids and self.has_sync_module_for_network( ++ network_id ++ ): + camera_list.append( + { + network_id: { +@@ -287,9 +336,11 @@ class Blink: + lotus_cameras = await self.setup_lotus() + for camera in mini_cameras: + for network, camera_info in camera.items(): ++ all_cameras.setdefault(network, []) + all_cameras[network].append(camera_info) + for camera in lotus_cameras: + for network, camera_info in camera.items(): ++ all_cameras.setdefault(network, []) + all_cameras[network].append(camera_info) + return all_cameras + except (KeyError, TypeError) as ex: diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index c970420c..375f9a63 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -75,6 +75,51 @@ def __init__( self.homescreen = {} self.no_owls = no_owls + def _iter_device_dicts(self, payload): + """Yield device-like dictionaries from potentially nested payloads.""" + if isinstance(payload, list): + for item in payload: + yield from self._iter_device_dicts(item) + return + if isinstance(payload, dict): + # Treat this as a device entry when it has the minimum fields. + if "name" in payload and "network_id" in payload: + yield payload + for value in payload.values(): + yield from self._iter_device_dicts(value) + + def get_homescreen_devices(self, kind): + """Return homescreen devices for a given kind.""" + key_candidates = { + "mini": ["owls", "mini_cameras", "minis"], + "doorbell": ["doorbells", "lotus", "doorbell_cameras"], + } + devices = [] + seen = set() + for key in key_candidates.get(kind, []): + payload = self.homescreen.get(key) + for device in self._iter_device_dicts(payload): + signature = ( + str(device.get("id")), + str(device.get("network_id")), + str(device.get("name")), + ) + if signature in seen: + continue + seen.add(signature) + devices.append(device) + return devices + + def has_sync_module_for_network(self, network_id): + """Check whether homescreen reports a real sync module for network.""" + try: + for sync in self.homescreen.get("sync_modules", []): + if str(sync.get("network_id")) == str(network_id): + return True + except AttributeError: + return False + return False + @property def client_id(self): """Return the client id.""" @@ -220,10 +265,12 @@ async def setup_owls(self): network_list = [] camera_list = [] try: - for owl in self.homescreen["owls"]: + for owl in self.get_homescreen_devices("mini"): name = owl["name"] network_id = str(owl["network_id"]) - if network_id in self.network_ids: + if network_id in self.network_ids and self.has_sync_module_for_network( + network_id + ): camera_list.append( {network_id: {"name": name, "id": network_id, "type": "mini"}} ) @@ -244,10 +291,12 @@ async def setup_lotus(self): network_list = [] camera_list = [] try: - for lotus in self.homescreen["doorbells"]: + for lotus in self.get_homescreen_devices("doorbell"): name = lotus["name"] network_id = str(lotus["network_id"]) - if network_id in self.network_ids: + if network_id in self.network_ids and self.has_sync_module_for_network( + network_id + ): camera_list.append( { network_id: { @@ -287,9 +336,11 @@ async def setup_camera_list(self): lotus_cameras = await self.setup_lotus() for camera in mini_cameras: for network, camera_info in camera.items(): + all_cameras.setdefault(network, []) all_cameras[network].append(camera_info) for camera in lotus_cameras: for network, camera_info in camera.items(): + all_cameras.setdefault(network, []) all_cameras[network].append(camera_info) return all_cameras except (KeyError, TypeError) as ex: diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 63b6aef6..103d88d1 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -225,8 +225,8 @@ async def update_cameras(self, camera_type=BlinkCamera): def get_unique_info(self, name): """Extract unique information for Minis and Doorbells.""" try: - for type_key in self.type_key_map.values(): - for device in self.blink.homescreen[type_key]: + for type_name in self.type_key_map: + for device in self.blink.get_homescreen_devices(type_name): _LOGGER.debug("checking device %s", device) if device["name"] == name: _LOGGER.debug("Found unique_info %s", device) @@ -551,7 +551,7 @@ async def update_cameras(self, camera_type=BlinkCameraMini): async def get_camera_info(self, camera_id, **kwargs): """Retrieve camera information.""" try: - for owl in self.blink.homescreen["owls"]: + for owl in self.blink.get_homescreen_devices("mini"): if owl["name"] == self.name: self.status = owl["enabled"] return owl @@ -614,7 +614,7 @@ async def update_cameras(self, camera_type=BlinkDoorbell): async def get_camera_info(self, camera_id, **kwargs): """Retrieve camera information.""" try: - for doorbell in self.blink.homescreen["doorbells"]: + for doorbell in self.blink.get_homescreen_devices("doorbell"): if doorbell["name"] == self.name: self.status = doorbell["enabled"] return doorbell diff --git a/sync-module-changes.patch b/sync-module-changes.patch new file mode 100644 index 00000000..834f64fe --- /dev/null +++ b/sync-module-changes.patch @@ -0,0 +1,33 @@ +diff --git a/blinkpy/sync_module.py b/share/blinkpy-upstream-0.25.5/blinkpy/sync_module.py +index 63b6aef..103d88d 100644 +--- a/blinkpy/sync_module.py ++++ b/share/blinkpy-upstream-0.25.5/blinkpy/sync_module.py +@@ -225,8 +225,8 @@ class BlinkSyncModule: + def get_unique_info(self, name): + """Extract unique information for Minis and Doorbells.""" + try: +- for type_key in self.type_key_map.values(): +- for device in self.blink.homescreen[type_key]: ++ for type_name in self.type_key_map: ++ for device in self.blink.get_homescreen_devices(type_name): + _LOGGER.debug("checking device %s", device) + if device["name"] == name: + _LOGGER.debug("Found unique_info %s", device) +@@ -551,7 +551,7 @@ class BlinkOwl(BlinkSyncModule): + async def get_camera_info(self, camera_id, **kwargs): + """Retrieve camera information.""" + try: +- for owl in self.blink.homescreen["owls"]: ++ for owl in self.blink.get_homescreen_devices("mini"): + if owl["name"] == self.name: + self.status = owl["enabled"] + return owl +@@ -614,7 +614,7 @@ class BlinkLotus(BlinkSyncModule): + async def get_camera_info(self, camera_id, **kwargs): + """Retrieve camera information.""" + try: +- for doorbell in self.blink.homescreen["doorbells"]: ++ for doorbell in self.blink.get_homescreen_devices("doorbell"): + if doorbell["name"] == self.name: + self.status = doorbell["enabled"] + return doorbell diff --git a/test-changes.patch b/test-changes.patch new file mode 100644 index 00000000..f3cf8f23 --- /dev/null +++ b/test-changes.patch @@ -0,0 +1,111 @@ +diff --git a/tests/test_blinkpy.py b/share/blinkpy-upstream-0.25.5/tests/test_blinkpy.py +index 771189f..a71082e 100644 +--- a/tests/test_blinkpy.py ++++ b/share/blinkpy-upstream-0.25.5/tests/test_blinkpy.py +@@ -230,6 +230,7 @@ class TestBlinkSetup(IsolatedAsyncioTestCase): + """Test that blink mini cameras are found if attached to sync module.""" + self.blink.network_ids = ["1234"] + self.blink.homescreen = { ++ "sync_modules": [{"network_id": 1234}], + "owls": [ + { + "id": 1, +@@ -261,6 +262,7 @@ class TestBlinkSetup(IsolatedAsyncioTestCase): + """Test that blink mini cameras are properly attached to sync module.""" + self.blink.network_ids = ["1234"] + self.blink.homescreen = { ++ "sync_modules": [{"network_id": 1234}], + "owls": [ + { + "id": 1, +@@ -322,6 +324,7 @@ class TestBlinkSetup(IsolatedAsyncioTestCase): + """Test that blink doorbell cameras are properly attached to sync module.""" + self.blink.network_ids = ["1234"] + self.blink.homescreen = { ++ "sync_modules": [{"network_id": 1234}], + "doorbells": [ + { + "id": 1, +@@ -341,11 +344,66 @@ class TestBlinkSetup(IsolatedAsyncioTestCase): + result, {"1234": [{"name": "foo", "id": "1234", "type": "doorbell"}]} + ) + ++ @mock.patch("blinkpy.api.request_camera_usage") ++ async def test_blink_doorbell_with_alt_homescreen_key(self, mock_usage): ++ """Test that doorbells are discovered from alternate homescreen keys.""" ++ self.blink.network_ids = ["1234"] ++ self.blink.homescreen = { ++ "sync_modules": [{"network_id": 1234}], ++ "lotus": { ++ "devices": [ ++ { ++ "id": 1, ++ "name": "foo", ++ "network_id": 1234, ++ "onboarded": True, ++ "enabled": True, ++ "status": "online", ++ "thumbnail": "/foo/bar", ++ "serial": "abc123", ++ } ++ ] ++ } ++ } ++ mock_usage.return_value = {"networks": [{"cameras": [], "network_id": 1234}]} ++ result = await self.blink.setup_camera_list() ++ self.assertEqual( ++ result, {"1234": [{"name": "foo", "id": "1234", "type": "doorbell"}]} ++ ) ++ ++ @mock.patch("blinkpy.blinkpy.BlinkLotus.start") ++ @mock.patch("blinkpy.api.request_camera_usage") ++ async def test_blink_syncless_doorbell_not_in_camera_usage( ++ self, mock_usage, mock_lotus_start ++ ): ++ """Test that sync-less doorbells initialize even without camera_usage network.""" ++ mock_lotus_start.return_value = True ++ self.blink.network_ids = [] ++ self.blink.homescreen = { ++ "doorbells": [ ++ { ++ "id": 1, ++ "name": "foo", ++ "network_id": 1234, ++ "onboarded": True, ++ "enabled": True, ++ "status": "online", ++ "thumbnail": "/foo/bar", ++ "serial": "abc123", ++ } ++ ] ++ } ++ mock_usage.return_value = {"networks": []} ++ result = await self.blink.setup_camera_list() ++ self.assertEqual(result, {}) ++ self.assertEqual(self.blink.sync["foo"].__class__, BlinkLotus) ++ + @mock.patch("blinkpy.api.request_camera_usage") + async def test_blink_multi_doorbell(self, mock_usage): + """Test that multiple doorbells are properly attached to sync module.""" + self.blink.network_ids = ["1234"] + self.blink.homescreen = { ++ "sync_modules": [{"network_id": 1234}], + "doorbells": [ + { + "id": 1, +@@ -384,6 +442,7 @@ class TestBlinkSetup(IsolatedAsyncioTestCase): + """Test that multiple minis are properly attached to sync module.""" + self.blink.network_ids = ["1234"] + self.blink.homescreen = { ++ "sync_modules": [{"network_id": 1234}], + "owls": [ + { + "id": 1, +@@ -422,6 +481,7 @@ class TestBlinkSetup(IsolatedAsyncioTestCase): + """Test that a mix of cameras are properly attached to sync module.""" + self.blink.network_ids = ["1234"] + self.blink.homescreen = { ++ "sync_modules": [{"network_id": 1234}], + "doorbells": [ + { + "id": 1, diff --git a/tests/test_blinkpy.py b/tests/test_blinkpy.py index 771189f7..b9953e0b 100644 --- a/tests/test_blinkpy.py +++ b/tests/test_blinkpy.py @@ -230,6 +230,7 @@ async def test_blink_mini_cameras_returned(self): """Test that blink mini cameras are found if attached to sync module.""" self.blink.network_ids = ["1234"] self.blink.homescreen = { + "sync_modules": [{"network_id": 1234}], "owls": [ { "id": 1, @@ -241,7 +242,7 @@ async def test_blink_mini_cameras_returned(self): "thumbnail": "/foo/bar", "serial": "abc123", } - ] + ], } result = await self.blink.setup_owls() self.assertEqual(self.blink.network_ids, ["1234"]) @@ -261,6 +262,7 @@ async def test_blink_mini_attached_to_sync(self, mock_usage): """Test that blink mini cameras are properly attached to sync module.""" self.blink.network_ids = ["1234"] self.blink.homescreen = { + "sync_modules": [{"network_id": 1234}], "owls": [ { "id": 1, @@ -272,7 +274,7 @@ async def test_blink_mini_attached_to_sync(self, mock_usage): "thumbnail": "/foo/bar", "serial": "abc123", } - ] + ], } mock_usage.return_value = {"networks": [{"cameras": [], "network_id": 1234}]} result = await self.blink.setup_camera_list() @@ -322,6 +324,7 @@ async def test_blink_doorbell_attached_to_sync(self, mock_usage): """Test that blink doorbell cameras are properly attached to sync module.""" self.blink.network_ids = ["1234"] self.blink.homescreen = { + "sync_modules": [{"network_id": 1234}], "doorbells": [ { "id": 1, @@ -333,7 +336,34 @@ async def test_blink_doorbell_attached_to_sync(self, mock_usage): "thumbnail": "/foo/bar", "serial": "abc123", } - ] + ], + } + mock_usage.return_value = {"networks": [{"cameras": [], "network_id": 1234}]} + result = await self.blink.setup_camera_list() + self.assertEqual( + result, {"1234": [{"name": "foo", "id": "1234", "type": "doorbell"}]} + ) + + @mock.patch("blinkpy.api.request_camera_usage") + async def test_blink_doorbell_with_alt_homescreen_key(self, mock_usage): + """Test that doorbells are discovered from alternate homescreen keys.""" + self.blink.network_ids = ["1234"] + self.blink.homescreen = { + "sync_modules": [{"network_id": 1234}], + "lotus": { + "devices": [ + { + "id": 1, + "name": "foo", + "network_id": 1234, + "onboarded": True, + "enabled": True, + "status": "online", + "thumbnail": "/foo/bar", + "serial": "abc123", + } + ] + }, } mock_usage.return_value = {"networks": [{"cameras": [], "network_id": 1234}]} result = await self.blink.setup_camera_list() @@ -341,11 +371,42 @@ async def test_blink_doorbell_attached_to_sync(self, mock_usage): result, {"1234": [{"name": "foo", "id": "1234", "type": "doorbell"}]} ) + @mock.patch("blinkpy.blinkpy.BlinkLotus.start") + @mock.patch("blinkpy.api.request_camera_usage") + async def test_blink_syncless_doorbell_not_in_camera_usage( + self, mock_usage, mock_lotus_start + ): + """Test that sync-less doorbells initialization. + + Ensure they initialize even without camera_usage network.. + """ + mock_lotus_start.return_value = True + self.blink.network_ids = [] + self.blink.homescreen = { + "doorbells": [ + { + "id": 1, + "name": "foo", + "network_id": 1234, + "onboarded": True, + "enabled": True, + "status": "online", + "thumbnail": "/foo/bar", + "serial": "abc123", + } + ] + } + mock_usage.return_value = {"networks": []} + result = await self.blink.setup_camera_list() + self.assertEqual(result, {}) + self.assertEqual(self.blink.sync["foo"].__class__, BlinkLotus) + @mock.patch("blinkpy.api.request_camera_usage") async def test_blink_multi_doorbell(self, mock_usage): """Test that multiple doorbells are properly attached to sync module.""" self.blink.network_ids = ["1234"] self.blink.homescreen = { + "sync_modules": [{"network_id": 1234}], "doorbells": [ { "id": 1, @@ -367,7 +428,7 @@ async def test_blink_multi_doorbell(self, mock_usage): "thumbnail": "/bar/foo", "serial": "zxc456", }, - ] + ], } expected = { "1234": [ @@ -384,6 +445,7 @@ async def test_blink_multi_mini(self, mock_usage): """Test that multiple minis are properly attached to sync module.""" self.blink.network_ids = ["1234"] self.blink.homescreen = { + "sync_modules": [{"network_id": 1234}], "owls": [ { "id": 1, @@ -405,7 +467,7 @@ async def test_blink_multi_mini(self, mock_usage): "thumbnail": "/bar/foo", "serial": "zxc456", }, - ] + ], } expected = { "1234": [ @@ -422,6 +484,7 @@ async def test_blink_camera_mix(self, mock_usage): """Test that a mix of cameras are properly attached to sync module.""" self.blink.network_ids = ["1234"] self.blink.homescreen = { + "sync_modules": [{"network_id": 1234}], "doorbells": [ { "id": 1, diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py index b35f56fd..cc503379 100644 --- a/tests/test_sync_module.py +++ b/tests/test_sync_module.py @@ -84,10 +84,7 @@ def test_bad_arm(self, mock_resp) -> None: def test_get_unique_info_valid_device(self, mock_resp) -> None: """Check that we get the correct info.""" - device = { - "enabled": True, - "name": "doorbell1", - } + device = {"enabled": True, "name": "doorbell1", "id": 1234, "network_id": 5678} self.blink.homescreen = {"doorbells": [device], "owls": []} self.assertEqual(self.blink.sync["test"].get_unique_info("doorbell1"), device)