Sync Module Snooze

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)

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")

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)

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}