Activities retrieval and testing.

skybellpy/device.py

@@ -86,6 +86,26 @@ def update(self, device_json=None, device_info_json=None,
                 {k: device_settings_json[k] for k in device_settings_json
                  if k in self._device_settings_json})
 
+    def activities(self, limit=1, event=None):
+        """Return device activity information."""
+        activities = self._device_activities
+
+        # Make sure we're working with an array
+        if not activities:
+            activities = []
+        elif not isinstance(activities, (list, tuple)):
+            activities = [activities]
+
+        # Filter our activity array
+        if event:
+            activities = list(
+                filter(
+                    lambda activity:
+                    activity[CONST.EVENT] == event, activities))
+
+        # Return the requested number
+        return activities[:limit]
+
     def _set_setting(self, settings):
         """Validate the settings and then send the PATCH request."""
         for key, value in settings.items():

skybellpy/helpers/constants.py

@@ -79,12 +79,15 @@
 CHECK_IN = 'checkedInAt'
 
 # DEVICE ACTIVITIES
+EVENT = 'event'
 EVENT_ON_DEMAND = 'application:on-demand'
 EVENT_BUTTON = 'device:sensor:button'
 EVENT_MOTION = 'device:sensor:motion'
 
+STATE = 'state'
 STATE_READY = 'ready'
 
+VIDEO_STATE = 'videoState'
 VIDEO_STATE_READY = 'download:ready'
 
 # DEVICE SETTINGS

tests/mock/device_activities.py

@@ -9,7 +9,7 @@
 def get_response_ok(dev_id=DEVICE.DEVID,
                     event=CONST.EVENT_BUTTON,
                     state=CONST.STATE_READY,
-                    video_state=CONST.VIDEO_STATE_READY,):
+                    video_state=CONST.VIDEO_STATE_READY):
     """Return the device activity response json."""
     return '''
     {

tests/test_device.py

@@ -392,3 +392,104 @@ def tests_settings_failed(self, m):
         # Test setting to false then validate still True
         device.do_not_disturb = False
         self.assertEqual(device.do_not_disturb, True)
+
+    @requests_mock.mock()
+    def tests_activities(self, m):
+        """Check that the Skybell device activities work."""
+        m.post(CONST.LOGIN_URL, text=LOGIN.post_response_ok())
+
+        # Set up device
+        device_text = '[' + DEVICE.get_response_ok() + ']'
+        info_text = DEVICE_INFO.get_response_ok()
+        info_url = str.replace(CONST.DEVICE_INFO_URL, '$DEVID$', DEVICE.DEVID)
+
+        settings_text = DEVICE_SETTINGS.get_response_ok()
+        settings_url = str.replace(CONST.DEVICE_SETTINGS_URL,
+                                   '$DEVID$', DEVICE.DEVID)
+
+        activities_text = '[' + \
+            DEVICE_ACTIVITIES.get_response_ok(
+                dev_id=DEVICE.DEVID,
+                event=CONST.EVENT_BUTTON) + ',' + \
+            DEVICE_ACTIVITIES.get_response_ok(
+                dev_id=DEVICE.DEVID,
+                event=CONST.EVENT_MOTION) + ',' + \
+            DEVICE_ACTIVITIES.get_response_ok(
+                dev_id=DEVICE.DEVID,
+                event=CONST.EVENT_ON_DEMAND) + ']'
+        activities_json = json.loads(activities_text)
+
+        activities_url = str.replace(CONST.DEVICE_ACTIVITIES_URL,
+                                     '$DEVID$', DEVICE.DEVID)
+
+        m.get(CONST.DEVICES_URL, text=device_text)
+        m.get(info_url, text=info_text)
+        m.get(settings_url, text=settings_text)
+        m.get(activities_url, text=activities_text)
+
+        # Logout to reset everything
+        self.skybell.logout()
+
+        # Get our specific device
+        device = self.skybell.get_device(DEVICE.DEVID)
+        self.assertIsNotNone(device)
+        # pylint: disable=W0212
+        self.assertEqual(device._device_activities, activities_json)
+
+        # Get all activities from device
+        activities = device.activities(limit=100)
+        self.assertIsNotNone(activities)
+        self.assertEqual(len(activities), 3)
+
+        # Get only button activities
+        activities = device.activities(event=CONST.EVENT_BUTTON)
+        self.assertIsNotNone(activities)
+        self.assertEqual(len(activities), 1)
+        self.assertEqual(activities[0][CONST.EVENT], CONST.EVENT_BUTTON)
+
+    @requests_mock.mock()
+    def tests_bad_activities(self, m):
+        """Check that device activities recovers from bad data."""
+        m.post(CONST.LOGIN_URL, text=LOGIN.post_response_ok())
+
+        # Set up device
+        device_text = '[' + DEVICE.get_response_ok() + ']'
+        info_text = DEVICE_INFO.get_response_ok()
+        info_url = str.replace(CONST.DEVICE_INFO_URL, '$DEVID$', DEVICE.DEVID)
+
+        settings_text = DEVICE_SETTINGS.get_response_ok()
+        settings_url = str.replace(CONST.DEVICE_SETTINGS_URL,
+                                   '$DEVID$', DEVICE.DEVID)
+
+        activities_text = DEVICE_ACTIVITIES.get_response_ok(
+            dev_id=DEVICE.DEVID,
+            event=CONST.EVENT_BUTTON)
+
+        activities_url = str.replace(CONST.DEVICE_ACTIVITIES_URL,
+                                     '$DEVID$', DEVICE.DEVID)
+
+        m.get(CONST.DEVICES_URL, text=device_text)
+        m.get(info_url, text=info_text)
+        m.get(settings_url, text=settings_text)
+        m.get(activities_url, text=activities_text)
+
+        # Logout to reset everything
+        self.skybell.logout()
+
+        # Get our specific device
+        device = self.skybell.get_device(DEVICE.DEVID)
+        self.assertIsNotNone(device)
+
+        # Get all activities from device
+        activities = device.activities(limit=100)
+        self.assertIsNotNone(activities)
+        self.assertEqual(len(activities), 1)
+
+        # Force our device variable empty
+        # pylint: disable=W0212
+        device._device_activities = None
+
+        # Get all activities from device
+        activities = device.activities(limit=100)
+        self.assertIsNotNone(activities)
+        self.assertEqual(len(activities), 0)