diff --git a/CHANGES.rst b/CHANGES.rst new file mode 100644 index 0000000..f9b04bd --- /dev/null +++ b/CHANGES.rst @@ -0,0 +1,53 @@ +# Changelog +----------- + +A list of changes between each release + +0.6.0.dev0 (unreleased) +^^^^^^^^^^^^^^^^^^ +- Removed redundent properties that only called hidden variables +- Revised request wrapper function to be more intelligent +- Added tests to ensure exceptions are caught and handled (100% coverage!) + +0.5.2 (2017-03-12) +^^^^^^^^^^^^^^^^^^ +- Fixed packaging mishap, same as 0.5.0 otherwise + +0.5.0 (2017-03-12) +^^^^^^^^^^^^^^^^^^ +- Fixed region handling problem +- Added rest.piri subdomain as a backup if region can't be found +- Improved the file writing function +- Large test coverage increase + +0.4.4 (2017-03-06) +^^^^^^^^^^^^^^^^^^ +- Fixed bug where region id was not being set in the header + +0.4.3 (2017-03-05) +^^^^^^^^^^^^^^^^^^ +- Changed to bdist_wheel release + +0.4.2 (2017-01-28) +^^^^^^^^^^^^^^^^^^ +- Fixed inability to retrieve motion data due to Key Error + +0.4.1 (2017-01-27) +^^^^^^^^^^^^^^^^^^ +- Fixed refresh bug (0.3.1 did not actually fix the problem) +- Image refresh routine added (per camera) +- Dictionary of thumbnails per camera added +- Improved test coverage + +0.3.1 (2017-01-25) +^^^^^^^^^^^^^^^^^^ +- Fixed refresh bug (Key Error) + +0.3.0 (2017-01-25) +^^^^^^^^^^^^^^^^^^ +- Added device id to camera lookup table +- Added image to file method + +0.2.0 (2017-01-21) +^^^^^^^^^^^^^^^^^^ +- Initial release of blinkpy \ No newline at end of file diff --git a/blinkpy.py b/blinkpy.py index 6391c56..4ec5a75 100644 --- a/blinkpy.py +++ b/blinkpy.py @@ -25,26 +25,23 @@ from helpers.constants import (BLINK_URL, LOGIN_URL, def _request(url, data=None, headers=None, reqtype='get', stream=False, json_resp=True): """Wrapper function for request.""" - if reqtype is 'post' and json_resp: - response = requests.post(url, headers=headers, - data=data).json() - elif reqtype is 'post' and not json_resp: + if reqtype == 'post': response = requests.post(url, headers=headers, data=data) - elif reqtype is 'get' and json_resp: - response = requests.get(url, headers=headers, - stream=stream).json() - elif reqtype is 'get' and not json_resp: + elif reqtype == 'get': response = requests.get(url, headers=headers, stream=stream) else: raise BlinkException(ERROR.REQUEST) - if json_resp and 'code' in response: + if json_resp and 'code' in response.json(): raise BlinkAuthenticationException( - (response['code'], response['message'])) + (response.json()['code'], response.json()['message'])) - return response + if json_resp: + return response.json() + else: + return response # pylint: disable=super-init-not-called @@ -81,169 +78,58 @@ class BlinkCamera(object): def __init__(self, config, urls): """Initiailize BlinkCamera.""" self.urls = urls - self._id = str(config['device_id']) - self._name = config['name'] + self.id = str(config['device_id']) # pylint: disable=invalid-name + self.name = config['name'] self._status = config['armed'] - self._thumb = self.urls.base_url + config['thumbnail'] + '.jpg' - self._clip = self.urls.base_url + config['thumbnail'] + '.mp4' - self._temperature = config['temp'] - self._battery = config['battery'] - self._notifications = config['notifications'] - self._motion = {} - self._header = None - self._image_link = None - self._arm_link = None - self._region_id = config['region_id'] - - @property - # pylint: disable=invalid-name - def id(self): - """Return camera id.""" - return self._id - - @property - def name(self): - """Return camera name.""" - return self._name - - @name.setter - def name(self, value): - """Set camera name.""" - self._name = value - - @property - def region_id(self): - """Return region id.""" - return self._region_id + self.thumbnail = self.urls.base_url + config['thumbnail'] + '.jpg' + self.clip = self.urls.base_url + config['thumbnail'] + '.mp4' + self.temperature = config['temp'] + self.battery = config['battery'] + self.notifications = config['notifications'] + self.motion = {} + self.header = None + self.image_link = None + self.arm_link = None + self.region_id = config['region_id'] @property def armed(self): """Return camera arm status.""" return self._status - @property - def clip(self): - """Return current clip.""" - return self._clip - - @clip.setter - def clip(self, value): - """Set current clip.""" - self._clip = value - - @property - def thumbnail(self): - """Return current thumbnail.""" - return self._thumb - - @thumbnail.setter - def thumbnail(self, value): - """Set current thumbnail.""" - self._thumb = value - - @property - def temperature(self): - """Return camera temperature.""" - return self._temperature - - @temperature.setter - def temperature(self, value): - """Set camera temperature.""" - self._temperature = value - - @property - def battery(self): - """Return battery level.""" - return self._battery - - @battery.setter - def battery(self, value): - """Set battery level.""" - self._battery = value - - @property - def notifications(self): - """Return number of notifications.""" - return self._notifications - - @notifications.setter - def notifications(self, value): - """Set number of notifications.""" - self._notifications = value - - @property - def image_link(self): - """Return image link.""" - return self._image_link - - @image_link.setter - def image_link(self, value): - """Set image link.""" - self._image_link = value - - @property - def arm_link(self): - """Return link to arm camera.""" - return self._arm_link - - @arm_link.setter - def arm_link(self, value): - """Set link to arm camera.""" - self._arm_link = value - - @property - def header(self): - """Return request header.""" - return self._header - - @header.setter - def header(self, value): - """Set request header.""" - self._header = value - - @property - def motion(self): - """Return last motion event detail.""" - return self._motion - - @motion.setter - def motion(self, value): - """Set link to last motion and timestamp.""" - self._motion = value - def snap_picture(self): """Take a picture with camera to create a new thumbnail.""" - _request(self._image_link, headers=self._header, reqtype='post') + _request(self.image_link, headers=self.header, reqtype='post') def set_motion_detect(self, enable): """Set motion detection.""" - url = self._arm_link + url = self.arm_link if enable: - _request(url + 'enable', headers=self._header, reqtype='post') + _request(url + 'enable', headers=self.header, reqtype='post') else: - _request(url + 'disable', headers=self._header, reqtype='post') + _request(url + 'disable', headers=self.header, reqtype='post') def update(self, values): """Update camera information.""" - self._name = values['name'] + self.name = values['name'] self._status = values['armed'] - self._thumb = self.urls.base_url + values['thumbnail'] + '.jpg' - self._clip = self.urls.base_url + values['thumbnail'] + '.mp4' - self._temperature = values['temp'] - self._battery = values['battery'] - self._notifications = values['notifications'] + self.thumbnail = self.urls.base_url + values['thumbnail'] + '.jpg' + self.clip = self.urls.base_url + values['thumbnail'] + '.mp4' + self.temperature = values['temp'] + self.battery = values['battery'] + self.notifications = values['notifications'] def image_refresh(self): """Refresh current thumbnail.""" url = self.urls.home_url - response = _request(url, headers=self._header, + response = _request(url, headers=self.header, reqtype='get')['devices'] for element in response: try: - if str(element['device_id']) == self._id: - self._thumb = (self.urls.base_url + - element['thumbnail'] + '.jpg') - return self._thumb + if str(element['device_id']) == self.id: + self.thumbnail = (self.urls.base_url + + element['thumbnail'] + '.jpg') + return self.thumbnail except KeyError: pass return None @@ -251,7 +137,7 @@ class BlinkCamera(object): def image_to_file(self, path): """Write image to file.""" thumb = self.image_refresh() - response = _request(thumb, headers=self._header, + response = _request(thumb, headers=self.header, reqtype='get', stream=True, json_resp=False) if response.status_code == 200: with open(path, 'wb') as imgfile: @@ -267,27 +153,22 @@ class Blink(object): self._password = password self._token = None self._auth_header = None - self._network_id = None - self._account_id = None - self._region = None - self._region_id = None + self.network_id = None + self.account_id = None + self.region = None + self.region_id = None self._host = None self._events = [] - self._cameras = {} + self.cameras = {} self._idlookup = {} self.urls = None - @property - def cameras(self): - """Return camera/id pairs.""" - return self._cameras - @property def camera_thumbs(self): """Return camera thumbnails.""" self.refresh() data = {} - for name, camera in self._cameras.items(): + for name, camera in self.cameras.items(): data[name] = camera.thumbnail return data @@ -297,30 +178,10 @@ class Blink(object): """Return id/camera pairs.""" return self._idlookup - @property - def network_id(self): - """Return network id.""" - return self._network_id - - @property - def account_id(self): - """Return account id.""" - return self._account_id - - @property - def region(self): - """Return current region.""" - return self._region - - @property - def region_id(self): - """Return region id.""" - return self._region_id - @property def events(self): """Get all events on server.""" - url = self.urls.event_url + self._network_id + url = self.urls.event_url + self.network_id headers = self._auth_header self._events = _request(url, headers=headers, reqtype='get')['event'] @@ -329,7 +190,7 @@ class Blink(object): @property def online(self): """Return boolean system online status.""" - url = self.urls.network_url + self._network_id + '/syncmodules' + url = self.urls.network_url + self.network_id + '/syncmodules' headers = self._auth_header return ONLINE[_request(url, headers=headers, reqtype='get')['syncmodule']['status']] @@ -340,8 +201,8 @@ class Blink(object): for element in recent: try: camera_id = str(element['camera_id']) - camera_name = self._idlookup[camera_id] - camera = self._cameras[camera_name] + camera_name = self.id_table[camera_id] + camera = self.cameras[camera_name] if element['type'] == 'motion': url = self.urls.base_url + element['video_url'] camera.motion = {'video': url, @@ -362,21 +223,22 @@ class Blink(object): value_to_append = 'arm' else: value_to_append = 'disarm' - url = self.urls.network_url + self._network_id + '/' + value_to_append + url = self.urls.network_url + self.network_id + '/' + value_to_append _request(url, headers=self._auth_header, reqtype='post') def refresh(self): """Get all blink cameras and pulls their most recent status.""" response = self.get_summary()['devices'] - for name in self._cameras: - camera = self._cameras[name] + for name in self.cameras: + camera = self.cameras[name] for element in response: try: if str(element['device_id']) == camera.id: camera.update(element) except KeyError: pass + return None def get_summary(self): """Get a full summary of device information.""" @@ -395,16 +257,16 @@ class Blink(object): if ('device_type' in element and element['device_type'] == 'camera'): # Add region to config - element['region_id'] = self._region_id + element['region_id'] = self.region_id device = BlinkCamera(element, self.urls) - self._cameras[device.name] = device + self.cameras[device.name] = device self._idlookup[device.id] = device.name def set_links(self): """Set access links and required headers for each camera in system.""" - for name in self._cameras: - camera = self._cameras[name] - network_id_url = self.urls.network_url + self._network_id + for name in self.cameras: + camera = self.cameras[name] + network_id_url = self.urls.network_url + self.network_id image_url = network_id_url + '/camera/' + camera.id + '/thumbnail' arm_url = network_id_url + '/camera/' + camera.id + '/' camera.image_link = image_url @@ -448,20 +310,20 @@ class Blink(object): data=data, json_resp=False, reqtype='post') if response.status_code is 200: response = response.json() - (self._region_id, self._region), = response['region'].items() + (self.region_id, self.region), = response['region'].items() else: response = _request(LOGIN_BACKUP_URL, headers=headers, data=data, reqtype='post') - self._region_id = 'rest.piri' - self._region = "UNKNOWN" + self.region_id = 'rest.piri' + self.region = "UNKNOWN" - self._host = self._region_id + '.' + BLINK_URL + self._host = self.region_id + '.' + BLINK_URL self._token = response['authtoken']['authtoken'] self._auth_header = {'Host': self._host, 'TOKEN_AUTH': self._token} - self.urls = BlinkURLHandler(self._region_id) + self.urls = BlinkURLHandler(self.region_id) def get_ids(self): """Set the network ID and Account ID.""" @@ -472,5 +334,5 @@ class Blink(object): raise BlinkException(ERROR.AUTH_TOKEN) response = _request(url, headers=headers, reqtype='get') - self._network_id = str(response['networks'][0]['id']) - self._account_id = str(response['networks'][0]['account_id']) + self.network_id = str(response['networks'][0]['id']) + self.account_id = str(response['networks'][0]['account_id']) diff --git a/helpers/constants.py b/helpers/constants.py index ae51d75..b323e12 100644 --- a/helpers/constants.py +++ b/helpers/constants.py @@ -5,8 +5,8 @@ Generates constants for use in blinkpy import os MAJOR_VERSION = 0 -MINOR_VERSION = 5 -PATCH_VERSION = '3.dev0' +MINOR_VERSION = 6 +PATCH_VERSION = '0.dev0' __version__ = '{}.{}.{}'.format(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION) diff --git a/requirements_test.txt b/requirements_test.txt index 27b840b..b21ce85 100644 --- a/requirements_test.txt +++ b/requirements_test.txt @@ -1,3 +1,5 @@ flake8==3.3 pylint==1.6.5 -pydocstyle==1.1.1 \ No newline at end of file +pydocstyle==1.1.1 +pytest>=2.9.2 +pytest-sugar>=0.8.0 \ No newline at end of file diff --git a/tests/mock_responses.py b/tests/mock_responses.py index 3161b43..346dc56 100644 --- a/tests/mock_responses.py +++ b/tests/mock_responses.py @@ -1,274 +1,286 @@ -""" -Mock responses that mimic actual responses from Blink servers. - -This file should be updated any time the Blink server responses -change so we can make sure blinkpy can still communicate. -""" - -import helpers.constants as const - -NETWORKS_RESPONSE = {} -NETWORKS_RESPONSE['summary'] = {'onboarded': True, 'name': 'Nilfgaard'} -NETWORKS_RESPONSE['networks'] = [{ - 'network_key': None, - 'name': NETWORKS_RESPONSE['summary']['name'], - 'account_id': 1989, - 'id': 9898, - 'encryption_key': None, - 'armed': True, - 'ping_interval': 60, - 'video_destination': 'server', - 'arm_string': 'Armed', - 'feature_plan_id': None -}] - -NEW_THUMBNAIL = '/NEW/THUMBNAIL/YAY' - -FIRST_CAMERA = {'device_type': 'camera', - 'notifications': 1, - 'battery': 2, - 'active': 'enabled', - 'enabled': True, - 'temp': 70, - 'updated_at': '2017-01-01T01:23:45+00:00', - 'lfr_strength': 3, - 'armed': True, - 'device_id': 2112, - 'wifi_strength': 5, - 'thumbnail': '/this/is/url1/thumb', - 'name': 'First Camera', - 'status': 'done'} - -SECOND_CAMERA = {'device_type': 'camera', - 'notifications': 0, - 'battery': 3, - 'active': 'disabled', - 'enabled': False, - 'temp': 82, - 'updated_at': '2017-12-21T01:21:12+00:00', - 'lfr_strength': 1, - 'armed': False, - 'device_id': 1221, - 'wifi_strength': 1, - 'thumbnail': '/this/is/url2/thumb', - 'name': 'Second Camera', - 'status': 'done'} - -SYNC_MODULE = {'updated_at': '1970-01-01T01:00:00+00:00', - 'device_type': 'sync_module', - 'notifications': 2, - 'device_id': 7990, - 'status': 'online'} - -FIRST_EVENT = {'camera_name': FIRST_CAMERA['name'], - 'updated_at': '2017-10-10T03:37:37+00:00', - 'sync_module_id': None, - 'camera': FIRST_CAMERA['device_id'], - 'type': 'motion', - 'duration': None, - 'status': None, - 'created_at': '2017-01-28T19:51:52+00:00', - 'camera_id': FIRST_CAMERA['device_id'], - 'id': 666777888, - 'siren_id': None, - 'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'], - 'notified': True, - 'siren': None, - 'syncmodule': None, - 'video_url': FIRST_CAMERA['thumbnail'] + '.mp4', - 'command_id': None, - 'network_id': None, - 'account': NETWORKS_RESPONSE['networks'][0]['account_id'], - 'video_id': 123000321} - -SECOND_EVENT = {'updated_at': '2017-01-28T19:51:29+00:00', - 'sync_module_id': SYNC_MODULE['device_id'], - 'camera': None, - 'type': 'armed', - 'duration': None, - 'status': None, - 'created_at': '2017-01-28T19:51:29+00:00', - 'camera_id': None, - 'id': 333777555, - 'siren_id': None, - 'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'], - 'notified': False, - 'siren': None, - 'syncmodule': SYNC_MODULE['device_id'], - 'command_id': None, - 'network_id': None, - 'account': NETWORKS_RESPONSE['networks'][0]['account_id']} - -"""Fake response content.""" -LOGIN_RESPONSE = {} -LOGIN_RESPONSE['region'] = {'ciri': 'Cintra'} -LOGIN_RESPONSE['networks'] = { - NETWORKS_RESPONSE['networks'][0]['id']: NETWORKS_RESPONSE['summary'] -} -LOGIN_RESPONSE['authtoken'] = {'authtoken': 'foobar7117', 'message': 'auth'} - -RESPONSE = {} -RESPONSE['account'] = {'notifications': SYNC_MODULE['notifications']} -RESPONSE['devices'] = [FIRST_CAMERA, SECOND_CAMERA, SYNC_MODULE] -RESPONSE['network'] = {'armed': NETWORKS_RESPONSE['networks'][0]['armed'], - 'wifi_strength': 4, - 'warning': 0, - 'name': NETWORKS_RESPONSE['summary']['name'], - 'notifications': SYNC_MODULE['notifications']} -RESPONSE['event'] = [FIRST_EVENT, SECOND_EVENT] -RESPONSE['syncmodule'] = {'name': 'Vengerberg', 'status': 'online'} - -MOCK_BYTES = '\x00\x10JFIF\x00\x01' - -IMAGE_TO_WRITE_URL = list() -IMAGE_TO_WRITE_URL.append('https://ciri.' + const.BLINK_URL + - FIRST_CAMERA['thumbnail'] + '.jpg') -IMAGE_TO_WRITE_URL.append('https://ciri.' + const.BLINK_URL + - SECOND_CAMERA['thumbnail'] + '.jpg') - -FAKE_FILES = list() - - -def mocked_requests_post(*args, **kwargs): - """Mock post request.""" - class MockPostResponse: - """Class for mock post response.""" - - def __init__(self, json_data, status_code): - """Initialze mock post response.""" - self.json_data = json_data - self.status_code = status_code - - def json(self): - """Return json data from post request.""" - return self.json_data - - # pylint: disable=global-variable-not-assigned - global RESPONSE - # pylint: disable=global-variable-not-assigned - global NETWORKS_RESPONSE - - if args[0] is not None: - url_tail = args[0].split("/")[-1] - else: - return MockPostResponse({'message': 'ERROR', 'code': 404}, 404) - - if args[0] == const.LOGIN_URL: - # Request to login - return MockPostResponse(LOGIN_RESPONSE, 200) - elif args[0] == const.LOGIN_BACKUP_URL: - return MockPostResponse(LOGIN_RESPONSE, 200) - elif url_tail == 'arm' or url_tail == 'disarm': - # Request to arm/disarm system - NETWORKS_RESPONSE['networks'][0]['armed'] = url_tail == 'arm' - RESPONSE['network']['armed'] = url_tail == 'arm' - return MockPostResponse({}, 200) - elif url_tail == 'enable' or url_tail == 'disable': - # Request to enable/disable motion detection per camera - received_id = args[0].split("/")[-2] - all_devices = list() - for element in RESPONSE['devices']: - all_devices.append(element) - current_id = element['device_id'] - if str(current_id) == received_id: - element['armed'] = url_tail == 'enable' - element['enabled'] = url_tail == 'enable' - RESPONSE['devices'] = all_devices - return MockPostResponse({}, 200) - elif url_tail == 'thumbnail': - # Requesting a new image - received_id = args[0].split("/")[-2] - all_devices = list() - for element in RESPONSE['devices']: - all_devices.append(element) - if str(element['device_id']) == received_id: - element['thumbnail'] = NEW_THUMBNAIL - RESPONSE['devices'] = all_devices - return MockPostResponse({}, 200) - - return MockPostResponse({'message': 'ERROR', 'code': 404}, 404) - - -def mocked_requests_get(*args, **kwargs): - """Mock get request.""" - class MockGetResponse: - """Class for mock get response.""" - - def __init__(self, json_data, status_code, raw_data=None): - """Initialize mock get response.""" - self.json_data = json_data - self.status_code = status_code - self.raw_data = raw_data - - def json(self): - """Return json data from get request.""" - return self.json_data - - @property - def raw(self): - """Return raw data from get request.""" - return self.raw_data - - # pylint: disable=unused-variable - (region_id, region), = LOGIN_RESPONSE['region'].items() - set_region_id = args[0].split('/')[2].split('.')[0] - if set_region_id == 'rest': - set_region_id = (set_region_id + '.' + - args[0].split('/')[2].split('.')[1]) - region_id = 'rest.piri' - neturl = 'https://' + set_region_id + '.' + const.BLINK_URL + '/networks' - if args[0] == neturl: - return MockGetResponse(NETWORKS_RESPONSE, 200) - elif set_region_id != region_id: - raise ConnectionError('Received region id ' + region_id + - ' Expected ' + set_region_id) - elif args[0] in IMAGE_TO_WRITE_URL: - return MockGetResponse({}, 200, raw_data=MOCK_BYTES) - else: - return MockGetResponse(RESPONSE, 200) - - return MockGetResponse({'message': 'ERROR', 'code': 404}, 404) - - -def mocked_copyfileobj(*args, **kwargs): - """Mock shutil.copyfileobj.""" - class MockCopyFileObj: - """Class for mock copy file.""" - - def __init__(self, src, dst): - """Initialize copyfile mock.""" - self.src = src - self.dst = dst - # pylint: disable=global-variable-not-assigned - global FAKE_FILES - mockobj = MockCopyFileObj(args[0], args[1]) - FAKE_FILES.append(mockobj.src) - return - - -def get_test_cameras(base_url): - """Helper function to return cameras named in this file.""" - test_cameras = dict() - for element in RESPONSE['devices']: - if ('device_type' in element and - element['device_type'] == 'camera'): - test_cameras[element['name']] = { - 'device_id': str(element['device_id']), - 'armed': element['armed'], - 'thumbnail': (base_url + - element['thumbnail'] + '.jpg'), - 'temperature': element['temp'], - 'battery': element['battery'], - 'notifications': element['notifications'] - } - return test_cameras - - -def get_test_id_table(): - """Helper function to return mock id table.""" - test_id_table = dict() - for element in RESPONSE['devices']: - if ('device_type' in element and - element['device_type'] == 'camera'): - test_id_table[str(element['device_id'])] = element['name'] - return test_id_table +""" +Mock responses that mimic actual responses from Blink servers. + +This file should be updated any time the Blink server responses +change so we can make sure blinkpy can still communicate. +""" + +import helpers.constants as const + +NETWORKS_RESPONSE = {} +NETWORKS_RESPONSE['summary'] = {'onboarded': True, 'name': 'Nilfgaard'} +NETWORKS_RESPONSE['networks'] = [{ + 'network_key': None, + 'name': NETWORKS_RESPONSE['summary']['name'], + 'account_id': 1989, + 'id': 9898, + 'encryption_key': None, + 'armed': True, + 'ping_interval': 60, + 'video_destination': 'server', + 'arm_string': 'Armed', + 'feature_plan_id': None +}] + +NEW_THUMBNAIL = '/NEW/THUMBNAIL/YAY' + +FIRST_CAMERA = {'device_type': 'camera', + 'notifications': 1, + 'battery': 2, + 'active': 'enabled', + 'enabled': True, + 'temp': 70, + 'updated_at': '2017-01-01T01:23:45+00:00', + 'lfr_strength': 3, + 'armed': True, + 'device_id': 2112, + 'wifi_strength': 5, + 'thumbnail': '/this/is/url1/thumb', + 'name': 'First Camera', + 'status': 'done'} + +SECOND_CAMERA = {'device_type': 'camera', + 'notifications': 0, + 'battery': 3, + 'active': 'disabled', + 'enabled': False, + 'temp': 82, + 'updated_at': '2017-12-21T01:21:12+00:00', + 'lfr_strength': 1, + 'armed': False, + 'device_id': 1221, + 'wifi_strength': 1, + 'thumbnail': '/this/is/url2/thumb', + 'name': 'Second Camera', + 'status': 'done'} + +SYNC_MODULE = {'updated_at': '1970-01-01T01:00:00+00:00', + 'device_type': 'sync_module', + 'notifications': 2, + 'device_id': 7990, + 'status': 'online'} + +FIRST_EVENT = {'camera_name': FIRST_CAMERA['name'], + 'updated_at': '2017-10-10T03:37:37+00:00', + 'sync_module_id': None, + 'camera': FIRST_CAMERA['device_id'], + 'type': 'motion', + 'duration': None, + 'status': None, + 'created_at': '2017-01-28T19:51:52+00:00', + 'camera_id': FIRST_CAMERA['device_id'], + 'id': 666777888, + 'siren_id': None, + 'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'], + 'notified': True, + 'siren': None, + 'syncmodule': None, + 'video_url': FIRST_CAMERA['thumbnail'] + '.mp4', + 'command_id': None, + 'network_id': None, + 'account': NETWORKS_RESPONSE['networks'][0]['account_id'], + 'video_id': 123000321} + +SECOND_EVENT = {'updated_at': '2017-01-28T19:51:29+00:00', + 'sync_module_id': SYNC_MODULE['device_id'], + 'camera': None, + 'type': 'armed', + 'duration': None, + 'status': None, + 'created_at': '2017-01-28T19:51:29+00:00', + 'camera_id': None, + 'id': 333777555, + 'siren_id': None, + 'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'], + 'notified': False, + 'siren': None, + 'syncmodule': SYNC_MODULE['device_id'], + 'command_id': None, + 'network_id': None, + 'account': NETWORKS_RESPONSE['networks'][0]['account_id']} + +"""Fake response content.""" +LOGIN_RESPONSE = {} +LOGIN_RESPONSE['region'] = {'ciri': 'Cintra'} +LOGIN_RESPONSE['networks'] = { + NETWORKS_RESPONSE['networks'][0]['id']: NETWORKS_RESPONSE['summary'] +} +LOGIN_RESPONSE['authtoken'] = {'authtoken': 'foobar7117', 'message': 'auth'} + +RESPONSE = {} +RESPONSE['account'] = {'notifications': SYNC_MODULE['notifications']} +RESPONSE['devices'] = [FIRST_CAMERA, SECOND_CAMERA, SYNC_MODULE] +RESPONSE['network'] = {'armed': NETWORKS_RESPONSE['networks'][0]['armed'], + 'wifi_strength': 4, + 'warning': 0, + 'name': NETWORKS_RESPONSE['summary']['name'], + 'notifications': SYNC_MODULE['notifications']} +RESPONSE['event'] = [FIRST_EVENT, SECOND_EVENT] +RESPONSE['syncmodule'] = {'name': 'Vengerberg', 'status': 'online'} + +BAD_RESPONSE = {} +BAD_RESPONSE['account'] = {'nothing': 'here'} +BAD_RESPONSE['devices'] = [{'foo': 'bar', 'device_type': 'camera'}, + {'device_type': 'camera', 'device_id': 2112}] +BAD_RESPONSE['network'] = {'bar': 'foo'} + +MOCK_BYTES = '\x00\x10JFIF\x00\x01' + +IMAGE_TO_WRITE_URL = list() +IMAGE_TO_WRITE_URL.append('https://ciri.' + const.BLINK_URL + + FIRST_CAMERA['thumbnail'] + '.jpg') +IMAGE_TO_WRITE_URL.append('https://ciri.' + const.BLINK_URL + + SECOND_CAMERA['thumbnail'] + '.jpg') + +FAKE_FILES = list() + + +def mocked_requests_post(*args, **kwargs): + """Mock post request.""" + class MockPostResponse: + """Class for mock post response.""" + + def __init__(self, json_data, status_code): + """Initialze mock post response.""" + self.json_data = json_data + self.status_code = status_code + + def json(self): + """Return json data from post request.""" + return self.json_data + + # pylint: disable=global-variable-not-assigned + global RESPONSE + # pylint: disable=global-variable-not-assigned + global NETWORKS_RESPONSE + + if args[0] is not None: + url_tail = args[0].split("/")[-1] + else: + return MockPostResponse({'message': 'ERROR', 'code': 404}, 404) + + if args[0] == const.LOGIN_URL: + # Request to login + return MockPostResponse(LOGIN_RESPONSE, 200) + elif args[0] == const.LOGIN_BACKUP_URL: + return MockPostResponse(LOGIN_RESPONSE, 200) + elif url_tail == 'arm' or url_tail == 'disarm': + # Request to arm/disarm system + NETWORKS_RESPONSE['networks'][0]['armed'] = url_tail == 'arm' + RESPONSE['network']['armed'] = url_tail == 'arm' + return MockPostResponse({}, 200) + elif url_tail == 'enable' or url_tail == 'disable': + # Request to enable/disable motion detection per camera + received_id = args[0].split("/")[-2] + all_devices = list() + for element in RESPONSE['devices']: + all_devices.append(element) + current_id = element['device_id'] + if str(current_id) == received_id: + element['armed'] = url_tail == 'enable' + element['enabled'] = url_tail == 'enable' + RESPONSE['devices'] = all_devices + return MockPostResponse({}, 200) + elif url_tail == 'thumbnail': + # Requesting a new image + received_id = args[0].split("/")[-2] + all_devices = list() + for element in RESPONSE['devices']: + all_devices.append(element) + if str(element['device_id']) == received_id: + element['thumbnail'] = NEW_THUMBNAIL + RESPONSE['devices'] = all_devices + return MockPostResponse({}, 200) + + return MockPostResponse({'message': 'ERROR', 'code': 404}, 404) + + +def mocked_requests_get(*args, **kwargs): + """Mock get request.""" + class MockGetResponse: + """Class for mock get response.""" + + def __init__(self, json_data, status_code, raw_data=None): + """Initialize mock get response.""" + self.json_data = json_data + self.status_code = status_code + self.raw_data = raw_data + + def json(self): + """Return json data from get request.""" + return self.json_data + + @property + def raw(self): + """Return raw data from get request.""" + return self.raw_data + + # pylint: disable=unused-variable + (region_id, region), = LOGIN_RESPONSE['region'].items() + if args[0] != 'use_bad_response': + set_region_id = args[0].split('/')[2].split('.')[0] + else: + set_region_id = 'ciri' + + if set_region_id == 'rest': + set_region_id = (set_region_id + '.' + + args[0].split('/')[2].split('.')[1]) + region_id = 'rest.piri' + neturl = 'https://' + set_region_id + '.' + const.BLINK_URL + '/networks' + if args[0] == neturl: + return MockGetResponse(NETWORKS_RESPONSE, 200) + elif set_region_id != region_id: + raise ConnectionError('Received region id ' + region_id + + ' Expected ' + set_region_id) + elif args[0] in IMAGE_TO_WRITE_URL: + return MockGetResponse({}, 200, raw_data=MOCK_BYTES) + elif args[0] == 'use_bad_response': + return MockGetResponse(BAD_RESPONSE, 200) + else: + return MockGetResponse(RESPONSE, 200) + + return MockGetResponse({'message': 'ERROR', 'code': 404}, 404) + + +def mocked_copyfileobj(*args, **kwargs): + """Mock shutil.copyfileobj.""" + class MockCopyFileObj: + """Class for mock copy file.""" + + def __init__(self, src, dst): + """Initialize copyfile mock.""" + self.src = src + self.dst = dst + # pylint: disable=global-variable-not-assigned + global FAKE_FILES + mockobj = MockCopyFileObj(args[0], args[1]) + FAKE_FILES.append(mockobj.src) + return + + +def get_test_cameras(base_url): + """Helper function to return cameras named in this file.""" + test_cameras = dict() + for element in RESPONSE['devices']: + if ('device_type' in element and + element['device_type'] == 'camera'): + test_cameras[element['name']] = { + 'device_id': str(element['device_id']), + 'armed': element['armed'], + 'thumbnail': (base_url + + element['thumbnail'] + '.jpg'), + 'temperature': element['temp'], + 'battery': element['battery'], + 'notifications': element['notifications'] + } + return test_cameras + + +def get_test_id_table(): + """Helper function to return mock id table.""" + test_id_table = dict() + for element in RESPONSE['devices']: + if ('device_type' in element and + element['device_type'] == 'camera'): + test_id_table[str(element['device_id'])] = element['name'] + return test_id_table diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index ad3b6f9..a4685e5 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -1,159 +1,172 @@ -"""Tests camera and system functions.""" - -import unittest -from unittest import mock -import blinkpy -import tests.mock_responses as mresp - -USERNAME = 'foobar' -PASSWORD = 'deadbeef' - - -class TestBlinkFunctions(unittest.TestCase): - """Test Blink and BlinkCamera functions in blinkpy.""" - - def setUp(self): - """Set up Blink module.""" - self.blink = blinkpy.Blink(username=USERNAME, - password=PASSWORD) - (self.region_id, self.region), = mresp.LOGIN_RESPONSE['region'].items() - self.test_urls = blinkpy.BlinkURLHandler(self.region_id) - - def tearDown(self): - """Clean up after test.""" - self.blink = None - self.region = None - self.region_id = None - self.test_urls = None - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_set_motion_detect(self, mock_get, mock_post): - """Checks if we can set motion detection.""" - self.blink.setup_system() - self.test_urls = blinkpy.BlinkURLHandler(self.region_id) - test_cameras = mresp.get_test_cameras(self.test_urls.base_url) - for camera_name in test_cameras: - self.blink.cameras[camera_name].set_motion_detect(True) - self.blink.refresh() - self.assertEqual(self.blink.cameras[camera_name].armed, True) - self.blink.cameras[camera_name].set_motion_detect(False) - self.blink.refresh() - self.assertEqual(self.blink.cameras[camera_name].armed, False) - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_last_motion(self, mock_get, mock_post): - """Checks that we can get the last motion info.""" - self.test_urls = blinkpy.BlinkURLHandler(self.region_id) - test_events = mresp.RESPONSE['event'] - test_video = dict() - test_image = dict() - test_time = dict() - for event in test_events: - if event['type'] == 'motion': - url = self.test_urls.base_url + event['video_url'] - test_video[event['camera_name']] = url - test_image[event['camera_name']] = url[:-3] + 'jpg' - test_time[event['camera_name']] = event['created_at'] - - self.blink.setup_system() - for name in self.blink.cameras: - camera = self.blink.cameras[name] - self.blink.last_motion() - if name in test_video: - self.assertEqual(camera.motion['video'], test_video[name]) - else: - self.assertEqual(camera.motion, {}) - if name in test_image: - self.assertEqual(camera.motion['image'], test_image[name]) - else: - self.assertEqual(camera.motion, {}) - if name in test_video: - self.assertEqual(camera.motion['time'], test_time[name]) - else: - self.assertEqual(camera.motion, {}) - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_take_new_picture(self, mock_get, mock_post): - """Checks if we can take a new picture and retrieve the thumbnail.""" - self.blink.setup_system() - test_cameras = mresp.get_test_cameras(self.test_urls.base_url) - test_thumbnail = self.test_urls.base_url + mresp.NEW_THUMBNAIL + '.jpg' - # Snap picture for each camera and check new thumb - for camera_name in test_cameras: - camera = self.blink.cameras[camera_name] - camera.snap_picture() - camera.image_refresh() - self.assertEqual(camera.thumbnail, test_thumbnail) - - # Manually set thumbnail, and then globally refresh and check - for camera_name in test_cameras: - camera = self.blink.cameras[camera_name] - camera.thumbnail = 'Testing' - self.assertEqual(camera.thumbnail, 'Testing') - - self.blink.refresh() - for camera_name in test_cameras: - camera = self.blink.cameras[camera_name] - self.assertEqual(camera.thumbnail, test_thumbnail) - - def test_camera_update(self): - """Checks that the update function is doing the right thing.""" - self.test_urls = blinkpy.BlinkURLHandler('test') - test_config = mresp.FIRST_CAMERA - test_camera = blinkpy.blinkpy.BlinkCamera(test_config, self.test_urls) - test_update = mresp.SECOND_CAMERA - test_camera.update(test_update) - test_image_url = self.test_urls.base_url + test_update['thumbnail'] - test_thumbnail = test_image_url + '.jpg' - test_clip = test_image_url + '.mp4' - self.assertEqual(test_camera.name, test_update['name']) - self.assertEqual(test_camera.armed, test_update['armed']) - self.assertEqual(test_camera.thumbnail, test_thumbnail) - self.assertEqual(test_camera.clip, test_clip) - self.assertEqual(test_camera.temperature, test_update['temp']) - self.assertEqual(test_camera.battery, test_update['battery']) - self.assertEqual(test_camera.notifications, - test_update['notifications']) - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_camera_thumbs(self, mock_get, mock_post): - """Checks to see if we can retrieve camera thumbs.""" - self.test_urls = blinkpy.BlinkURLHandler(self.region_id) - test_cameras = mresp.get_test_cameras(self.test_urls.base_url) - self.blink.setup_system() - for name in self.blink.cameras: - thumb = self.blink.camera_thumbs[name] - self.assertEqual(test_cameras[name]['thumbnail'], thumb) - - @mock.patch('blinkpy.blinkpy.copyfileobj', - side_effect=mresp.mocked_copyfileobj) - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_image_to_file(self, mock_get, mock_post, mock_copyfileobj): - """Checks that we can write an image to file.""" - self.blink.setup_system() - cameras = self.blink.cameras - filename = '/tmp/test.jpg' - test_files = list() - for camera_name in cameras: - camera = cameras[camera_name] - test_files.append(mresp.MOCK_BYTES) - mock_fh = mock.mock_open() - with mock.patch('builtins.open', mock_fh, create=True): - camera.image_to_file(camera_name + filename) - mock_fh.assert_called_once_with(camera_name + filename, 'wb') - self.assertEqual(test_files, mresp.FAKE_FILES) +"""Tests camera and system functions.""" + +import unittest +from unittest import mock +import blinkpy +import tests.mock_responses as mresp + +USERNAME = 'foobar' +PASSWORD = 'deadbeef' + + +class TestBlinkFunctions(unittest.TestCase): + """Test Blink and BlinkCamera functions in blinkpy.""" + + def setUp(self): + """Set up Blink module.""" + self.blink = blinkpy.Blink(username=USERNAME, + password=PASSWORD) + (self.region_id, self.region), = mresp.LOGIN_RESPONSE['region'].items() + self.test_urls = blinkpy.BlinkURLHandler(self.region_id) + + def tearDown(self): + """Clean up after test.""" + self.blink = None + self.region = None + self.region_id = None + self.test_urls = None + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_set_motion_detect(self, mock_get, mock_post): + """Checks if we can set motion detection.""" + self.blink.setup_system() + self.test_urls = blinkpy.BlinkURLHandler(self.region_id) + test_cameras = mresp.get_test_cameras(self.test_urls.base_url) + for camera_name in test_cameras: + self.blink.cameras[camera_name].set_motion_detect(True) + self.blink.refresh() + self.assertEqual(self.blink.cameras[camera_name].armed, True) + self.blink.cameras[camera_name].set_motion_detect(False) + self.blink.refresh() + self.assertEqual(self.blink.cameras[camera_name].armed, False) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_last_motion(self, mock_get, mock_post): + """Checks that we can get the last motion info.""" + self.test_urls = blinkpy.BlinkURLHandler(self.region_id) + test_events = mresp.RESPONSE['event'] + test_video = dict() + test_image = dict() + test_time = dict() + for event in test_events: + if event['type'] == 'motion': + url = self.test_urls.base_url + event['video_url'] + test_video[event['camera_name']] = url + test_image[event['camera_name']] = url[:-3] + 'jpg' + test_time[event['camera_name']] = event['created_at'] + + self.blink.setup_system() + for name in self.blink.cameras: + camera = self.blink.cameras[name] + self.blink.last_motion() + if name in test_video: + self.assertEqual(camera.motion['video'], test_video[name]) + else: + self.assertEqual(camera.motion, {}) + if name in test_image: + self.assertEqual(camera.motion['image'], test_image[name]) + else: + self.assertEqual(camera.motion, {}) + if name in test_video: + self.assertEqual(camera.motion['time'], test_time[name]) + else: + self.assertEqual(camera.motion, {}) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_take_new_picture(self, mock_get, mock_post): + """Checks if we can take a new picture and retrieve the thumbnail.""" + self.blink.setup_system() + test_cameras = mresp.get_test_cameras(self.test_urls.base_url) + test_thumbnail = self.test_urls.base_url + mresp.NEW_THUMBNAIL + '.jpg' + # Snap picture for each camera and check new thumb + for camera_name in test_cameras: + camera = self.blink.cameras[camera_name] + camera.snap_picture() + camera.image_refresh() + self.assertEqual(camera.thumbnail, test_thumbnail) + + # Manually set thumbnail, and then globally refresh and check + for camera_name in test_cameras: + camera = self.blink.cameras[camera_name] + camera.thumbnail = 'Testing' + self.assertEqual(camera.thumbnail, 'Testing') + + self.blink.refresh() + for camera_name in test_cameras: + camera = self.blink.cameras[camera_name] + self.assertEqual(camera.thumbnail, test_thumbnail) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_image_with_bad_data(self, mock_get, mock_post): + """Checks for handling of bad keys.""" + self.blink.setup_system() + for camera_name in self.blink.cameras: + camera = self.blink.cameras[camera_name] + camera.snap_picture() + camera.urls.home_url = "use_bad_response" + self.assertEqual(camera.image_refresh(), None) + + def test_camera_update(self): + """Checks that the update function is doing the right thing.""" + self.test_urls = blinkpy.BlinkURLHandler('test') + test_config = mresp.FIRST_CAMERA + test_camera = blinkpy.blinkpy.BlinkCamera(test_config, self.test_urls) + test_update = mresp.SECOND_CAMERA + test_camera.update(test_update) + test_image_url = self.test_urls.base_url + test_update['thumbnail'] + test_thumbnail = test_image_url + '.jpg' + test_clip = test_image_url + '.mp4' + self.assertEqual(test_camera.name, test_update['name']) + self.assertEqual(test_camera.armed, test_update['armed']) + self.assertEqual(test_camera.thumbnail, test_thumbnail) + self.assertEqual(test_camera.clip, test_clip) + self.assertEqual(test_camera.temperature, test_update['temp']) + self.assertEqual(test_camera.battery, test_update['battery']) + self.assertEqual(test_camera.notifications, + test_update['notifications']) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_camera_thumbs(self, mock_get, mock_post): + """Checks to see if we can retrieve camera thumbs.""" + self.test_urls = blinkpy.BlinkURLHandler(self.region_id) + test_cameras = mresp.get_test_cameras(self.test_urls.base_url) + self.blink.setup_system() + for name in self.blink.cameras: + thumb = self.blink.camera_thumbs[name] + self.assertEqual(test_cameras[name]['thumbnail'], thumb) + + @mock.patch('blinkpy.blinkpy.copyfileobj', + side_effect=mresp.mocked_copyfileobj) + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_image_to_file(self, mock_get, mock_post, mock_copyfileobj): + """Checks that we can write an image to file.""" + self.blink.setup_system() + cameras = self.blink.cameras + filename = '/tmp/test.jpg' + test_files = list() + for camera_name in cameras: + camera = cameras[camera_name] + test_files.append(mresp.MOCK_BYTES) + mock_fh = mock.mock_open() + with mock.patch('builtins.open', mock_fh, create=True): + camera.image_to_file(camera_name + filename) + mock_fh.assert_called_once_with(camera_name + filename, 'wb') + self.assertEqual(test_files, mresp.FAKE_FILES) diff --git a/tests/test_blink_system.py b/tests/test_blink_system.py index 37b2be3..a67f362 100644 --- a/tests/test_blink_system.py +++ b/tests/test_blink_system.py @@ -1,163 +1,173 @@ -""" -Tests the system initialization and attributes of -the main Blink system. Tests if we properly catch -any communication related errors at startup. -""" - -import unittest -from unittest import mock -import blinkpy -import tests.mock_responses as mresp -import helpers.constants as const - -USERNAME = 'foobar' -PASSWORD = 'deadbeef' - - -class TestBlinkSetup(unittest.TestCase): - """Test the Blink class in blinkpy.""" - - def setUp(self): - """Set up Blink module.""" - self.blink_no_cred = blinkpy.Blink() - self.blink = blinkpy.Blink(username=USERNAME, - password=PASSWORD) - - def tearDown(self): - """Clean up after test.""" - self.blink = None - self.blink_no_cred = None - - def test_initialization(self): - """Verify we can initialize blink.""" - # pylint: disable=protected-access - self.assertEqual(self.blink._username, USERNAME) - # pylint: disable=protected-access - self.assertEqual(self.blink._password, PASSWORD) - - def test_no_credentials(self): - """Check that we throw an exception when no username/password.""" - with self.assertRaises(blinkpy.BlinkAuthenticationException): - self.blink_no_cred.get_auth_token() - with self.assertRaises(blinkpy.BlinkAuthenticationException): - self.blink_no_cred.setup_system() - # pylint: disable=protected-access - self.blink_no_cred._username = USERNAME - with self.assertRaises(blinkpy.BlinkAuthenticationException): - self.blink_no_cred.get_auth_token() - - def test_no_auth_header(self): - """Check that we throw an excpetion when no auth header given.""" - # pylint: disable=unused-variable - (region_id, region), = mresp.LOGIN_RESPONSE['region'].items() - self.blink.urls = blinkpy.BlinkURLHandler(region_id) - with self.assertRaises(blinkpy.BlinkException): - self.blink.get_ids() - with self.assertRaises(blinkpy.BlinkException): - self.blink.get_summary() - - @mock.patch('blinkpy.blinkpy.getpass.getpass') - def test_manual_login(self, getpwd): - """Check that we can manually use the login() function.""" - getpwd.return_value = PASSWORD - with mock.patch('builtins.input', return_value=USERNAME): - self.blink_no_cred.login() - # pylint: disable=protected-access - self.assertEqual(self.blink_no_cred._username, USERNAME) - # pylint: disable=protected-access - self.assertEqual(self.blink_no_cred._password, PASSWORD) - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_bad_request(self, mock_get, mock_post): - """Check that we raise an Exception with a bad request.""" - with self.assertRaises(blinkpy.BlinkException): - # pylint: disable=protected-access - blinkpy.blinkpy._request(None, reqtype='bad') - - with self.assertRaises(blinkpy.BlinkAuthenticationException): - # pylint: disable=protected-access - blinkpy.blinkpy._request(None, reqtype='post') - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_full_setup(self, mock_get, mock_post): - """Check that we can set Blink up properly.""" - self.blink.setup_system() - - # Get all test values - authtoken = mresp.LOGIN_RESPONSE['authtoken']['authtoken'] - (region_id, region), = mresp.LOGIN_RESPONSE['region'].items() - host = region_id + '.' + const.BLINK_URL - network_id = mresp.NETWORKS_RESPONSE['networks'][0]['id'] - account_id = mresp.NETWORKS_RESPONSE['networks'][0]['account_id'] - test_urls = blinkpy.BlinkURLHandler(region_id) - test_cameras = mresp.get_test_cameras(test_urls.base_url) - test_camera_id = mresp.get_test_id_table() - - # Check that all links have been set properly - self.assertEqual(self.blink.region_id, region_id) - self.assertEqual(self.blink.urls.base_url, test_urls.base_url) - self.assertEqual(self.blink.urls.home_url, test_urls.home_url) - self.assertEqual(self.blink.urls.event_url, test_urls.event_url) - self.assertEqual(self.blink.urls.network_url, test_urls.network_url) - self.assertEqual(self.blink.urls.networks_url, test_urls.networks_url) - - # Check that all properties have been set after startup - # pylint: disable=protected-access - self.assertEqual(self.blink._token, authtoken) - # pylint: disable=protected-access - self.assertEqual(self.blink._host, host) - self.assertEqual(self.blink.network_id, str(network_id)) - self.assertEqual(self.blink.account_id, str(account_id)) - self.assertEqual(self.blink.region, region) - - # Verify we have initialized all the cameras - self.assertEqual(self.blink.id_table, test_camera_id) - for camera in test_cameras: - self.assertTrue(camera in self.blink.cameras) - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_arm_disarm_system(self, mock_get, mock_post): - """Check that we can arm/disarm the system""" - self.blink.setup_system() - self.blink.arm = False - self.assertIs(self.blink.arm, False) - self.blink.arm = True - self.assertIs(self.blink.arm, True) - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_check_online_status(self, mock_get, mock_post): - """Check that we can get our online status.""" - self.blink.setup_system() - expected_status = const.ONLINE[mresp.RESPONSE['syncmodule']['status']] - self.assertIs(self.blink.online, expected_status) - - @mock.patch('blinkpy.blinkpy.requests.post', - side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', - side_effect=mresp.mocked_requests_get) - def test_setup_backup_subdomain(self, mock_get, mock_post): - """Check that we can use the 'rest.piri' subdomain.""" - test_urls = blinkpy.BlinkURLHandler('rest.piri') - with mock.patch('helpers.constants.LOGIN_URL', - return_value=const.LOGIN_URL + 'NO'): - self.blink.setup_system() - self.assertEqual(self.blink.region_id, 'rest.piri') - # pylint: disable=protected-access - self.assertEqual(self.blink._host, 'rest.piri.' + const.BLINK_URL) - self.assertEqual(self.blink.urls.base_url, test_urls.base_url) - self.assertEqual(self.blink.urls.home_url, test_urls.home_url) - self.assertEqual(self.blink.urls.event_url, test_urls.event_url) - self.assertEqual(self.blink.urls.network_url, test_urls.network_url) - self.assertEqual(self.blink.urls.networks_url, test_urls.networks_url) +""" +Tests the system initialization and attributes of +the main Blink system. Tests if we properly catch +any communication related errors at startup. +""" + +import unittest +from unittest import mock +import blinkpy +import tests.mock_responses as mresp +import helpers.constants as const + +USERNAME = 'foobar' +PASSWORD = 'deadbeef' + + +class TestBlinkSetup(unittest.TestCase): + """Test the Blink class in blinkpy.""" + + def setUp(self): + """Set up Blink module.""" + self.blink_no_cred = blinkpy.Blink() + self.blink = blinkpy.Blink(username=USERNAME, + password=PASSWORD) + + def tearDown(self): + """Clean up after test.""" + self.blink = None + self.blink_no_cred = None + + def test_initialization(self): + """Verify we can initialize blink.""" + # pylint: disable=protected-access + self.assertEqual(self.blink._username, USERNAME) + # pylint: disable=protected-access + self.assertEqual(self.blink._password, PASSWORD) + + def test_no_credentials(self): + """Check that we throw an exception when no username/password.""" + with self.assertRaises(blinkpy.BlinkAuthenticationException): + self.blink_no_cred.get_auth_token() + with self.assertRaises(blinkpy.BlinkAuthenticationException): + self.blink_no_cred.setup_system() + # pylint: disable=protected-access + self.blink_no_cred._username = USERNAME + with self.assertRaises(blinkpy.BlinkAuthenticationException): + self.blink_no_cred.get_auth_token() + + def test_no_auth_header(self): + """Check that we throw an excpetion when no auth header given.""" + # pylint: disable=unused-variable + (region_id, region), = mresp.LOGIN_RESPONSE['region'].items() + self.blink.urls = blinkpy.BlinkURLHandler(region_id) + with self.assertRaises(blinkpy.BlinkException): + self.blink.get_ids() + with self.assertRaises(blinkpy.BlinkException): + self.blink.get_summary() + + @mock.patch('blinkpy.blinkpy.getpass.getpass') + def test_manual_login(self, getpwd): + """Check that we can manually use the login() function.""" + getpwd.return_value = PASSWORD + with mock.patch('builtins.input', return_value=USERNAME): + self.blink_no_cred.login() + # pylint: disable=protected-access + self.assertEqual(self.blink_no_cred._username, USERNAME) + # pylint: disable=protected-access + self.assertEqual(self.blink_no_cred._password, PASSWORD) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_bad_request(self, mock_get, mock_post): + """Check that we raise an Exception with a bad request.""" + with self.assertRaises(blinkpy.BlinkException): + # pylint: disable=protected-access + blinkpy.blinkpy._request(None, reqtype='bad') + + with self.assertRaises(blinkpy.BlinkAuthenticationException): + # pylint: disable=protected-access + blinkpy.blinkpy._request(None, reqtype='post') + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_system_bad_refresh(self, mock_get, mock_post): + """Checks for handling of bad keys.""" + self.blink.setup_system() + self.blink.urls.home_url = 'use_bad_response' + self.assertEqual(self.blink.refresh(), None) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_full_setup(self, mock_get, mock_post): + """Check that we can set Blink up properly.""" + self.blink.setup_system() + + # Get all test values + authtoken = mresp.LOGIN_RESPONSE['authtoken']['authtoken'] + (region_id, region), = mresp.LOGIN_RESPONSE['region'].items() + host = region_id + '.' + const.BLINK_URL + network_id = mresp.NETWORKS_RESPONSE['networks'][0]['id'] + account_id = mresp.NETWORKS_RESPONSE['networks'][0]['account_id'] + test_urls = blinkpy.BlinkURLHandler(region_id) + test_cameras = mresp.get_test_cameras(test_urls.base_url) + test_camera_id = mresp.get_test_id_table() + + # Check that all links have been set properly + self.assertEqual(self.blink.region_id, region_id) + self.assertEqual(self.blink.urls.base_url, test_urls.base_url) + self.assertEqual(self.blink.urls.home_url, test_urls.home_url) + self.assertEqual(self.blink.urls.event_url, test_urls.event_url) + self.assertEqual(self.blink.urls.network_url, test_urls.network_url) + self.assertEqual(self.blink.urls.networks_url, test_urls.networks_url) + + # Check that all properties have been set after startup + # pylint: disable=protected-access + self.assertEqual(self.blink._token, authtoken) + # pylint: disable=protected-access + self.assertEqual(self.blink._host, host) + self.assertEqual(self.blink.network_id, str(network_id)) + self.assertEqual(self.blink.account_id, str(account_id)) + self.assertEqual(self.blink.region, region) + + # Verify we have initialized all the cameras + self.assertEqual(self.blink.id_table, test_camera_id) + for camera in test_cameras: + self.assertTrue(camera in self.blink.cameras) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_arm_disarm_system(self, mock_get, mock_post): + """Check that we can arm/disarm the system""" + self.blink.setup_system() + self.blink.arm = False + self.assertIs(self.blink.arm, False) + self.blink.arm = True + self.assertIs(self.blink.arm, True) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_check_online_status(self, mock_get, mock_post): + """Check that we can get our online status.""" + self.blink.setup_system() + expected_status = const.ONLINE[mresp.RESPONSE['syncmodule']['status']] + self.assertIs(self.blink.online, expected_status) + + @mock.patch('blinkpy.blinkpy.requests.post', + side_effect=mresp.mocked_requests_post) + @mock.patch('blinkpy.blinkpy.requests.get', + side_effect=mresp.mocked_requests_get) + def test_setup_backup_subdomain(self, mock_get, mock_post): + """Check that we can use the 'rest.piri' subdomain.""" + test_urls = blinkpy.BlinkURLHandler('rest.piri') + with mock.patch('helpers.constants.LOGIN_URL', + return_value=const.LOGIN_URL + 'NO'): + self.blink.setup_system() + self.assertEqual(self.blink.region_id, 'rest.piri') + # pylint: disable=protected-access + self.assertEqual(self.blink._host, 'rest.piri.' + const.BLINK_URL) + self.assertEqual(self.blink.urls.base_url, test_urls.base_url) + self.assertEqual(self.blink.urls.home_url, test_urls.home_url) + self.assertEqual(self.blink.urls.event_url, test_urls.event_url) + self.assertEqual(self.blink.urls.network_url, test_urls.network_url) + self.assertEqual(self.blink.urls.networks_url, test_urls.networks_url) diff --git a/tox.ini b/tox.ini index 9a9d595..9fdf6f4 100644 --- a/tox.ini +++ b/tox.ini @@ -7,7 +7,7 @@ setenv = LANG=en_US.UTF-8 PYTHONPATH = {toxinidir} commands = - py.test -v --timeout=30 --duration=10 --cov=blinkpy --cov-report term-missing {posargs} + py.test --timeout=30 --duration=10 --cov=blinkpy --cov-report term-missing {posargs} deps = -r{toxinidir}/requirements.txt -r{toxinidir}/requirements_test.txt