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

Merge pull request #83 from fronzbot/use-sessions

Browse Source 
Use sessions for requests
This commit is contained in:
Kevin Fronczak
2018-09-27 15:19:11 -04:00
committed by GitHub
parent fe0aef69e4 674dae4440
commit 2e6fb97cc5
8 changed files with 131 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
blinkpy/blinkpy.py
+17 -9
View File
@@ -19,7 +19,8 @@ import logging
import blinkpy.helpers.errors as ERROR
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.helpers.util import (
http_req, BlinkURLHandler, BlinkException, BlinkAuthenticationException)
http_req, create_session, BlinkURLHandler,
BlinkException, BlinkAuthenticationException)
from blinkpy.helpers.constants import (
DEFAULT_URL, BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL)
@@ -50,6 +51,8 @@ class Blink():
self.region_id = None
self.last_refresh = None
self.refresh_rate = refresh_rate
self.session = None
self._login_url = LOGIN_URL
@property
def events(self):
@@ -100,7 +103,8 @@ class Blink():
"password": self._password,
"client_specifier": "iPhone 9.2 | 2.2 | 222"
})
response = http_req(self, url=LOGIN_URL, headers=headers,
self.session = create_session()
response = http_req(self, url=self._login_url, headers=headers,
data=data, json_resp=False, reqtype='post')
if response.status_code == 200:
response = response.json()
@@ -111,7 +115,8 @@ class Blink():
"when authenticating, "
"trying new url"), response.status_code
)
response = http_req(self, url=LOGIN_BACKUP_URL, headers=headers,
self._login_url = LOGIN_BACKUP_URL
response = http_req(self, url=self._login_url, headers=headers,
data=data, reqtype='post')
self.region_id = 'piri'
self.region = "UNKNOWN"
@@ -140,22 +145,23 @@ class Blink():
raise BlinkException(ERROR.AUTH_TOKEN)
return http_req(self, url=url, headers=headers, reqtype='get')
def events_request(self):
def events_request(self, skip_throttle=False):
"""Get events on server."""
url = "{}/{}".format(self.urls.event_url, self.network_id)
headers = self._auth_header
if self.check_if_ok_to_update() or self._last_events is None:
self._last_events = http_req(
self, url=url, headers=headers, reqtype='get')
if self.check_if_ok_to_update() or skip_throttle:
self._last_events = http_req(self, url=url,
headers=headers,
reqtype='get')
return self._last_events
def summary_request(self):
def summary_request(self, skip_throttle=False):
"""Get blink summary."""
url = self.urls.home_url
headers = self._auth_header
if headers is None:
raise BlinkException(ERROR.AUTH_TOKEN)
if self.check_if_ok_to_update() or self._last_summary is None:
if self.check_if_ok_to_update() or skip_throttle:
self._last_summary = http_req(
self, url=url, headers=headers, reqtype='get')
return self._last_summary
@@ -164,6 +170,8 @@ class Blink():
"""Perform a system refresh."""
if self.check_if_ok_to_update() or force_cache:
_LOGGER.debug("Attempting refresh of cameras.")
self._last_events = self.events_request(skip_throttle=True)
self._last_summary = self.summary_request(skip_throttle=True)
self.sync.refresh(force_cache=force_cache)
def check_if_ok_to_update(self):
blinkpy/helpers/util.py
+17 -10
View File
@@ -1,7 +1,7 @@
"""Useful functions for blinkpy."""
import logging
import requests
from requests import Request, Session
import blinkpy.helpers.errors as ERROR
from blinkpy.helpers.constants import BLINK_URL
@@ -9,6 +9,12 @@ from blinkpy.helpers.constants import BLINK_URL
_LOGGER = logging.getLogger(__name__)
def create_session():
"""Create a session for blink communication."""
sess = Session()
return sess
def attempt_reauthorization(blink):
"""Attempt to refresh auth token and links."""
_LOGGER.debug("Auth token expired, attempting reauthorization.")
@@ -17,18 +23,19 @@ def attempt_reauthorization(blink):
return headers
def http_req(blink, url='http://google.com', data=None, headers=None,
def http_req(blink, url='http://example.com', data=None, headers=None,
reqtype='get', stream=False, json_resp=True, is_retry=False):
"""Perform server requests and check if reauthorization neccessary."""
if reqtype == 'post':
response = requests.post(url, headers=headers,
data=data)
req = Request('POST', url, headers=headers, data=data)
elif reqtype == 'get':
response = requests.get(url, headers=headers,
stream=stream)
req = Request('GET', url, headers=headers)
else:
raise BlinkException(ERROR.REQUEST)
prepped = req.prepare()
response = blink.session.send(prepped, stream=stream)
if json_resp and 'code' in response.json():
if is_retry:
raise BlinkAuthenticationException(
@@ -38,19 +45,19 @@ def http_req(blink, url='http://google.com', data=None, headers=None,
return http_req(blink, url=url, data=data, headers=headers,
reqtype=reqtype, stream=stream,
json_resp=json_resp, is_retry=True)
# pylint: disable=no-else-return
if json_resp:
return response.json()
else:
return response
return response
# pylint: disable=super-init-not-called
class BlinkException(Exception):
"""Class to throw general blink exception."""
def __init__(self, errcode):
"""Initialize BlinkException."""
super().__init__()
self.errid = errcode[0]
self.message = errcode[1]
blinkpy/sync_module.py
+7 -5
View File
@@ -85,7 +85,7 @@ class BlinkSyncModule():
@property
def arm(self):
"""Return status of sync module: armed/disarmed."""
return self.summary['network']['armed']
return self._summary['network']['armed']
@arm.setter
def arm(self, value):
@@ -101,9 +101,9 @@ class BlinkSyncModule():
def refresh(self, force_cache=False):
"""Get all blink cameras and pulls their most recent status."""
self._summary = self._summary_request()
self._events = self._events_request()
response = self.summary['devices']
summary = self._summary_request()
events = self._events_request()
response = summary['devices']
self.get_videos()
for name in self.cameras:
camera = self.cameras[name]
@@ -116,6 +116,8 @@ class BlinkSyncModule():
camera.update(element, force_cache=force_cache)
except KeyError:
pass
self._summary = summary
self._events = events
def get_videos(self, start_page=0, end_page=1):
"""Retrieve last recorded videos per camera."""
@@ -158,7 +160,7 @@ class BlinkSyncModule():
def get_cameras(self):
"""Find and creates cameras."""
self._summary = self._summary_request()
response = self.summary['devices']
response = self._summary['devices']
for element in response:
if ('device_type' in element and
element['device_type'] == 'camera'):
tests/mock_responses.py
+45 -58
View File
@@ -20,70 +20,57 @@ LOGIN_RESPONSE = {
}
def mocked_requests_post(*args, **kwargs):
"""Mock post request."""
class MockPostResponse:
"""Class for mock post response."""
class MockResponse:
"""Class for mock request response."""
def __init__(self, json_data, status_code):
"""Initialize mock post response."""
self.json_data = json_data
self.status_code = status_code
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 post request."""
return self.json_data
def json(self):
"""Return json data from get_request."""
return self.json_data
url_arg = args[0]
response_to_return = {'message': 'Error', 'code': 404}
code_to_return = 404
if url_arg in (const.LOGIN_URL, const.LOGIN_BACKUP_URL):
response_to_return = LOGIN_RESPONSE
code_to_return = 200
elif url_arg is not None:
response_to_return = {'message': 'foobar', 'code': 200}
code_to_return = 200
return MockPostResponse(response_to_return, code_to_return)
@property
def raw(self):
"""Return raw data from get request."""
return self.raw_data
def mocked_requests_get(*args, **kwargs):
"""Mock get request."""
class MockGetResponse:
"""Class for mock get response."""
def mocked_session_send(*args, **kwargs):
"""Mock session."""
prepped = args[0]
url = prepped.url
header = prepped.headers
method = prepped.method
if method == 'GET':
expected_token = LOGIN_RESPONSE['authtoken']['authtoken']
if header['TOKEN_AUTH'] != expected_token:
response = {'message': 'Not Authorized', 'code': 400}
status = 400
elif url == 'use_bad_response':
response = {'foo': 'bar'}
status = 200
elif url == 'reauth':
response = {'message': 'REAUTH', 'code': 777}
status = 777
else:
response = {'test': 'foo'}
status = 200
elif method == 'POST':
if url in (const.LOGIN_URL, const.LOGIN_BACKUP_URL):
response = LOGIN_RESPONSE
status = 200
elif url == 'http://wrong.url/' or url is None:
response = {'message': 'Error', 'code': 404}
status = 404
else:
response = {'message': 'foo', 'code': 200}
status = 200
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
rx_header = kwargs.pop('headers')
expected_token = LOGIN_RESPONSE['authtoken']['authtoken']
if ('Content-Type' not in rx_header
and rx_header['TOKEN_AUTH'] != expected_token):
return MockGetResponse({'message': 'Not Authorized', 'code': 400}, 400)
url_arg = args[0]
# pylint: disable=R1711
if url_arg == 'use_bad_response':
return MockGetResponse({'foo': 'bar'}, 200)
elif url_arg == 'reauth':
return MockGetResponse({'message': 'REAUTH', 'code': 777}, 777)
return MockGetResponse({'test': 'foo'}, 200)
return MockResponse(response, status)
class MockURLHandler(BlinkURLHandler):
tests/test_blink_functions.py
+21 -14
View File
@@ -1,14 +1,13 @@
"""Tests camera and system functions."""
import time
import unittest
from unittest import mock
# import pytest
from requests import Request
from blinkpy import blinkpy
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.camera import BlinkCamera
from blinkpy.helpers.util import create_session
from blinkpy.helpers.constants import BLINK_URL
import tests.mock_responses as mresp
@@ -38,6 +37,8 @@ class MockSyncModule(BlinkSyncModule):
return self.return_value
@mock.patch('blinkpy.helpers.util.Session.send',
side_effect=mresp.mocked_session_send)
class TestBlinkFunctions(unittest.TestCase):
"""Test Blink and BlinkCamera functions in blinkpy."""
@@ -66,7 +67,7 @@ class TestBlinkFunctions(unittest.TestCase):
}
self.blink.sync = MockSyncModule(
self.blink, self.blink._auth_header)
self.blink.session = create_session()
self.camera = BlinkCamera(self.config, self.blink.sync)
def tearDown(self):
@@ -75,7 +76,7 @@ class TestBlinkFunctions(unittest.TestCase):
self.config = {}
self.camera = None
def test_image_refresh(self):
def test_image_refresh(self, mock_sess):
"""Test image refresh function."""
self.blink.sync.return_value = {'devices': [self.config]}
image = self.camera.image_refresh()
@@ -84,15 +85,15 @@ class TestBlinkFunctions(unittest.TestCase):
@mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request')
@mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request')
def test_refresh(self, vid_req, req):
def test_refresh(self, vid_req, req, mock_sess):
"""Test blinkpy refresh function."""
req.return_value = {'foo': 'bar'}
self.blink.sync.cameras = {'foobar': self.camera}
self.blink.sync.return_value = {'devices': [{'foo': 'bar'}]}
# pylint: disable=protected-access
self.blink._last_summary = {'devices': [self.config]}
summary = {'devices': [self.config]}
# pylint: disable=protected-access
self.blink._last_events = {'foo': 'bar'}
events = {'foo': 'bar'}
vid_req.return_value = [
{
'camera_name': 'foobar',
@@ -100,16 +101,19 @@ class TestBlinkFunctions(unittest.TestCase):
'thumbnail': '/new',
}
]
self.blink.last_refresh = int(time.time())
self.blink.refresh_rate = 100000
self.blink.sync.refresh()
with mock.patch('blinkpy.blinkpy.Blink.summary_request',
return_value=summary):
with mock.patch('blinkpy.blinkpy.Blink.events_request',
return_value=events):
self.blink.refresh_rate = 0
self.blink.refresh()
test_camera = self.blink.sync.cameras['foobar']
self.assertEqual(test_camera.clip,
'https://rest.test.{}/new.mp4'.format(BLINK_URL))
self.assertEqual(test_camera.thumbnail,
'https://rest.test.{}/new.jpg'.format(BLINK_URL))
def test_set_links(self):
def test_set_links(self, mock_sess):
"""Test the link set method."""
self.blink.sync.cameras = {'foobar': self.camera}
self.blink.network_id = 9999
@@ -121,12 +125,15 @@ class TestBlinkFunctions(unittest.TestCase):
"{}/camera/1111/".format(net_url))
@mock.patch('blinkpy.blinkpy.http_req')
def test_backup_url(self, req):
def test_backup_url(self, req, mock_sess):
"""Test backup login method."""
fake_req = Request('POST', 'http://wrong.url').prepare()
req.side_effect = [
mresp.mocked_requests_post(None),
mresp.mocked_session_send(fake_req),
{'authtoken': {'authtoken': 'foobar123'}}
]
self.blink.get_auth_token()
self.assertEqual(self.blink.region_id, 'piri')
self.assertEqual(self.blink.region, 'UNKNOWN')
# pylint: disable=protected-access
self.assertEqual(self.blink._token, 'foobar123')
tests/test_blink_setup.py
+14 -27
View File
@@ -11,14 +11,16 @@ from unittest import mock
from blinkpy import blinkpy
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.helpers.util import (
http_req, BlinkAuthenticationException, BlinkException,
BlinkURLHandler)
http_req, create_session, BlinkAuthenticationException,
BlinkException, BlinkURLHandler)
import tests.mock_responses as mresp
USERNAME = 'foobar'
PASSWORD = 'deadbeef'
@mock.patch('blinkpy.helpers.util.Session.send',
side_effect=mresp.mocked_session_send)
class TestBlinkSetup(unittest.TestCase):
"""Test the Blink class in blinkpy."""
@@ -34,14 +36,14 @@ class TestBlinkSetup(unittest.TestCase):
self.blink = None
self.blink_no_cred = None
def test_initialization(self):
def test_initialization(self, mock_sess):
"""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):
def test_no_credentials(self, mock_sess):
"""Check that we throw an exception when no username/password."""
with self.assertRaises(BlinkAuthenticationException):
self.blink_no_cred.get_auth_token()
@@ -50,7 +52,7 @@ class TestBlinkSetup(unittest.TestCase):
with self.assertRaises(BlinkAuthenticationException):
self.blink_no_cred.get_auth_token()
def test_no_auth_header(self):
def test_no_auth_header(self, mock_sess):
"""Check that we throw an exception when no auth header given."""
# pylint: disable=unused-variable
(region_id, region), = mresp.LOGIN_RESPONSE['region'].items()
@@ -60,10 +62,8 @@ class TestBlinkSetup(unittest.TestCase):
with self.assertRaises(BlinkException):
self.blink.summary_request()
@mock.patch('blinkpy.helpers.util.requests.post',
side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.blinkpy.getpass.getpass')
def test_manual_login(self, getpwd, mock_post):
def test_manual_login(self, getpwd, mock_sess):
"""Check that we can manually use the login() function."""
getpwd.return_value = PASSWORD
with mock.patch('builtins.input', return_value=USERNAME):
@@ -73,35 +73,22 @@ class TestBlinkSetup(unittest.TestCase):
# pylint: disable=protected-access
self.assertEqual(self.blink_no_cred._password, PASSWORD)
@mock.patch('blinkpy.helpers.util.requests.post',
side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.helpers.util.requests.get',
side_effect=mresp.mocked_requests_get)
def test_bad_request(self, mock_get, mock_post):
def test_bad_request(self, mock_sess):
"""Check that we raise an Exception with a bad request."""
self.blink.session = create_session()
with self.assertRaises(BlinkException):
# pylint: disable=protected-access
http_req(None, reqtype='bad')
http_req(self.blink, reqtype='bad')
with self.assertRaises(BlinkAuthenticationException):
# pylint: disable=protected-access
http_req(None, reqtype='post', is_retry=True)
http_req(self.blink, reqtype='post', is_retry=True)
@mock.patch('blinkpy.helpers.util.requests.post',
side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.helpers.util.requests.get',
side_effect=mresp.mocked_requests_get)
def test_authentication(self, mock_get, mock_post):
def test_authentication(self, mock_sess):
"""Check that we can authenticate Blink up properly."""
authtoken = self.blink.get_auth_token()['TOKEN_AUTH']
expected = mresp.LOGIN_RESPONSE['authtoken']['authtoken']
self.assertEqual(authtoken, expected)
@mock.patch('blinkpy.helpers.util.requests.post',
side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.helpers.util.requests.get',
side_effect=mresp.mocked_requests_get)
def test_reauthorization_attempt(self, mock_get, mock_post):
def test_reauthorization_attempt(self, mock_sess):
"""Check that we can reauthorize after first unsuccessful attempt."""
original_header = self.blink.get_auth_token()
# pylint: disable=protected-access
tests/test_cameras.py
+9 -12
View File
@@ -9,7 +9,7 @@ Blink system is set up.
import unittest
from unittest import mock
from blinkpy import blinkpy
from blinkpy.helpers.util import BlinkURLHandler
from blinkpy.helpers.util import create_session, BlinkURLHandler
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.camera import BlinkCamera
from blinkpy.helpers.constants import BLINK_URL
@@ -30,6 +30,8 @@ CAMERA_CFG = {
}
@mock.patch('blinkpy.helpers.util.Session.send',
side_effect=mresp.mocked_session_send)
class TestBlinkCameraSetup(unittest.TestCase):
"""Test the Blink class in blinkpy."""
@@ -54,6 +56,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
'Host': 'abc.zxc',
'TOKEN_AUTH': mresp.LOGIN_RESPONSE['authtoken']['authtoken']
}
self.blink.session = create_session()
self.blink.urls = BlinkURLHandler('test')
self.blink.network_id = '0000'
self.sync = BlinkSyncModule(self.blink, header, self.blink.urls)
@@ -64,11 +67,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
@mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request',
return_value=CAMERA_CFG)
@mock.patch('blinkpy.helpers.util.requests.post',
side_effect=mresp.mocked_requests_post)
@mock.patch('blinkpy.helpers.util.requests.get',
side_effect=mresp.mocked_requests_get)
def test_camera_properties(self, mock_get, mock_post, mock_cfg):
def test_camera_properties(self, mock_cfg, mock_sess):
"""Tests all property set/recall."""
self.blink.urls = BlinkURLHandler('test')
@@ -126,7 +125,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
camera.update(camera_config, skip_cache=True)
self.assertEqual(camera.battery_string, "Unknown")
def test_camera_case(self):
def test_camera_case(self, mock_sess):
"""Tests camera case sensitivity."""
camera_object = BlinkCamera(self.camera_config, self.sync)
self.sync.cameras['foobar'] = camera_object
@@ -134,7 +133,7 @@ class TestBlinkCameraSetup(unittest.TestCase):
@mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request',
return_value=CAMERA_CFG)
def test_camera_attributes(self, mock_cfg):
def test_camera_attributes(self, mock_cfg, mock_sess):
"""Tests camera attributes."""
self.blink.urls = BlinkURLHandler('test')
@@ -166,10 +165,8 @@ class TestBlinkCameraSetup(unittest.TestCase):
self.assertEqual(camera_attr['wifi_strength'], -30)
@mock.patch('blinkpy.camera.BlinkCamera.image_refresh',
side_effect='refresh/url')
@mock.patch('blinkpy.helpers.util.requests.get',
side_effect=mresp.mocked_requests_get)
def test_camera_cache(self, req, img_refresh):
return_value='https://fake.url')
def test_camera_cache(self, img_refresh, mock_sess):
"""Tests camera cache."""
update_vals = {
'name': 'foobar',
tests/test_sync_module.py
+1 -1
View File
@@ -84,7 +84,7 @@ class TestBlinkSyncModule(unittest.TestCase):
self.assertEqual(self.blink.sync.videos['foobar'][0]['thumb'],
'/test/thumb')
@mock.patch('blinkpy.sync_module.BlinkSyncModule.refresh')
@mock.patch('blinkpy.blinkpy.Blink.refresh')
@mock.patch('blinkpy.sync_module.BlinkSyncModule._summary_request')
@mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request')
def test_get_cameras(self, vid_req, req, refresh):