Finished events with tests.

skybellpy/__init__.py

@@ -78,13 +78,16 @@ def login(self, username=None, password=None):
         if self._password is None or not isinstance(self._password, str):
             raise SkybellAuthenticationException(ERROR.PASSWORD)
 
-        self._cache[CONST.ACCESS_TOKEN] = None
+        self.update_cache(
+            {
+                CONST.ACCESS_TOKEN: None
+            })
 
         login_data = {
             'username': self._username,
             'password': self._password,
-            'appId': self._cache[CONST.APP_ID],
-            CONST.TOKEN: self._cache[CONST.TOKEN]
+            'appId': self.cache(CONST.APP_ID),
+            CONST.TOKEN: self.cache(CONST.TOKEN)
         }
 
         try:
@@ -106,7 +109,7 @@ def login(self, username=None, password=None):
 
     def logout(self):
         """Explicit Skybell logout."""
-        if self._cache[CONST.ACCESS_TOKEN]:
+        if self.cache(CONST.ACCESS_TOKEN):
             # No explicit logout call as it doesn't seem to matter
             # if a logout happens without registering the app which
             # we aren't currently doing.
@@ -158,23 +161,23 @@ def get_device(self, device_id, refresh=False):
     def send_request(self, method, url, headers=None,
                      json_data=None, retry=True):
         """Send requests to Skybell."""
-        if not self._cache[CONST.ACCESS_TOKEN] and url != CONST.LOGIN_URL:
+        if not self.cache(CONST.ACCESS_TOKEN) and url != CONST.LOGIN_URL:
             self.login()
 
         if not headers:
             headers = {}
 
-        if self._cache[CONST.ACCESS_TOKEN]:
+        if self.cache(CONST.ACCESS_TOKEN):
             headers['Authorization'] = 'Bearer ' + \
-                self._cache[CONST.ACCESS_TOKEN]
+                self.cache(CONST.ACCESS_TOKEN)
 
         headers['user-agent'] = (
             'SkyBell/3.4.1 (iPhone9,2; iOS 11.0; loc=en_US; lang=en-US) '
             'com.skybell.doorbell/1')
         headers['content-type'] = 'application/json'
         headers['accepts'] = '*/*'
-        headers['x-skybell-app-id'] = self._cache[CONST.APP_ID]
-        headers['x-skybell-client-id'] = self._cache[CONST.CLIENT_ID]
+        headers['x-skybell-app-id'] = self.cache(CONST.APP_ID)
+        headers['x-skybell-client-id'] = self.cache(CONST.CLIENT_ID)
 
         try:
             response = getattr(self._session, method)(
@@ -214,7 +217,9 @@ def update_dev_cache(self, device, data):
         """Update cached values for a device."""
         self.update_cache(
             {
-                device.device_id: data
+                CONST.DEVICES: {
+                    device.device_id: data
+                }
             })
 
     def _load_cache(self):

skybellpy/device.py

@@ -127,6 +127,12 @@ def activities(self, limit=1, event=None):
         # Return the requested number
         return activities[:limit]
 
+    def latest(self, event):
+        """Return the latest event activity."""
+        events = self._skybell.dev_cache(self, CONST.EVENT) or {}
+        # self._skybell._cache)
+        return events.get(event)
+
     def _set_setting(self, settings):
         """Validate the settings and then send the PATCH request."""
         for key, value in settings.items():

skybellpy/utils.py

@@ -32,8 +32,6 @@ def gen_token():
 
 def update(dct, dct_merge):
     """Recursively merge dicts."""
-    if not isinstance(dct_merge, dict):
-        return dct_merge
     for key, value in dct_merge.items():
         if key in dct and isinstance(dct[key], dict):
             dct[key] = update(dct[key], value)

tests/mock/device_activities.py

@@ -1,5 +1,7 @@
 """Mock Skybell Device Activities Response."""
 
+import datetime
+
 import skybellpy.helpers.constants as CONST
 import tests.mock.device as DEVICE
 
@@ -9,18 +11,20 @@
 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,
+                    created_at=datetime.datetime.now()):
     """Return the device activity response json."""
