More robust style testing... next up is more robust testing

This commit is contained in:
Kevin Fronczak
2017-03-08 23:03:09 -05:00
parent 69b7be744a
commit daa27e4989
4 changed files with 252 additions and 132 deletions
+193 -127
View File
@@ -4,15 +4,17 @@
'''
blinkpy by Kevin Fronczak - A Blink camera Python library
https://github.com/fronzbot/blinkpy
Original protocol hacking by MattTW : https://github.com/MattTW/BlinkMonitorProtocol
Original protocol hacking by MattTW :
https://github.com/MattTW/BlinkMonitorProtocol
Published under the MIT license - See LICENSE file for more details.
"Blink Wire-Free HS Home Monitoring & Alert Systems" is a trademark owned by Immedia Inc., see www.blinkforhome.com for more information.
"Blink Wire-Free HS Home Monitoring & Alert Systems" is a trademark
owned by Immedia Inc., see www.blinkforhome.com for more information.
I am in no way affiliated with Blink, nor Immedia Inc.
'''
import requests
import getpass
import json
import getpass
import requests
import errors as ERROR
from constants import (BLINK_URL, LOGIN_URL,
BASE_URL, DEFAULT_URL,
@@ -21,184 +23,223 @@ from constants import (BLINK_URL, LOGIN_URL,
ONLINE)
def _request(url, data=None, headers=None, type='get', stream=False, json=True):
def _request(url, data=None, headers=None, reqtype='get',
stream=False, json_resp=True):
"""Wrapper function for request"""
if type is 'post':
response = requests.post(url, headers=headers, data=data).json()
elif type is 'get' and json:
response = requests.get(url, headers=headers, stream=stream).json()
elif type is 'get' and not json:
response = requests.get(url, headers=headers, stream=stream)
if reqtype is 'post':
response = requests.post(url, headers=headers,
data=data).json()
elif reqtype is 'get' and json_resp:
response = requests.get(url, headers=headers,
stream=stream).json()
elif reqtype is 'get' and not json_resp:
response = requests.get(url, headers=headers,
stream=stream)
else:
raise BlinkException(ERROR.REQUEST)
if json and 'message' in response.keys():
raise BlinkAuthenticationException((response['code'], response['message']))
raise BlinkAuthenticationException(
(response['code'], response['message']))
return response
# pylint: disable=super-init-not-called
class BlinkException(Exception):
"""
Class to throw general blink exception.
"""
def __init__(self, errcode):
self.id = errcode[0]
self.errid = errcode[0]
self.message = errcode[1]
class BlinkAuthenticationException(BlinkException):
"""
Class to throw authentication exception.
"""
pass
class BlinkCamera(object):
"""Class to initialize individual camera"""
"""Class to initialize individual camera."""
def __init__(self, config):
self._ID = str(config['device_id'])
self._NAME = config['name']
self._STATUS = config['armed']
self._THUMB = BASE_URL + config['thumbnail'] + '.jpg'
self._CLIP = 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']
self._id = str(config['device_id'])
self._name = config['name']
self._status = config['armed']
self._thumb = BASE_URL + config['thumbnail'] + '.jpg'
self._clip = 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 self._ID
"""Returns camera id"""
return self._id
@property
def name(self):
return self._NAME
"""Returns camera name"""
return self._name
@name.setter
def name(self, value):
self._NAME = value
"""Sets camera name"""
self._name = value
@property
def region_id(self):
return self._REGION_ID
"""Returns region id"""
return self._region_id
@property
def armed(self):
return self._STATUS
"""Returns camera arm status"""
return self._status
@property
def clip(self):
return self._CLIP
"""Returns current clip"""
return self._clip
@clip.setter
def clip(self, value):
self._CLIP = value
"""Sets current clip"""
self._clip = value
@property
def thumbnail(self):
# RUN THUMB ACQ HERE
return self._THUMB
"""Returns current thumbnail"""
return self._thumb
@thumbnail.setter
def thumbnail(self, value):
self._THUMB = value
"""Sets current thumbnail"""
self._thumb = value
@property
def temperature(self):
return self._TEMPERATURE
"""Returns camera temperature"""
return self._temperature
@temperature.setter
def temperature(self, value):
self._TEMPERATURE = value
"""Sets camera temperature"""
self._temperature = value
@property
def battery(self):
return self._BATTERY
"""Returns battery level"""
return self._battery
@battery.setter
def battery(self, value):
self._BATTERY = value
"""Sets battery level"""
self._battery = value
@property
def notifications(self):
return self._NOTIFICATIONS
"""Returns number of notifications"""
return self._notifications
@notifications.setter
def notifications(self, value):
self._NOTIFICATIONS = value
"""Sets number of notifications"""
self._notifications = value
@property
def image_link(self):
return self._IMAGE_LINK
"""Returns image link"""
return self._image_link
@image_link.setter
def image_link(self, value):
self._IMAGE_LINK = value
"""Sets image link"""
self._image_link = value
@property
def arm_link(self):
return self._ARM_LINK
"""Returns link to arm camera"""
return self._arm_link
@arm_link.setter
def arm_link(self, value):
self._ARM_LINK = value
"""Sets link to arm camera"""
self._arm_link = value
@property
def header(self):
return self._HEADER
"""Returns request header"""
return self._header
@header.setter
def header(self, value):
self._HEADER = value
"""Sets request header"""
self._header = value
@property
def motion(self):
return self._MOTION
"""Returns last motion event detail"""
return self._motion
@motion.setter
def motion(self, value):
"""Sets link to last motion and timestamp"""
self._MOTION = value
self._motion = value
def snap_picture(self):
"""Takes a picture with camera to create a new thumbnail"""
_request(self._IMAGE_LINK, headers=self._HEADER, type='post')
_request(self._image_link, headers=self._header, reqtype='post')
def set_motion_detect(self, enable):
"""Sets motion detection"""
url = self._ARM_LINK
url = self._arm_link
if enable:
_request(url + 'enable', headers=self._HEADER, type='post')
_request(url + 'enable', headers=self._header, reqtype='post')
else:
_request(url + 'disable', headers=self._HEADER, type='post')
_request(url + 'disable', headers=self._header, reqtype='post')
def update(self, values):
self._NAME = values['name']
self._STATUS = values['armed']
self._THUMB = BASE_URL + values['thumbnail'] + '.jpg'
self._CLIP = BASE_URL + values['thumbnail'] + '.mp4'
self._TEMPERATURE = values['temp']
self._BATTERY = values['battery']
self._NOTIFICATIONS = values['notifications']
"""Updates camera information"""
self._name = values['name']
self._status = values['armed']
self._thumb = BASE_URL + values['thumbnail'] + '.jpg'
self._clip = BASE_URL + values['thumbnail'] + '.mp4'
self._temperature = values['temp']
self._battery = values['battery']
self._notifications = values['notifications']
def image_refresh(self):
"""Refreshs current thumbnail"""
url = HOME_URL
response = _request(url, headers=self._HEADER, type='get')['devices']
response = _request(url, headers=self._header,
reqtype='get')['devices']
for element in response:
try:
if str(element['device_id']) == self._ID:
self._THUMB = BASE_URL + element['thumbnail'] + '.jpg'
return self._THUMB
if str(element['device_id']) == self._id:
self._thumb = BASE_URL + element['thumbnail'] + '.jpg'
return self._thumb
except KeyError:
pass
return None
def image_to_file(self, path):
"""Writes image to file"""
thumb = self.image_refresh()
response = _request(thumb, headers=self._HEADER, stream=True, json=False)
response = _request(thumb, headers=self._header,
stream=True, json_resp=False)
if response.status_code == 200:
with open(path, 'wb') as f:
with open(path, 'wb') as imgfile:
for chunk in response:
f.write(chunk)
imgfile.write(chunk)
class Blink(object):
@@ -207,64 +248,76 @@ class Blink(object):
"""Constructor for class"""
self._username = username
self._password = password
self._TOKEN = None
self._AUTH_HEADER = None
self._NETWORKID = None
self._ACCOUNTID = None
self._REGION = None
self._REGION_ID = None
self._HOST = None
self._EVENTS = []
self._CAMERAS = {}
self._IDLOOKUP = {}
self._token = None
self._auth_header = None
self._network_id = None
self._account_id = None
self._region = None
self._region_id = None
self._host = None
self._events = []
self._cameras = {}
self._idlookup = {}
@property
def cameras(self):
return self._CAMERAS
"""Returns camera/id pairs"""
return self._cameras
@property
def camera_thumbs(self):
"""Returns 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
@property
def id_table(self):
return self._IDLOOKUP
"""Returns id/camera pairs"""
return self._idlookup
@property
def network_id(self):
return self._NETWORKID
"""Returns network id"""
return self._network_id
@property
def account_id(self):
return self._ACCOUNTID
"""Returns account id"""
return self._account_id
@property
def region(self):
return self._REGION
"""Returns current region"""
return self._region
@property
def region_id(self):
return self._REGION_ID
"""Returns region id"""
return self._region_id
@property
def events(self):
"""Gets all events on server"""
url = EVENT_URL + self._NETWORKID
headers = self._AUTH_HEADER
self._EVENTS = _request(url, headers=headers, type='get')['event']
return self._EVENTS
url = EVENT_URL + self._network_id
headers = self._auth_header
self._events = _request(url, headers=headers,
reqtype='get')['event']
return self._events
@property
def online(self):
"""Returns True or False depending on if sync module is online/offline"""
url = NETWORK_URL + self._NETWORKID + '/syncmodules'
headers = self._AUTH_HEADER
return ONLINE[_request(url, headers=headers, type='get')['syncmodule']['status']]
"""
Returns True or False depending on if
sync module is online/offline
"""
url = NETWORK_URL + self._network_id + '/syncmodules'
headers = self._auth_header
return ONLINE[_request(url, headers=headers,
reqtype='get')['syncmodule']['status']]
def last_motion(self):
"""Finds last motion of each camera"""
@@ -272,11 +325,13 @@ 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._idlookup[camera_id]
camera = self._cameras[camera_name]
if element['type'] == 'motion':
url = BASE_URL + element['video_url']
camera.motion = {'video': url, 'image': url[:-3] + 'jpg', 'time': element['created_at']}
camera.motion = {'video': url,
'image': url[:-3] + 'jpg',
'time': element['created_at']}
except KeyError:
pass
@@ -287,19 +342,23 @@ class Blink(object):
@arm.setter
def arm(self, value):
"""Arms or disarms system. Arms/disarms all if camera not named"""
"""
Arms or disarms system.
Arms/disarms all if camera not named.
"""
if value:
value_to_append = 'arm'
else:
value_to_append = 'disarm'
url = NETWORK_URL + self._NETWORKID + '/' + value_to_append
_request(url, headers=self._AUTH_HEADER, type='post')
url = NETWORK_URL + self._network_id + '/' + value_to_append
_request(url, headers=self._auth_header, reqtype='post')
def refresh(self):
"""Gets all blink cameras and pulls their most recent status"""
response = self.get_summary()['devices']
for name, camera in self._CAMERAS.items():
for name in self._cameras:
camera = self._cameras[name]
for element in response:
try:
if str(element['device_id']) == camera.id:
@@ -310,12 +369,12 @@ class Blink(object):
def get_summary(self):
"""Gets a full summary of device information"""
url = BASE_URL + '/homescreen'
headers = self._AUTH_HEADER
headers = self._auth_header
if self._AUTH_HEADER is None:
if self._auth_header is None:
raise BlinkException(ERROR.AUTH_TOKEN)
return _request(url, headers=headers, type='get')
return _request(url, headers=headers, reqtype='get')
def get_cameras(self):
"""Finds and creates cameras"""
@@ -324,22 +383,30 @@ class Blink(object):
if 'device_type' in element.keys():
if element['device_type'] == 'camera':
# Add region to config
element['region_id'] = self._REGION_ID
element['region_id'] = self._region_id
device = BlinkCamera(element)
self._CAMERAS[device.name] = device
self._IDLOOKUP[device.id] = device.name
self._cameras[device.name] = device
self._idlookup[device.id] = device.name
def set_links(self):
"""Sets access links and required headers for each camera in system"""
for name, camera in self._CAMERAS.items():
image_url = NETWORK_URL + self._NETWORKID + '/camera/' + camera.id + '/thumbnail'
arm_url = NETWORK_URL + self._NETWORKID + '/camera/' + camera.id + '/'
"""
Sets access links and required headers
for each camera in system
"""
for name in self._cameras:
camera = self._cameras[name]
network_id_url = 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
camera.arm_link = arm_url
camera.header = self._AUTH_HEADER
camera.header = self._auth_header
def setup_system(self):
"""Method logs in and sets auth token and network ids for future requests"""
"""
Method logs in and sets auth token and
network ids for future requests.
"""
if self._username is None or self._password is None:
raise BlinkAuthenticationException(ERROR.AUTHENTICATE)
@@ -361,29 +428,28 @@ class Blink(object):
raise BlinkAuthenticationException(ERROR.PASSWORD)
headers = {'Host': DEFAULT_URL,
'Content-Type': 'application/json'
}
'Content-Type': 'application/json'}
data = json.dumps({
"email": self._username,
"password": self._password,
"client_specifier": "iPhone 9.2 | 2.2 | 222"
})
response = _request(LOGIN_URL, headers=headers, data=data, type='post')
self._TOKEN = response['authtoken']['authtoken']
(self._REGION_ID, self._REGION), = response['region'].items()
self._HOST = self._REGION_ID + '.' + BLINK_URL
self._AUTH_HEADER = {'Host': self._HOST,
'TOKEN_AUTH': self._TOKEN
}
response = _request(LOGIN_URL, headers=headers,
data=data, reqtype='post')
self._token = response['authtoken']['authtoken']
(self._region_id, self._region), = response['region'].items()
self._host = self._region_id + '.' + BLINK_URL
self._auth_header = {'Host': self._host,
'TOKEN_AUTH': self._token}
def get_ids(self):
"""Sets the network ID and Account ID"""
url = NETWORKS_URL
headers = self._AUTH_HEADER
headers = self._auth_header
if self._AUTH_HEADER is None:
if self._auth_header is None:
raise BlinkException(ERROR.AUTH_TOKEN)
response = _request(url, headers=headers, type='get')
self._NETWORKID = str(response['networks'][0]['id'])
self._ACCOUNTID = str(response['networks'][0]['account_id'])
response = _request(url, headers=headers, reqtype='get')
self._network_id = str(response['networks'][0]['id'])
self._account_id = str(response['networks'][0]['account_id'])
+38
View File
@@ -0,0 +1,38 @@
[MASTER]
reports=no
# Reasons disabled:
# locally-disabled - it spams too much
# duplicate-code - unavoidable
# cyclic-import - doesn't test if both import on load
# abstract-class-little-used - prevents from setting right foundation
# abstract-class-not-used - is flaky, should not show up but does
# unused-argument - generic callbacks and setup methods create a lot of warnings
# global-statement - used for the on-demand requirement installation
# redefined-variable-type - this is Python, we're duck typing!
# too-many-* - are not enforced for the sake of readability
# too-few-* - same as too-many-*
# abstract-method - with intro of async there are always methods missing
disable=
locally-disabled,
duplicate-code,
cyclic-import,
abstract-class-little-used,
abstract-class-not-used,
unused-argument,
global-statement,
redefined-variable-type,
too-many-arguments,
too-many-branches,
too-many-instance-attributes,
too-many-locals,
too-many-public-methods,
too-many-return-statements,
too-many-statements,
too-many-lines,
too-few-public-methods,
abstract-method
[EXCEPTIONS]
overgeneral-exceptions=Exception
+8
View File
@@ -0,0 +1,8 @@
import requests
import unittest
from constants import BASE_URL
class TestRemoteServerResponse(unittest.TestCase):
def test_request_response(self):
response = requests.get(BASE_URL)
self.assertEqual(response.ok, True)
+13 -5
View File
@@ -1,5 +1,6 @@
[tox]
envlist = py34, pep8
envlist = py34, py35, py36, lint
skip_missing_interpreters = True
[testenv]
setenv =
@@ -10,14 +11,21 @@ commands =
deps =
-r{toxinidir}/requirements.txt
[testenv:pep8]
[testenv:lint]
deps =
-r{toxinidir}/requirements.txt
flake8
pydocstyle
pylint
basepython = python3
ignore_errors = True
commands =
;flake8 blinkpy.py tests
flake8 blinkpy.py
;pylint blinkpy.py tests
pylint blinkpy.py
;pydocstyle blinkpy.py tests
[flake8]
ignore = E501
exclude = .venv,.git,.tox,dist,doc,*lib/python,*egg,build
;[flake8]
;ignore = E501
;exclude = .venv,.git,.tox,dist,doc,*lib/python,*egg,build