Merge pull request #949 from fronzbot/add-notification-config

Added ability to get and set notification flags

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.

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

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 = (

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