Removed redundant functions, increased test coverage
This commit is contained in:
Kevin Fronczak
+286
-274
@@ -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
|
||||
|
||||
+172
-159
@@ -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)
|
||||
|
||||
+173
-163
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user