caelebfi/blinkpy
Archived
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #21 from fronzbot/cleanup

Browse Source 
Cleanup and Coverage Increase
This commit is contained in:
Kevin Fronczak
2017-03-20 15:57:43 -04:00
committed by GitHub
parent c7fe0a5b5c 9fa2e967dc
commit ebf4cb53b5
9 changed files with 814 additions and 861 deletions
Download Patch File Download Diff File Expand all files Collapse all files
API.md
+8 -8
View File
@@ -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
CHANGES.rst
+53
View File
@@ -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
blinkpy.py
+62 -200
View File
@@ -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'])
helpers/constants.py
+56 -55
View File
@@ -1,55 +1,56 @@
'''
constants.py
Generates constants for use in blinkpy
'''
import os
MAJOR_VERSION = 0
MINOR_VERSION = 5
PATCH_VERSION = '3.dev0'
__version__ = '{}.{}.{}'.format(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION)
REQUIRED_PYTHON_VER = (3, 4, 2)
PROJECT_NAME = 'blinkpy'
PROJECT_PACKAGE_NAME = 'blinkpy'
PROJECT_LICENSE = 'MIT'
PROJECT_AUTHOR = 'Kevin Fronczak'
PROJECT_COPYRIGHT = ' 2017, {}'.format(PROJECT_AUTHOR)
PROJECT_URL = 'https://github.com/fronzbot/blinkpy'
PROJECT_EMAIL = 'kfronczak@gmail.com'
PROJECT_DESCRIPTION = ('A Blink camera Python library '
'running on Python 3.')
PROJECT_LONG_DESCRIPTION = ('blinkpy is an open-source '
'unofficial API for the Blink Camera '
'system with the intention for easy '
'integration into various home '
'automation platforms.')
if os.path.exists('README.rst'):
PROJECT_LONG_DESCRIPTION = open('README.rst').read()
PROJECT_CLASSIFIERS = [
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
'Programming Language :: Python :: 3.4',
'Topic :: Home Automation'
]
PROJECT_GITHUB_USERNAME = 'fronzbot'
PROJECT_GITHUB_REPOSITORY = 'blinkpy'
PYPI_URL = 'https://pypi.python.org/pypi/{}'.format(PROJECT_PACKAGE_NAME)
'''
URLS
'''
BLINK_URL = 'immedia-semi.com'
LOGIN_URL = 'https://prod.' + BLINK_URL + '/login'
LOGIN_BACKUP_URL = 'https://rest.piri/' + BLINK_URL + '/login'
BASE_URL = 'https://prod.' + BLINK_URL
DEFAULT_URL = 'prod.' + BLINK_URL
'''
Dictionaries
'''
ONLINE = {'online': True, 'offline': False}
'''
constants.py
Generates constants for use in blinkpy
'''
import os
MAJOR_VERSION = 0
MINOR_VERSION = 6
PATCH_VERSION = '0.dev0'
__version__ = '{}.{}.{}'.format(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION)
REQUIRED_PYTHON_VER = (3, 4, 2)
PROJECT_NAME = 'blinkpy'
PROJECT_PACKAGE_NAME = 'blinkpy'
PROJECT_LICENSE = 'MIT'
PROJECT_AUTHOR = 'Kevin Fronczak'
PROJECT_COPYRIGHT = ' 2017, {}'.format(PROJECT_AUTHOR)
PROJECT_URL = 'https://github.com/fronzbot/blinkpy'
PROJECT_EMAIL = 'kfronczak@gmail.com'
PROJECT_DESCRIPTION = ('A Blink camera Python library '
'running on Python 3.')
PROJECT_LONG_DESCRIPTION = ('blinkpy is an open-source '
'unofficial API for the Blink Camera '
'system with the intention for easy '
'integration into various home '
'automation platforms.')
if os.path.exists('README.rst'):
PROJECT_LONG_DESCRIPTION = open('README.rst').read()
PROJECT_CLASSIFIERS = [
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
'Programming Language :: Python :: 3.4',
'Topic :: Home Automation'
]
PROJECT_GITHUB_USERNAME = 'fronzbot'
PROJECT_GITHUB_REPOSITORY = 'blinkpy'
PYPI_URL = 'https://pypi.python.org/pypi/{}'.format(PROJECT_PACKAGE_NAME)
'''
URLS
'''
BLINK_URL = 'immedia-semi.com'
LOGIN_URL = 'https://prod.' + BLINK_URL + '/login'
LOGIN_BACKUP_URL = 'https://rest.piri/' + BLINK_URL + '/login'
BASE_URL = 'https://prod.' + BLINK_URL
DEFAULT_URL = 'prod.' + BLINK_URL
'''
Dictionaries
'''
ONLINE = {'online': True, 'offline': False}
requirements_test.txt
+3 -1
View File
@@ -1,3 +1,5 @@
flake8==3.3
pylint==1.6.5
pydocstyle==1.1.1
pydocstyle==1.1.1
pytest>=2.9.2
pytest-sugar>=0.8.0
tests/mock_responses.py
+286 -274
View File
@@ -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
tests/test_blink_functions.py
+172 -159
View File
@@ -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)
tests/test_blink_system.py
+173 -163
View File
@@ -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)
tox.ini
+1 -1
View File
@@ -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