diff --git a/blinkpy/api.py b/blinkpy/api.py index 68063f92..ef03cea6 100644 --- a/blinkpy/api.py +++ b/blinkpy/api.py @@ -1063,3 +1063,21 @@ async def oauth_refresh_token(auth, refresh_token, hardware_id): _LOGGER.error("OAuth token refresh failed with status %s", response.status) return None + + +async def request_sync_snooze(blink, network, data=None): + """ + Get or update sync snooze configuration. + + :param blink: Blink instance. + :param network: Sync module network id. + :param data: string w/JSON dict of parameters/values to update. + If None, performs a GET to read current snooze state. + """ + url = ( + f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}" + f"/networks/{network}/snooze" + ) + if data is None: + return await http_get(blink, url) + return await http_post(blink, url, json=True, data=data) diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 63b6aef6..333e90d7 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -127,6 +127,34 @@ async def async_arm(self, value): return await api.request_system_arm(self.blink, self.network_id) return await api.request_system_disarm(self.blink, self.network_id) + @property + async def snoozed(self): + """Return snooze status as boolean.""" + res = None + try: + res = await api.request_sync_snooze(self.blink, self.network_id) + if not isinstance(res, dict): + return False + snooze_till = res.get("snooze_till") + if not snooze_till: + return False + expiry = datetime.datetime.fromisoformat(snooze_till.replace("Z", "+00:00")) + if expiry.tzinfo is None: + expiry = expiry.replace(tzinfo=datetime.timezone.utc) + return expiry > datetime.datetime.now(datetime.timezone.utc) + except (TypeError, ValueError): + return False + + async def async_snooze(self, snooze_time=240): + """Set sync snooze status.""" + data = json_dumps({"snooze_time": snooze_time}, indent=None) + res = await api.request_sync_snooze( + self.blink, + self.network_id, + data=data, + ) + return res + async def start(self): """Initialize the system.""" _LOGGER.debug("Initializing the sync module") diff --git a/tests/test_api.py b/tests/test_api.py index f8f77b89..dba6959d 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -189,6 +189,14 @@ async def test_request_update_config(self, mock_resp): ) ) + async def test_request_sync_snooze(self, mock_resp): + """Test sync snooze request.""" + mock_resp.return_value = {"message": "Sync snoozed"} + response = await api.request_sync_snooze( + self.blink, "network", '{"snooze_time": 300}' + ) + self.assertEqual(response, {"message": "Sync snoozed"}) + async def test_wait_for_command(self, mock_resp): """Test Motion detect enable.""" mock_resp.side_effect = (COMMAND_NOT_COMPLETE, COMMAND_COMPLETE) diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py index b35f56fd..61b90e49 100644 --- a/tests/test_sync_module.py +++ b/tests/test_sync_module.py @@ -100,6 +100,87 @@ def test_get_unique_info_invalid_device(self, mock_resp) -> None: self.blink.homescreen = {"doorbells": [device], "owls": []} self.assertEqual(self.blink.sync["test"].get_unique_info("doorbell2"), None) + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"snooze_till": "2099-01-01T12:00:00+00:00"}), + ) + async def test_snoozed(self, mock_resp) -> None: + """Check that we get snoozed status.""" + result = await self.blink.sync["test"].snoozed + self.assertTrue(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value=None), + ) + async def test_snoozed_none(self, mock_resp) -> None: + """Check that we handle None response.""" + result = await self.blink.sync["test"].snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={}), + ) + async def test_snoozed_malformed(self, mock_resp) -> None: + """Check that we handle malformed response.""" + result = await self.blink.sync["test"].snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"snooze_till": ""}), + ) + async def test_snoozed_empty_string(self, mock_resp) -> None: + """Check that we handle empty string response.""" + result = await self.blink.sync["test"].snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"snooze_till": "2000-01-01T00:00:00+00:00"}), + ) + async def test_snoozed_expired(self, mock_resp) -> None: + """Check that expired snooze_till returns False.""" + result = await self.blink.sync["test"].snoozed + self.assertFalse(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"snooze_till": "2099-01-01T00:00:00Z"}), + ) + async def test_snoozed_z_suffix(self, mock_resp) -> None: + """Check that Z-suffix timestamps are parsed correctly.""" + result = await self.blink.sync["test"].snoozed + self.assertTrue(result) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"status": 200}), + ) + async def test_async_snooze(self, mock_resp) -> None: + """Check that we can set snooze.""" + result = await self.blink.sync["test"].async_snooze(300) + self.assertEqual(result, {"status": 200}) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value={"status": 400}), + ) + async def test_async_snooze_failure(self, mock_resp) -> None: + """Check that we handle snooze failure.""" + result = await self.blink.sync["test"].async_snooze(300) + self.assertEqual(result, {"status": 400}) + + @mock.patch( + "blinkpy.api.request_sync_snooze", + mock.AsyncMock(return_value=None), + ) + async def test_async_snooze_none_response(self, mock_resp) -> None: + """Check that we handle None response when setting snooze.""" + result = await self.blink.sync["test"].async_snooze(300) + self.assertIsNone(result) + async def test_get_events(self, mock_resp) -> None: """Test get events function.""" mock_resp.return_value = {"event": True}