+    str_created_at = created_at.strftime('%Y-%m-%dT%H:%M:%SZ')
     return '''
     {
         "_id": "activityId",
-        "updatedAt": "2017-09-28T21:50:41.740Z",
-        "createdAt": "2017-09-28T21:50:41.740Z",
+        "updatedAt": "''' + str_created_at + '''",
+        "createdAt": "''' + str_created_at + '''",
         "device": "''' + dev_id + '''",
-        "callId": "activityCallId",
+        "callId": "''' + str_created_at + '''",
         "event": "''' + event + '''",
         "state": "''' + state + '''",
-        "ttlStartDate": "2017-09-28T21:50:41.739Z",
+        "ttlStartDate": "''' + str_created_at + '''",
         "videoState": "''' + video_state + '''",
         "id": "activityId",
         "media": "http://www.image.com/image.jpg",

tests/test_device.py

@@ -3,6 +3,7 @@
 
 Tests the device initialization and attributes of the Skybell device class.
 """
+import datetime
 import json
 import unittest
 
@@ -493,3 +494,117 @@ def tests_bad_activities(self, m):
         activities = device.activities(limit=100)
         self.assertIsNotNone(activities)
         self.assertEqual(len(activities), 0)
+
+    @requests_mock.mock()
+    def tests_latest_event(self, m):
+        """Check that the latest event is always obtained."""
+        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)
+
+        activity_1 = DEVICE_ACTIVITIES.get_response_ok(
+            dev_id=DEVICE.DEVID,
+            event=CONST.EVENT_BUTTON,
+            state='alpha',
+            created_at=datetime.datetime(2017, 1, 1, 0, 0, 0))
+
+        activity_2 = DEVICE_ACTIVITIES.get_response_ok(
+            dev_id=DEVICE.DEVID,
+            event=CONST.EVENT_BUTTON,
+            state='beta',
+            created_at=datetime.datetime(2017, 1, 1, 0, 0, 1))
+
+        activities_text = '[' + activity_1 + ',' + activity_2 + ']'
+
+        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 latest button event
+        event = device.latest(CONST.EVENT_BUTTON)
+
+        # Test
+        self.assertIsNotNone(event)
+        self.assertEqual(event.get(CONST.STATE), 'beta')
+
+    @requests_mock.mock()
+    def tests_newest_event_cached(self, m):
+        """Check that the a newer cached event is kept over an older event."""
+        m.post(CONST.LOGIN_URL, text=LOGIN.post_response_ok())
+
+        # Set up device
+        device = DEVICE.get_response_ok()
+        device_text = '[' + device + ']'
+        device_url = str.replace(CONST.DEVICE_URL, '$DEVID$', DEVICE.DEVID)
+
+        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)
+
+        activity_1 = DEVICE_ACTIVITIES.get_response_ok(
+            dev_id=DEVICE.DEVID,
+            event=CONST.EVENT_BUTTON,
+            state='alpha',
+            created_at=datetime.datetime(2017, 1, 1, 0, 0, 0))
+
+        activities_url = str.replace(CONST.DEVICE_ACTIVITIES_URL,
+                                     '$DEVID$', DEVICE.DEVID)
+
+        m.get(CONST.DEVICES_URL, text=device_text)
+        m.get(device_url, text=device)
+        m.get(info_url, text=info_text)
+        m.get(settings_url, text=settings_text)
+        m.get(activities_url, text='[' + activity_1 + ']')
+
+        # Logout to reset everything
+        self.skybell.logout()
+
+        # Get our specific device
+        device = self.skybell.get_device(DEVICE.DEVID)
+        self.assertIsNotNone(device)
+
+        # Get latest button event
+        event = device.latest(CONST.EVENT_BUTTON)
+
+        # Test
+        self.assertIsNotNone(event)
+        self.assertEqual(event.get(CONST.STATE), 'alpha')
+
+        activity_2 = DEVICE_ACTIVITIES.get_response_ok(
+            dev_id=DEVICE.DEVID,
+            event=CONST.EVENT_BUTTON,
+            state='beta',
+            created_at=datetime.datetime(2014, 1, 1, 0, 0, 1))
+
+        m.get(activities_url, text='[' + activity_2 + ']')
+
+        # Refresh device
+        device.refresh()
+
+        # Get latest button event
+        event = device.latest(CONST.EVENT_BUTTON)
+
+        # Test
+        self.assertIsNotNone(event)
+        self.assertEqual(event.get(CONST.STATE), 'alpha')