From e56b6751f8482653dcd8b1032cb588200e469639 Mon Sep 17 00:00:00 2001
From: Kevin Fronczak <kfronczak@gmail.com>
Date: Wed, 19 Jun 2024 15:55:58 -0400
Subject: [PATCH] Added ability to get and set notification flags

---
 blinkpy/api.py                | 32 ++++++++++++++++++++++++++++++++
 blinkpy/blinkpy.py            | 15 +++++++++++++++
 tests/test_api.py             | 17 +++++++++++++++++
 tests/test_blink_functions.py | 18 ++++++++++++++++++
 4 files changed, 82 insertions(+)

diff --git a/blinkpy/api.py b/blinkpy/api.py
index c4121248..95dc280b 100644
--- a/blinkpy/api.py
+++ b/blinkpy/api.py
@@ -166,6 +166,38 @@ async def request_system_disarm(blink, network, **kwargs):
     return response
 
 
+async def request_notification_flags(blink, **kwargs):
+    """
+    Get system notification flags.
+
+    :param blink: Blink instance.
+    """
+    url = (
+        f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}"
+        "/notifications/configuration"
+    )
+    response = await http_get(blink, url)
+    await wait_for_command(blink, response)
+    return response
+
+
+async def request_set_notification_flag(blink, data_dict):
+    """
+    Set a system notification flag.
+
+    :param blink: Blink instance.
+    :param data_dict: Dictionary of notifications to set.
+    """
+    url = (
+        f"{blink.urls.base_url}/api/v1/accounts/{blink.account_id}"
+        "/notifications/configuration"
+    )
+    data = dumps({"notifications": data_dict})
+    response = await http_post(blink, url, data=data, json=False)
+    await wait_for_command(blink, response)
+    return response
+
+
 async def request_command_status(blink, network, command_id):
     """
     Request command status.
diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py
index dc031ac5..80ea937b 100644
--- a/blinkpy/blinkpy.py
+++ b/blinkpy/blinkpy.py
@@ -320,6 +320,21 @@ async def save(self, file_name):
         """Save login data to file."""
         await util.json_save(self.auth.login_attributes, file_name)
 
+    async def get_status(self):
+        """Get the blink system notification status."""
+        response = await api.request_notification_flags(self)
+        return response.get("notifications", response)
+
+    async def set_status(self, data_dict={}):
+        """
+        Set the blink system notification status.
+
+        :param data_dict: Dictionary of notification keys to modify.
+                          Example: {'low_battery': False, 'motion': False}
+        """
+        response = await api.request_set_notification_flag(self, data_dict)
+        return response
+
     async def download_videos(
         self, path, since=None, camera="all", stop=10, delay=1, debug=False
     ):
diff --git a/tests/test_api.py b/tests/test_api.py
index 2b3a850d..89581dbc 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -106,6 +106,23 @@ async def test_request_camera_usage(self, mock_resp):
             await api.request_camera_usage(self.blink), {"cameras": "1111"}
         )
 
+    async def test_request_notification_flags(self, mock_resp):
+        """Test notification flag request."""
+        mock_resp.return_value = {"notifications": {"some_key": False}}
+        self.assertEqual(
+            await api.request_notification_flags(self.blink),
+            {"notifications": {"some_key": False}},
+        )
+
+    async def test_request_set_notification_flag(self, mock_resp):
+        """Test set of notifiaction flags."""
+        mock_resp.side_effect = (
+            mresp.MockResponse(COMMAND_RESPONSE, 200),
+            COMMAND_COMPLETE,
+        )
+        response = await api.request_set_notification_flag(self.blink, {})
+        self.assertEqual(response.status, 200)
+
     async def test_request_motion_detection_enable(self, mock_resp):
         """Test  Motion detect enable."""
         mock_resp.side_effect = (
diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py
index eedebfd3..35477a71 100644
--- a/tests/test_blink_functions.py
+++ b/tests/test_blink_functions.py
@@ -278,3 +278,21 @@ async def test_refresh(self, mock_req, mock_update):
         self.blink.cameras = {"bar": MockCamera(self.blink.sync)}
         self.blink.sync["foo"].cameras = self.blink.cameras
         self.assertTrue(await self.blink.refresh())
+
+    @mock.patch("blinkpy.blinkpy.api.request_notification_flags")
+    async def test_get_status(self, mock_req):
+        """Test get of notification flags."""
+        mock_req.return_value = {"notifications": {"foo": True}}
+        self.assertDictEqual(await self.blink.get_status(), {"foo": True})
+
+    @mock.patch("blinkpy.blinkpy.api.request_notification_flags")
+    async def test_get_status_malformed(self, mock_req):
+        """Test get of notification flags with malformed response."""
+        mock_req.return_value = {"nobueno": {"foo": False}}
+        self.assertDictEqual(await self.blink.get_status(), {"nobueno": {"foo": False}})
+
+    @mock.patch("blinkpy.blinkpy.api.request_set_notification_flag")
+    async def test_set_status(self, mock_req):
+        """Test set of notification flags."""
+        mock_req.return_value = True
+        self.assertTrue(await self.blink.set_status())