diff --git a/skybellpy/__init__.py b/skybellpy/__init__.py index 5b157d8..46d24ee 100644 --- a/skybellpy/__init__.py +++ b/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): diff --git a/skybellpy/device.py b/skybellpy/device.py index d44fdd3..16a5830 100644 --- a/skybellpy/device.py +++ b/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(): diff --git a/skybellpy/utils.py b/skybellpy/utils.py index 670dac2..ade7c7b 100644 --- a/skybellpy/utils.py +++ b/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) diff --git a/tests/mock/device_activities.py b/tests/mock/device_activities.py index 9b86c5b..82c6459 100644 --- a/tests/mock/device_activities.py +++ b/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", diff --git a/tests/test_device.py b/tests/test_device.py index 24ace6c..63ee79b 100644 --- a/tests/test_device.py +++ b/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')