From a0a1e2837cbe0bd57423d2d2e97786cf821bdcb8 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Sun, 12 Mar 2017 20:04:42 -0400 Subject: [PATCH 1/4] Fixed coveralls badge --- README.rst | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.rst b/README.rst index 67d2859..3d78bb9 100644 --- a/README.rst +++ b/README.rst @@ -116,7 +116,8 @@ The ``BlinkURLHandler`` class expects to be initialized with the region id found .. |Build Status| image:: https://travis-ci.org/fronzbot/blinkpy.svg?branch=master :target: https://travis-ci.org/fronzbot/blinkpy .. |Coverage Status| image:: https://coveralls.io/repos/github/fronzbot/blinkpy/badge.svg?branch=master - :target: https://coveralls.io/r/fronzbot/blinkpy?branch=master + :target: https://coveralls.io/github/fronzbot/blinkpy?branch=master + From 479a41c900005a6fad3ffa24700b1beb776e4c43 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Sun, 12 Mar 2017 22:06:39 -0400 Subject: [PATCH 2/4] Had to update version due to messing up the module distribution --- MANIFEST.in | 1 - helpers/constants.py | 2 +- setup.py | 3 ++- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/MANIFEST.in b/MANIFEST.in index 3eb1ac3..00970ff 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -2,4 +2,3 @@ include README.rst include LICENSE.md include API.md include tests/*.py -include helpers/*.py diff --git a/helpers/constants.py b/helpers/constants.py index 6ef8d94..554bc61 100644 --- a/helpers/constants.py +++ b/helpers/constants.py @@ -6,7 +6,7 @@ import os MAJOR_VERSION = 0 MINOR_VERSION = 5 -PATCH_VERSION = 0 +PATCH_VERSION = 2 __version__ = '{}.{}.{}'.format(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION) REQUIRED_PYTHON_VER = (3, 4, 2) diff --git a/setup.py b/setup.py index 5f5cd7c..5f9882b 100644 --- a/setup.py +++ b/setup.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -from setuptools import setup +from setuptools import setup, find_packages from helpers.constants import (__version__, PROJECT_PACKAGE_NAME, PROJECT_LICENSE, PROJECT_URL, PROJECT_EMAIL, PROJECT_DESCRIPTION, @@ -17,6 +17,7 @@ setup( url = PROJECT_URL, platforms = 'any', py_modules = ['blinkpy'], + packages=find_packages(), install_requires = ['requests>=2,<3'], test_suite = 'tests', classifiers = PROJECT_CLASSIFIERS From 874f717bd1ee14f28ad86e1aa47df3149208531b Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Mon, 20 Mar 2017 13:22:12 -0400 Subject: [PATCH 3/4] Fixed formatting --- API.md | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/API.md b/API.md index 90549bc..6dbdd2c 100644 --- a/API.md +++ b/API.md @@ -10,7 +10,7 @@ The Client API is a straightforward REST API using JSON and HTTPS. ## This Document The purpose here is to describe what is going on behind the scenes, and what commands we know are available to communicate with Blink servers. This is NOT a description of the blinkpy module, but a description of the commands blinkpy relies on to effectively communicate. -##Login +## Login Client login to the Blink Servers. @@ -41,7 +41,7 @@ JSON response containing information including Network ID and Account ID. Network ID is needed to issue arm/disarm calls -##Sync Modules +## Sync Modules Obtain information about the Blink Sync Modules on the given network. @@ -55,7 +55,7 @@ JSON response containing information about the known state of the Sync module, m Probably not strictly needed but checking result can verify that the sync module is online and will respond to requests to arm/disarm, etc. -##Arm +## Arm Arm the given network (start recording/reporting motion events) @@ -68,7 +68,7 @@ JSON response containing information about the arm command request, including th **Notes:** When this call returns, it does not mean the arm request is complete, the client must gather the request ID from the response and poll for the status of the command. -##Disarm +## Disarm Disarm the given network (stop recording/reporting motion events) @@ -82,7 +82,7 @@ JSON response containing information about the disarm command request, including When this call returns, it does not mean the disarm request is complete, the client must gather the request ID from the response and poll for the status of the command. -##Command Status +## Command Status Get status info on the given command @@ -98,7 +98,7 @@ After an arm/disarm command, the client appears to poll this URL every second or **Known Commands:** lv_relay, arm, disarm, thumbnail, clip -##Home Screen +## Home Screen Return information displayed on the home screen of the mobile client @@ -207,7 +207,7 @@ Deletes all videos **Response** Unknown - not tested -##Cameras +## Cameras **Request** Gets a list of cameras @@ -252,7 +252,7 @@ JSON response containing camera information *Note*: enabling or disabling motion detection is independent of arming or disarming the system. No motion detection or video recording will take place unless the system is armed. -##Miscellaneous +## Miscellaneous **Request** Gets information about devices that have connected to the blink service From 9fa2e967dc939a84249f8fc4718528894dc1216d Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Mon, 20 Mar 2017 15:53:53 -0400 Subject: [PATCH 4/4] Removed redundant functions, increased test coverage --- CHANGES.rst | 53 ++++ blinkpy.py | 262 ++++------------ helpers/constants.py | 4 +- requirements_test.txt | 4 +- tests/mock_responses.py | 560 +++++++++++++++++----------------- tests/test_blink_functions.py | 331 ++++++++++---------- tests/test_blink_system.py | 336 ++++++++++---------- tox.ini | 2 +- 8 files changed, 752 insertions(+), 800 deletions(-) create mode 100644 CHANGES.rst 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