From e1ff8510686f219fcfe97eee3f2f6113e083fb67 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Tue, 25 Sep 2018 14:16:14 -0400 Subject: [PATCH 1/9] First cut of refactoring --- blinkpy/blinkpy.py | 538 +++----------------------------------- blinkpy/camera.py | 219 ++++++++++++++++ blinkpy/helpers/util.py | 75 ++++++ blinkpy/sync_module.py | 207 +++++++++++++++ tests/mock_responses.py | 2 +- tests/test_blink_setup.py | 43 +-- 6 files changed, 567 insertions(+), 517 deletions(-) create mode 100644 blinkpy/camera.py create mode 100644 blinkpy/helpers/util.py create mode 100644 blinkpy/sync_module.py diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index 1a3c952..74c3f8c 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -1,4 +1,4 @@ -#!/usr/bin/python + # -*- coding: utf-8 -*- """ blinkpy by Kevin Fronczak - A Blink camera Python library. @@ -12,301 +12,24 @@ owned by Immedia Inc., see www.blinkforhome.com for more information. I am in no way affiliated with Blink, nor Immedia Inc. """ +import time import json import getpass -from shutil import copyfileobj import logging -import time -import requests -from requests.structures import CaseInsensitiveDict import blinkpy.helpers.errors as ERROR +from blinkpy.sync_module import BlinkSyncModule +from blinkpy.helpers.util import ( + http_req, BlinkURLHandler, BlinkException, BlinkAuthenticationException) from blinkpy.helpers.constants import ( - BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL, - DEFAULT_URL, ONLINE -) + DEFAULT_URL, BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL) + +REFRESH_RATE = 30 _LOGGER = logging.getLogger('blinkpy') -MAX_CLIPS = 5 -REFRESH_RATE = 30 - - -def _attempt_reauthorization(blink): - """Attempt to refresh auth token and links.""" - _LOGGER.debug("Auth token expired, attempting reauthorization.") - headers = blink.get_auth_token() - blink.set_links() - return headers - - -def _request(blink, url='http://google.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) - elif reqtype == 'get': - response = requests.get(url, headers=headers, - stream=stream) - else: - raise BlinkException(ERROR.REQUEST) - - if json_resp and 'code' in response.json(): - if is_retry: - raise BlinkAuthenticationException( - (response.json()['code'], response.json()['message'])) - else: - headers = _attempt_reauthorization(blink) - return _request(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 - - -# pylint: disable=super-init-not-called -class BlinkException(Exception): - """Class to throw general blink exception.""" - - def __init__(self, errcode): - """Initialize BlinkException.""" - self.errid = errcode[0] - self.message = errcode[1] - - -class BlinkAuthenticationException(BlinkException): - """Class to throw authentication exception.""" - - pass - - -class BlinkURLHandler(): - """Class that handles Blink URLS.""" - - def __init__(self, region_id): - """Initialize the urls.""" - self.base_url = "https://rest.{}.{}".format(region_id, BLINK_URL) - self.home_url = "{}/homescreen".format(self.base_url) - self.event_url = "{}/events/network".format(self.base_url) - self.network_url = "{}/network".format(self.base_url) - self.networks_url = "{}/networks".format(self.base_url) - self.video_url = "{}/api/v2/videos".format(self.base_url) - _LOGGER.debug("Setting base url to %s.", self.base_url) - - -class BlinkCamera(): - """Class to initialize individual camera.""" - - def __init__(self, config, blink): - """Initiailize BlinkCamera.""" - self.blink = blink - self.urls = self.blink.urls - self.id = str(config['device_id']) # pylint: disable=invalid-name - self.name = config['name'] - self._status = config['active'] - self.thumbnail = "{}{}.jpg".format(self.urls.base_url, - config['thumbnail']) - self.clip = "{}{}".format(self.urls.base_url, config['video']) - self.temperature = config['temp'] - self._battery_string = config['battery'] - self.notifications = config['notifications'] - self.motion = dict() - self.header = None - self.image_link = None - self.arm_link = None - self.region_id = config['region_id'] - self.battery_voltage = -180 - self.motion_detected = None - self.wifi_strength = None - self.camera_config = dict() - self.motion_enabled = None - self.last_record = list() - self._cached_image = None - self._cached_video = None - - @property - def attributes(self): - """Return dictionary of all camera attributes.""" - attributes = { - 'name': self.name, - 'device_id': self.id, - 'status': self._status, - 'armed': self.armed, - 'temperature': self.temperature, - 'temperature_c': self.temperature_c, - 'battery': self.battery, - 'thumbnail': self.thumbnail, - 'video': self.clip, - 'motion_enabled': self.motion_enabled, - 'notifications': self.notifications, - 'motion_detected': self.motion_detected, - 'wifi_strength': self.wifi_strength, - 'network_id': self.blink.network_id, - 'last_record': self.last_record - } - return attributes - - @property - def status(self): - """Return camera status.""" - return self._status - - @property - def armed(self): - """Return camera arm status.""" - return True if self._status == 'armed' else False - - @property - def battery(self): - """Return battery level as percentage.""" - return round(self.battery_voltage / 180 * 100) - - @property - def battery_string(self): - """Return string indicating battery status.""" - status = "Unknown" - if self._battery_string > 1 and self._battery_string <= 3: - status = "OK" - elif self._battery_string >= 0: - status = "Low" - return status - - @property - def temperature_c(self): - """Return temperature in celcius.""" - return round((self.temperature - 32) / 9.0 * 5.0, 1) - - @property - def image_from_cache(self): - """Return the most recently cached image.""" - if self._cached_image: - return self._cached_image - return None - - @property - def video_from_cache(self): - """Return the most recently cached video.""" - if self._cached_video: - return self._cached_video - return None - - def snap_picture(self): - """Take a picture with camera to create a new thumbnail.""" - _request(self.blink, url=self.image_link, - headers=self.header, reqtype='post') - - def set_motion_detect(self, enable): - """Set motion detection.""" - url = self.arm_link - if enable: - _request(self.blink, url=url + 'enable', - headers=self.header, reqtype='post') - else: - _request(self.blink, url=url + 'disable', - headers=self.header, reqtype='post') - - def update(self, values, force_cache=False, skip_cache=False): - """Update camera information.""" - self.name = values['name'] - self._status = values['active'] - self.clip = "{}{}".format( - self.urls.base_url, values['video']) - new_thumbnail = "{}{}.jpg".format( - self.urls.base_url, values['thumbnail']) - self._battery_string = values['battery'] - self.notifications = values['notifications'] - - update_cached_image = False - if new_thumbnail != self.thumbnail or self._cached_image is None: - update_cached_image = True - self.thumbnail = new_thumbnail - try: - cfg = self.blink.camera_config_request(self.id) - self.camera_config = cfg - except requests.exceptions.RequestException as err: - _LOGGER.warning("Could not get config for %s with id %s", - self.name, self.id) - _LOGGER.warning("Exception raised: %s", err) - - try: - self.battery_voltage = cfg['camera'][0]['battery_voltage'] - self.motion_enabled = cfg['camera'][0]['motion_alert'] - self.wifi_strength = cfg['camera'][0]['wifi_strength'] - self.temperature = cfg['camera'][0]['temperature'] - except KeyError: - _LOGGER.warning("Problem extracting config for camera %s", - self.name) - - # Check if the most recent clip is included in the last_record list - # and that the last_record list is populated - try: - records = sorted(self.blink.record_dates[self.name]) - new_clip = records.pop() - if new_clip not in self.last_record and self.last_record: - self.motion_detected = True - self.last_record.insert(0, new_clip) - if len(self.last_record) > MAX_CLIPS: - self.last_record.pop() - elif not self.last_record: - self.last_record.insert(0, new_clip) - self.motion_detected = False - else: - self.motion_detected = False - except KeyError: - _LOGGER.warning("Could not extract clip info from camera %s", - self.name) - - if not skip_cache: - if update_cached_image or force_cache: - self._cached_image = _request( - self.blink, url=self.image_refresh(), headers=self.header, - reqtype='get', stream=True, json_resp=False) - if (self.clip is None) or self.motion_detected or force_cache: - self._cached_video = _request( - self.blink, url=self.clip, headers=self.header, - reqtype='get', stream=True, json_resp=False) - - def image_refresh(self): - """Refresh current thumbnail.""" - url = self.urls.home_url - response = _request(self.blink, url=url, headers=self.header, - reqtype='get')['devices'] - for element in response: - try: - if str(element['device_id']) == self.id: - self.thumbnail = ( - "{}{}.jpg".format( - self.urls.base_url, element['thumbnail']) - ) - return self.thumbnail - except KeyError: - pass - return None - - def image_to_file(self, path): - """Write image to file.""" - _LOGGER.debug("Writing image from %s to %s", self.name, path) - response = self._cached_image - if response.status_code == 200: - with open(path, 'wb') as imgfile: - copyfileobj(response.raw, imgfile) - else: - _LOGGER.error("Cannot write image to file, response %s", - response.status_code) - - def video_to_file(self, path): - """Write video to file.""" - _LOGGER.debug("Writing video from %s to %s", self.name, path) - response = self._cached_video - with open(path, 'wb') as vidfile: - copyfileobj(response.raw, vidfile) - class Blink(): - """Class to initialize communication and sync module.""" + """Class to initialize communication.""" def __init__(self, username=None, password=None, refresh_rate=REFRESH_RATE): @@ -315,190 +38,22 @@ class Blink(): 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._host = None self._events = [] - self.cameras = CaseInsensitiveDict({}) - self._idlookup = {} + self.network_id = None + self.account_id = None self.urls = None - self._video_count = 0 - self._all_videos = {} - self._summary = None - self.record_dates = dict() - self.refresh_rate = refresh_rate + self.sync = None + self.region = None + self.region_id = None self.last_refresh = None - - @property - def camera_thumbs(self): - """Return camera thumbnails.""" - self.refresh() - data = {} - for name, camera in self.cameras.items(): - data[name] = camera.thumbnail - - return data - - @property - def id_table(self): - """Return id/camera pairs.""" - return self._idlookup - - @property - def video_count(self): - """Return number of videos on server.""" - url = "{}/count".format(self.urls.video_url) - headers = self._auth_header - self._video_count = _request(self, url=url, headers=headers, - reqtype='get')['count'] - return self._video_count + self.refresh_rate = refresh_rate @property def events(self): """Get all events on server.""" return self._events - @property - def online(self): - """Return boolean system online status.""" - return ONLINE[self._status_request()['syncmodule']['status']] - - @property - def videos(self): - """Return video list.""" - return self._all_videos - - @property - def summary(self): - """Get a full summary of device information.""" - return self._summary - - @property - def arm(self): - """Return status of sync module: armed/disarmed.""" - return self.summary['network']['armed'] - - @arm.setter - def arm(self, value): - """Arm or disarm system.""" - if value: - value_to_append = 'arm' - else: - value_to_append = 'disarm' - url = "{}/{}/{}".format(self.urls.network_url, - self.network_id, - value_to_append) - _request(self, url=url, headers=self._auth_header, reqtype='post') - - def refresh(self, force_cache=False): - """Get all blink cameras and pulls their most recent status.""" - current_time = int(time.time()) - last_refresh = self.last_refresh - if last_refresh is None: - last_refresh = 0 - force_cache = True - if current_time >= (last_refresh + self.refresh_rate): - _LOGGER.debug("Attempting refresh of cameras.") - self._summary = self._summary_request() - self._events = self._events_request() - response = self.summary['devices'] - self.get_videos() - for name in self.cameras: - camera = self.cameras[name] - for element in response: - try: - if str(element['device_id']) == camera.id: - element['video'] = self.videos[name][0]['clip'] - thumb = self.videos[name][0]['thumb'] - element['thumbnail'] = thumb - camera.update(element, force_cache=force_cache) - except KeyError: - pass - self.last_refresh = int(time.time()) - - def get_videos(self, start_page=0, end_page=1): - """Retrieve last recorded videos per camera.""" - videos = list() - all_dates = dict() - for page_num in range(start_page, end_page + 1): - this_page = self._video_request(page_num) - if not this_page: - break - videos.append(this_page) - - for page in videos: - _LOGGER.debug("Retrieved video page %s", page) - for entry in page: - camera_name = entry['camera_name'] - clip_addr = entry['address'] - thumb_addr = entry['thumbnail'] - clip_date = clip_addr.split('_')[-6:] - clip_date = '_'.join(clip_date) - clip_date = clip_date.split('.')[0] - if camera_name not in all_dates: - all_dates[camera_name] = list() - all_dates[camera_name].append(clip_date) - try: - self._all_videos[camera_name].append( - { - 'clip': clip_addr, - 'thumb': thumb_addr, - } - ) - except KeyError: - self._all_videos[camera_name] = [ - { - 'clip': clip_addr, - 'thumb': thumb_addr, - } - ] - self.record_dates = all_dates - - def get_cameras(self): - """Find and creates cameras.""" - self._summary = self._summary_request() - response = self.summary['devices'] - for element in response: - if ('device_type' in element and - element['device_type'] == 'camera'): - # Add region to config - element['region_id'] = self.region_id - try: - name = element['name'] - element['video'] = self.videos[name][0]['clip'] - element['thumbnail'] = self.videos[name][0]['thumb'] - except KeyError: - element['video'] = None - element['thumbnail'] = None - device = BlinkCamera(element, self) - self.cameras[device.name] = device - self._idlookup[device.id] = device.name - self.refresh() - - 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 = "{}/{}".format(self.urls.network_url, - self.network_id) - image_url = "{}/camera/{}/thumbnail".format(network_id_url, - camera.id) - arm_url = "{}/camera/{}/".format(network_id_url, - camera.id) - camera.image_link = image_url - camera.arm_link = arm_url - camera.header = self._auth_header - - def setup_system(self): - """Legacy method support.""" - _LOGGER.warning( - ("Blink.setup_system() will be deprecated in future release. " - "Please use Blink.start() instead.") - ) - self.start() - def start(self): """ Perform full system setup. @@ -512,11 +67,12 @@ class Blink(): self.get_auth_token() self.get_ids() - self.get_videos() - if self.video_count > 0: - self.get_cameras() - self.set_links() - self._events = self._events_request() + self.sync = BlinkSyncModule(self, self._auth_header, self.urls) + self.sync.get_videos() + if self.sync.video_count > 0: + self.sync.get_cameras() + self.sync.set_links() + self._events = self.events_request() def login(self): """Prompt user for username and password.""" @@ -542,7 +98,7 @@ class Blink(): "password": self._password, "client_specifier": "iPhone 9.2 | 2.2 | 222" }) - response = _request(self, url=LOGIN_URL, headers=headers, + response = http_req(self, url=LOGIN_URL, headers=headers, data=data, json_resp=False, reqtype='post') if response.status_code == 200: response = response.json() @@ -553,7 +109,7 @@ class Blink(): "when authenticating, " "trying new url"), response.status_code ) - response = _request(self, url=LOGIN_BACKUP_URL, headers=headers, + response = http_req(self, url=LOGIN_BACKUP_URL, headers=headers, data=data, reqtype='post') self.region_id = 'piri' self.region = "UNKNOWN" @@ -574,45 +130,35 @@ class Blink(): self.network_id = str(response['networks'][0]['id']) self.account_id = str(response['networks'][0]['account_id']) - def _video_request(self, page=0): - """Perform a request for videos.""" - url = "{}/page/{}".format(self.urls.video_url, page) - headers = self._auth_header - return _request(self, url=url, headers=headers, reqtype='get') - - def _summary_request(self): - """Get blink summary.""" - url = self.urls.home_url - headers = self._auth_header - if headers is None: - raise BlinkException(ERROR.AUTH_TOKEN) - return _request(self, url=url, headers=headers, reqtype='get') - def _network_request(self): """Get network and account information.""" url = self.urls.networks_url headers = self._auth_header if headers is None: raise BlinkException(ERROR.AUTH_TOKEN) - return _request(self, url=url, headers=headers, reqtype='get') + return http_req(self, url=url, headers=headers, reqtype='get') - def _events_request(self): + def events_request(self): """Get events on server.""" url = "{}/{}".format(self.urls.event_url, self.network_id) headers = self._auth_header - return _request(self, url=url, headers=headers, reqtype='get') + return http_req(self, url=url, headers=headers, reqtype='get') - def _status_request(self): - """Get syncmodule status.""" - url = "{}/{}/syncmodules".format(self.urls.network_url, - self.network_id) + def summary_request(self): + """Get blink summary.""" + url = self.urls.home_url headers = self._auth_header - return _request(self, url=url, headers=headers, reqtype='get') + if headers is None: + raise BlinkException(ERROR.AUTH_TOKEN) + return http_req(self, url=url, headers=headers, reqtype='get') - def camera_config_request(self, camera_id): - """Retrieve more info about Blink config.""" - url = "{}/network/{}/camera/{}/config".format(self.urls.base_url, - self.network_id, - str(camera_id)) - headers = self._auth_header - return _request(self, url=url, headers=headers, reqtype='get') + def refresh(self, force_cache=False): + """Check if it is ok to refresh.""" + current_time = int(time.time()) + last_refresh = self.last_refresh + if last_refresh is None: + last_refresh = 0 + force_cache = True + if current_time >= (last_refresh + self.refresh_rate): + _LOGGER.debug("Attempting refresh of cameras.") + self.sync.refresh(force_cache=force_cache) diff --git a/blinkpy/camera.py b/blinkpy/camera.py new file mode 100644 index 0000000..af318a5 --- /dev/null +++ b/blinkpy/camera.py @@ -0,0 +1,219 @@ +"""Defines Blink cameras.""" + +from shutil import copyfileobj +import logging +from requests.exceptions import RequestException +from blinkpy.helpers.util import http_req + +_LOGGER = logging.getLogger(__name__) + +MAX_CLIPS = 5 + + +class BlinkCamera(): + """Class to initialize individual camera.""" + + def __init__(self, config, sync): + """Initiailize BlinkCamera.""" + self.blink = sync.blink + self.sync = sync + self.urls = self.blink.urls + self.id = str(config['device_id']) # pylint: disable=invalid-name + self.name = config['name'] + self._status = config['active'] + self.thumbnail = "{}{}.jpg".format(self.urls.base_url, + config['thumbnail']) + self.clip = "{}{}".format(self.urls.base_url, config['video']) + self.temperature = config['temp'] + self._battery_string = config['battery'] + self.notifications = config['notifications'] + self.motion = dict() + self.header = None + self.image_link = None + self.arm_link = None + self.region_id = config['region_id'] + self.battery_voltage = -180 + self.motion_detected = None + self.wifi_strength = None + self.camera_config = dict() + self.motion_enabled = None + self.last_record = list() + self._cached_image = None + self._cached_video = None + + @property + def attributes(self): + """Return dictionary of all camera attributes.""" + attributes = { + 'name': self.name, + 'device_id': self.id, + 'status': self._status, + 'armed': self.armed, + 'temperature': self.temperature, + 'temperature_c': self.temperature_c, + 'battery': self.battery, + 'thumbnail': self.thumbnail, + 'video': self.clip, + 'motion_enabled': self.motion_enabled, + 'notifications': self.notifications, + 'motion_detected': self.motion_detected, + 'wifi_strength': self.wifi_strength, + 'network_id': self.blink.network_id, + 'last_record': self.last_record + } + return attributes + + @property + def status(self): + """Return camera status.""" + return self._status + + @property + def armed(self): + """Return camera arm status.""" + return True if self._status == 'armed' else False + + @property + def battery(self): + """Return battery level as percentage.""" + return round(self.battery_voltage / 180 * 100) + + @property + def battery_string(self): + """Return string indicating battery status.""" + status = "Unknown" + if self._battery_string > 1 and self._battery_string <= 3: + status = "OK" + elif self._battery_string >= 0: + status = "Low" + return status + + @property + def temperature_c(self): + """Return temperature in celcius.""" + return round((self.temperature - 32) / 9.0 * 5.0, 1) + + @property + def image_from_cache(self): + """Return the most recently cached image.""" + if self._cached_image: + return self._cached_image + return None + + @property + def video_from_cache(self): + """Return the most recently cached video.""" + if self._cached_video: + return self._cached_video + return None + + def snap_picture(self): + """Take a picture with camera to create a new thumbnail.""" + http_req(self.blink, url=self.image_link, + headers=self.header, reqtype='post') + + def set_motion_detect(self, enable): + """Set motion detection.""" + url = self.arm_link + if enable: + http_req(self.blink, url=url + 'enable', + headers=self.header, reqtype='post') + else: + http_req(self.blink, url=url + 'disable', + headers=self.header, reqtype='post') + + def update(self, values, force_cache=False, skip_cache=False): + """Update camera information.""" + self.name = values['name'] + self._status = values['active'] + self.clip = "{}{}".format( + self.urls.base_url, values['video']) + new_thumbnail = "{}{}.jpg".format( + self.urls.base_url, values['thumbnail']) + self._battery_string = values['battery'] + self.notifications = values['notifications'] + + update_cached_image = False + if new_thumbnail != self.thumbnail or self._cached_image is None: + update_cached_image = True + self.thumbnail = new_thumbnail + try: + cfg = self.sync.camera_confighttp_req(self.id) + self.camera_config = cfg + except RequestException as err: + _LOGGER.warning("Could not get config for %s with id %s", + self.name, self.id) + _LOGGER.warning("Exception raised: %s", err) + + try: + self.battery_voltage = cfg['camera'][0]['battery_voltage'] + self.motion_enabled = cfg['camera'][0]['motion_alert'] + self.wifi_strength = cfg['camera'][0]['wifi_strength'] + self.temperature = cfg['camera'][0]['temperature'] + except KeyError: + _LOGGER.warning("Problem extracting config for camera %s", + self.name) + + # Check if the most recent clip is included in the last_record list + # and that the last_record list is populated + try: + records = sorted(self.sync.record_dates[self.name]) + new_clip = records.pop() + if new_clip not in self.last_record and self.last_record: + self.motion_detected = True + self.last_record.insert(0, new_clip) + if len(self.last_record) > MAX_CLIPS: + self.last_record.pop() + elif not self.last_record: + self.last_record.insert(0, new_clip) + self.motion_detected = False + else: + self.motion_detected = False + except KeyError: + _LOGGER.warning("Could not extract clip info from camera %s", + self.name) + + if not skip_cache: + if update_cached_image or force_cache: + self._cached_image = http_req( + self.blink, url=self.image_refresh(), headers=self.header, + reqtype='get', stream=True, json_resp=False) + if (self.clip is None) or self.motion_detected or force_cache: + self._cached_video = http_req( + self.blink, url=self.clip, headers=self.header, + reqtype='get', stream=True, json_resp=False) + + def image_refresh(self): + """Refresh current thumbnail.""" + url = self.urls.home_url + response = http_req(self.blink, url=url, headers=self.header, + reqtype='get')['devices'] + for element in response: + try: + if str(element['device_id']) == self.id: + self.thumbnail = ( + "{}{}.jpg".format( + self.urls.base_url, element['thumbnail']) + ) + return self.thumbnail + except KeyError: + pass + return None + + def image_to_file(self, path): + """Write image to file.""" + _LOGGER.debug("Writing image from %s to %s", self.name, path) + response = self._cached_image + if response.status_code == 200: + with open(path, 'wb') as imgfile: + copyfileobj(response.raw, imgfile) + else: + _LOGGER.error("Cannot write image to file, response %s", + response.status_code) + + def video_to_file(self, path): + """Write video to file.""" + _LOGGER.debug("Writing video from %s to %s", self.name, path) + response = self._cached_video + with open(path, 'wb') as vidfile: + copyfileobj(response.raw, vidfile) diff --git a/blinkpy/helpers/util.py b/blinkpy/helpers/util.py new file mode 100644 index 0000000..4ac805e --- /dev/null +++ b/blinkpy/helpers/util.py @@ -0,0 +1,75 @@ +"""Useful functions for blinkpy.""" + +import logging +import requests +import blinkpy.helpers.errors as ERROR +from blinkpy.helpers.constants import BLINK_URL + + +_LOGGER = logging.getLogger(__name__) + + +def attempt_reauthorization(blink): + """Attempt to refresh auth token and links.""" + _LOGGER.debug("Auth token expired, attempting reauthorization.") + headers = blink.get_auth_token() + blink.sync.set_links() + return headers + + +def http_req(blink, url='http://google.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) + elif reqtype == 'get': + response = requests.get(url, headers=headers, + stream=stream) + else: + raise BlinkException(ERROR.REQUEST) + + if json_resp and 'code' in response.json(): + if is_retry: + raise BlinkAuthenticationException( + (response.json()['code'], response.json()['message'])) + else: + headers = attempt_reauthorization(blink) + 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 + + +# pylint: disable=super-init-not-called +class BlinkException(Exception): + """Class to throw general blink exception.""" + + def __init__(self, errcode): + """Initialize BlinkException.""" + self.errid = errcode[0] + self.message = errcode[1] + + +class BlinkAuthenticationException(BlinkException): + """Class to throw authentication exception.""" + + pass + + +class BlinkURLHandler(): + """Class that handles Blink URLS.""" + + def __init__(self, region_id): + """Initialize the urls.""" + self.base_url = "https://rest.{}.{}".format(region_id, BLINK_URL) + self.home_url = "{}/homescreen".format(self.base_url) + self.event_url = "{}/events/network".format(self.base_url) + self.network_url = "{}/network".format(self.base_url) + self.networks_url = "{}/networks".format(self.base_url) + self.video_url = "{}/api/v2/videos".format(self.base_url) + _LOGGER.debug("Setting base url to %s.", self.base_url) diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py new file mode 100644 index 0000000..79ababe --- /dev/null +++ b/blinkpy/sync_module.py @@ -0,0 +1,207 @@ +"""Defines a sync module for Blink.""" + +import logging +import time + +from requests.structures import CaseInsensitiveDict +from blinkpy.camera import BlinkCamera +from blinkpy.helpers.util import http_req, BlinkException +from blinkpy.helpers.constants import ONLINE +import blinkpy.helpers.errors as ERROR + +_LOGGER = logging.getLogger(__name__) + + +class BlinkSyncModule(): + """Class to initialize sync module.""" + + def __init__(self, blink, header, urls): + """Initialize Blink sync module.""" + self.blink = blink + self._auth_header = header + self.sync_id = None + self.region = None + self.region_id = None + self._host = None + self._events = [] + self.cameras = CaseInsensitiveDict({}) + self._idlookup = {} + self.urls = urls + self._video_count = 0 + self._all_videos = {} + self._summary = None + self.record_dates = dict() + + @property + def camera_thumbs(self): + """Return camera thumbnails.""" + self.refresh() + data = {} + for name, camera in self.cameras.items(): + data[name] = camera.thumbnail + + return data + + @property + def id_table(self): + """Return id/camera pairs.""" + return self._idlookup + + @property + def video_count(self): + """Return number of videos on server.""" + url = "{}/count".format(self.urls.video_url) + headers = self._auth_header + self._video_count = http_req(self.blink, url=url, headers=headers, + reqtype='get')['count'] + return self._video_count + + @property + def events(self): + """Get all events on server.""" + return self._events + + @property + def online(self): + """Return boolean system online status.""" + return ONLINE[self._status_request()['syncmodule']['status']] + + @property + def videos(self): + """Return video list.""" + return self._all_videos + + @property + def summary(self): + """Get a full summary of device information.""" + return self._summary + + @property + def arm(self): + """Return status of sync module: armed/disarmed.""" + return self.summary['network']['armed'] + + @arm.setter + def arm(self, value): + """Arm or disarm system.""" + if value: + value_to_append = 'arm' + else: + value_to_append = 'disarm' + url = "{}/{}/{}".format(self.urls.network_url, + self.blink.network_id, + value_to_append) + http_req(self.blink, url=url, headers=self._auth_header, + reqtype='post') + + def refresh(self, force_cache=False): + """Get all blink cameras and pulls their most recent status.""" + self._summary = self._summary_request() + self._events = self.blink.events_request() + response = self.summary['devices'] + self.get_videos() + for name in self.cameras: + camera = self.cameras[name] + for element in response: + try: + if str(element['device_id']) == camera.id: + element['video'] = self.videos[name][0]['clip'] + thumb = self.videos[name][0]['thumb'] + element['thumbnail'] = thumb + camera.update(element, force_cache=force_cache) + except KeyError: + pass + + def get_videos(self, start_page=0, end_page=1): + """Retrieve last recorded videos per camera.""" + videos = list() + all_dates = dict() + for page_num in range(start_page, end_page + 1): + this_page = self._video_request(page_num) + if not this_page: + break + videos.append(this_page) + + for page in videos: + _LOGGER.debug("Retrieved video page %s", page) + for entry in page: + camera_name = entry['camera_name'] + clip_addr = entry['address'] + thumb_addr = entry['thumbnail'] + clip_date = clip_addr.split('_')[-6:] + clip_date = '_'.join(clip_date) + clip_date = clip_date.split('.')[0] + if camera_name not in all_dates: + all_dates[camera_name] = list() + all_dates[camera_name].append(clip_date) + try: + self._all_videos[camera_name].append( + { + 'clip': clip_addr, + 'thumb': thumb_addr, + } + ) + except KeyError: + self._all_videos[camera_name] = [ + { + 'clip': clip_addr, + 'thumb': thumb_addr, + } + ] + self.record_dates = all_dates + + def get_cameras(self): + """Find and creates cameras.""" + self._summary = self._summary_request() + response = self.summary['devices'] + for element in response: + if ('device_type' in element and + element['device_type'] == 'camera'): + # Add region to config + element['region_id'] = self.region_id + try: + name = element['name'] + element['video'] = self.videos[name][0]['clip'] + element['thumbnail'] = self.videos[name][0]['thumb'] + except KeyError: + element['video'] = None + element['thumbnail'] = None + device = BlinkCamera(element, self) + self.cameras[device.name] = device + self._idlookup[device.id] = device.name + self.refresh() + + 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 = "{}/{}".format(self.urls.network_url, + self.blink.network_id) + image_url = "{}/camera/{}/thumbnail".format(network_id_url, + camera.id) + arm_url = "{}/camera/{}/".format(network_id_url, + camera.id) + camera.image_link = image_url + camera.arm_link = arm_url + camera.header = self._auth_header + + def _video_request(self, page=0): + """Perform a request for videos.""" + url = "{}/page/{}".format(self.urls.video_url, page) + headers = self._auth_header + return http_req(self.blink, url=url, headers=headers, reqtype='get') + + def _status_request(self): + """Get syncmodule status.""" + url = "{}/{}/syncmodules".format(self.urls.network_url, + self.blink.network_id) + headers = self._auth_header + return http_req(self.blink, url=url, headers=headers, reqtype='get') + + def camera_config_request(self, camera_id): + """Retrieve more info about Blink config.""" + url = "{}/network/{}/camera/{}/config".format(self.urls.base_url, + self.blink.network_id, + str(camera_id)) + headers = self._auth_header + return http_req(self.blink, url=url, headers=headers, reqtype='get') diff --git a/tests/mock_responses.py b/tests/mock_responses.py index d626a78..5fffd6f 100644 --- a/tests/mock_responses.py +++ b/tests/mock_responses.py @@ -1,6 +1,6 @@ """Simple mock responses definitions.""" -from blinkpy.blinkpy import BlinkURLHandler +from blinkpy.helpers.util import BlinkURLHandler import blinkpy.helpers.constants as const LOGIN_RESPONSE = { diff --git a/tests/test_blink_setup.py b/tests/test_blink_setup.py index 86fa797..7b8e4fb 100644 --- a/tests/test_blink_setup.py +++ b/tests/test_blink_setup.py @@ -9,6 +9,10 @@ any communication related errors at startup. import unittest from unittest import mock from blinkpy import blinkpy +from blinkpy.sync_module import BlinkSyncModule +from blinkpy.helpers.util import ( + http_req, BlinkAuthenticationException, BlinkException, + BlinkURLHandler) import tests.mock_responses as mresp USERNAME = 'foobar' @@ -23,6 +27,7 @@ class TestBlinkSetup(unittest.TestCase): self.blink_no_cred = blinkpy.Blink() self.blink = blinkpy.Blink(username=USERNAME, password=PASSWORD) + self.blink.sync = BlinkSyncModule(self.blink, dict(), self.blink.urls) def tearDown(self): """Clean up after test.""" @@ -38,25 +43,24 @@ class TestBlinkSetup(unittest.TestCase): def test_no_credentials(self): """Check that we throw an exception when no username/password.""" - with self.assertRaises(blinkpy.BlinkAuthenticationException): + with self.assertRaises(BlinkAuthenticationException): self.blink_no_cred.get_auth_token() # pylint: disable=protected-access self.blink_no_cred._username = USERNAME - with self.assertRaises(blinkpy.BlinkAuthenticationException): + with self.assertRaises(BlinkAuthenticationException): self.blink_no_cred.get_auth_token() def test_no_auth_header(self): """Check that we throw an exception 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.urls = BlinkURLHandler(region_id) + with self.assertRaises(BlinkException): self.blink.get_ids() - with self.assertRaises(blinkpy.BlinkException): - # pylint: disable=protected-access - self.blink._summary_request() + with self.assertRaises(BlinkException): + self.blink.summary_request() - @mock.patch('blinkpy.blinkpy.requests.post', + @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): @@ -69,23 +73,23 @@ class TestBlinkSetup(unittest.TestCase): # pylint: disable=protected-access self.assertEqual(self.blink_no_cred._password, PASSWORD) - @mock.patch('blinkpy.blinkpy.requests.post', + @mock.patch('blinkpy.helpers.util.requests.post', side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', + @mock.patch('blinkpy.helpers.util.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): + with self.assertRaises(BlinkException): # pylint: disable=protected-access - blinkpy._request(None, reqtype='bad') + http_req(None, reqtype='bad') - with self.assertRaises(blinkpy.BlinkAuthenticationException): + with self.assertRaises(BlinkAuthenticationException): # pylint: disable=protected-access - blinkpy._request(None, reqtype='post', is_retry=True) + http_req(None, reqtype='post', is_retry=True) - @mock.patch('blinkpy.blinkpy.requests.post', + @mock.patch('blinkpy.helpers.util.requests.post', side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', + @mock.patch('blinkpy.helpers.util.requests.get', side_effect=mresp.mocked_requests_get) def test_authentication(self, mock_get, mock_post): """Check that we can authenticate Blink up properly.""" @@ -93,9 +97,9 @@ class TestBlinkSetup(unittest.TestCase): expected = mresp.LOGIN_RESPONSE['authtoken']['authtoken'] self.assertEqual(authtoken, expected) - @mock.patch('blinkpy.blinkpy.requests.post', + @mock.patch('blinkpy.helpers.util.requests.post', side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', + @mock.patch('blinkpy.helpers.util.requests.get', side_effect=mresp.mocked_requests_get) def test_reauthorization_attempt(self, mock_get, mock_post): """Check that we can reauthorize after first unsuccessful attempt.""" @@ -106,7 +110,6 @@ class TestBlinkSetup(unittest.TestCase): self.blink._auth_header = bad_header # pylint: disable=protected-access self.assertEqual(self.blink._auth_header, bad_header) - # pylint: disable=protected-access - self.blink._summary_request() + self.blink.summary_request() # pylint: disable=protected-access self.assertEqual(self.blink._auth_header, original_header) From 40e9e4bd6cb34bac1f0f7eef1738dc2e8bbda640 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Tue, 25 Sep 2018 14:31:23 -0400 Subject: [PATCH 2/9] Fixed camera tests --- blinkpy/camera.py | 2 +- tests/test_blink_cameras.py | 44 ++++++++++++++++++++----------------- 2 files changed, 25 insertions(+), 21 deletions(-) diff --git a/blinkpy/camera.py b/blinkpy/camera.py index af318a5..5cc314c 100644 --- a/blinkpy/camera.py +++ b/blinkpy/camera.py @@ -138,7 +138,7 @@ class BlinkCamera(): update_cached_image = True self.thumbnail = new_thumbnail try: - cfg = self.sync.camera_confighttp_req(self.id) + cfg = self.sync.camera_config_request(self.id) self.camera_config = cfg except RequestException as err: _LOGGER.warning("Could not get config for %s with id %s", diff --git a/tests/test_blink_cameras.py b/tests/test_blink_cameras.py index c358e48..068de45 100644 --- a/tests/test_blink_cameras.py +++ b/tests/test_blink_cameras.py @@ -9,6 +9,9 @@ Blink system is set up. import unittest from unittest import mock from blinkpy import blinkpy +from blinkpy.helpers.util import BlinkURLHandler +from blinkpy.sync_module import BlinkSyncModule +from blinkpy.camera import BlinkCamera from blinkpy.helpers.constants import BLINK_URL import tests.mock_responses as mresp @@ -47,29 +50,30 @@ class TestBlinkCameraSetup(unittest.TestCase): 'region_id': 'test' } - self.blink.urls = blinkpy.BlinkURLHandler('test') + self.blink.urls = BlinkURLHandler('test') self.blink.network_id = '0000' + self.blink.sync = BlinkSyncModule(self.blink, dict(), self.blink.urls) def tearDown(self): """Clean up after test.""" self.blink = None - @mock.patch('blinkpy.blinkpy.Blink.camera_config_request', + @mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request', return_value=CAMERA_CFG) - @mock.patch('blinkpy.blinkpy.requests.post', + @mock.patch('blinkpy.helpers.util.requests.post', side_effect=mresp.mocked_requests_post) - @mock.patch('blinkpy.blinkpy.requests.get', + @mock.patch('blinkpy.helpers.util.requests.get', side_effect=mresp.mocked_requests_get) def test_camera_properties(self, mock_get, mock_post, mock_cfg): """Tests all property set/recall.""" - self.blink.urls = blinkpy.BlinkURLHandler('test') + self.blink.urls = BlinkURLHandler('test') - self.blink.cameras = { - 'foobar': blinkpy.BlinkCamera(self.camera_config, self.blink) + self.blink.sync.cameras = { + 'foobar': BlinkCamera(self.camera_config, self.blink.sync) } - for name in self.blink.cameras: - camera = self.blink.cameras[name] + for name in self.blink.sync.cameras: + camera = self.blink.sync.cameras[name] camera.update(self.camera_config, skip_cache=True) self.assertEqual(camera.id, '1111') self.assertEqual(camera.name, 'foobar') @@ -98,8 +102,8 @@ class TestBlinkCameraSetup(unittest.TestCase): camera_config['temp'] = 60 camera_config['battery'] = 0 camera_config['notifications'] = 4 - for name in self.blink.cameras: - camera = self.blink.cameras[name] + for name in self.blink.sync.cameras: + camera = self.blink.sync.cameras[name] camera.update(camera_config, skip_cache=True) self.assertEqual(camera.armed, True) self.assertEqual( @@ -120,22 +124,22 @@ class TestBlinkCameraSetup(unittest.TestCase): def test_camera_case(self): """Tests camera case sensitivity.""" - camera_object = blinkpy.BlinkCamera(self.camera_config, self.blink) - self.blink.cameras['foobar'] = camera_object - self.assertEqual(camera_object, self.blink.cameras['fOoBaR']) + camera_object = BlinkCamera(self.camera_config, self.blink.sync) + self.blink.sync.cameras['foobar'] = camera_object + self.assertEqual(camera_object, self.blink.sync.cameras['fOoBaR']) - @mock.patch('blinkpy.blinkpy.Blink.camera_config_request', + @mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request', return_value=CAMERA_CFG) def test_camera_attributes(self, mock_cfg): """Tests camera attributes.""" - self.blink.urls = blinkpy.BlinkURLHandler('test') + self.blink.urls = BlinkURLHandler('test') - self.blink.cameras = { - 'foobar': blinkpy.BlinkCamera(self.camera_config, self.blink) + self.blink.sync.cameras = { + 'foobar': BlinkCamera(self.camera_config, self.blink.sync) } - for name in self.blink.cameras: - camera = self.blink.cameras[name] + for name in self.blink.sync.cameras: + camera = self.blink.sync.cameras[name] camera.update(self.camera_config, skip_cache=True) camera_attr = camera.attributes self.assertEqual(camera_attr['device_id'], '1111') From 1c6a5eb6c63b4a4b27c4081076bd4b0cb09b5363 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Tue, 25 Sep 2018 16:02:59 -0400 Subject: [PATCH 3/9] Fixed tests. Need to improve mocking in functions test. --- blinkpy/sync_module.py | 12 ++++---- tests/test_blink_functions.py | 55 +++++++++++++++++++++-------------- 2 files changed, 40 insertions(+), 27 deletions(-) diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 79ababe..e56a86f 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -1,13 +1,11 @@ """Defines a sync module for Blink.""" import logging -import time from requests.structures import CaseInsensitiveDict from blinkpy.camera import BlinkCamera -from blinkpy.helpers.util import http_req, BlinkException +from blinkpy.helpers.util import http_req from blinkpy.helpers.constants import ONLINE -import blinkpy.helpers.errors as ERROR _LOGGER = logging.getLogger(__name__) @@ -35,7 +33,7 @@ class BlinkSyncModule(): @property def camera_thumbs(self): """Return camera thumbnails.""" - self.refresh() + self.blink.refresh() data = {} for name, camera in self.cameras.items(): data[name] = camera.thumbnail @@ -169,7 +167,7 @@ class BlinkSyncModule(): device = BlinkCamera(element, self) self.cameras[device.name] = device self._idlookup[device.id] = device.name - self.refresh() + self.blink.refresh() def set_links(self): """Set access links and required headers for each camera in system.""" @@ -185,6 +183,10 @@ class BlinkSyncModule(): camera.arm_link = arm_url camera.header = self._auth_header + def _summary_request(self): + """Request a summary from blink.""" + return self.blink.summary_request + def _video_request(self, page=0): """Perform a request for videos.""" url = "{}/page/{}".format(self.urls.video_url, page) diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index 97911c7..1b46584 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -1,8 +1,12 @@ """Tests camera and system functions.""" - import unittest from unittest import mock + +import pytest + from blinkpy import blinkpy +from blinkpy.sync_module import BlinkSyncModule +from blinkpy.camera import BlinkCamera from blinkpy.helpers.constants import BLINK_URL import tests.mock_responses as mresp @@ -36,7 +40,10 @@ class TestBlinkFunctions(unittest.TestCase): 'region_id': 'test', 'device_type': 'camera' } - self.camera = blinkpy.BlinkCamera(self.config, self.blink) + self.blink.sync = BlinkSyncModule( + self.blink, self.blink._auth_header, self.blink.urls) + + self.camera = BlinkCamera(self.config, self.blink.sync) def tearDown(self): """Clean up after test.""" @@ -44,7 +51,7 @@ class TestBlinkFunctions(unittest.TestCase): self.config = {} self.camera = None - @mock.patch('blinkpy.blinkpy._request') + @mock.patch('blinkpy.sync_module.http_req') def test_get_videos(self, req): """Test video access.""" req.return_value = [ @@ -54,16 +61,18 @@ class TestBlinkFunctions(unittest.TestCase): 'thumbnail': '/test/thumb' } ] - self.blink.get_videos() - self.assertEqual(self.blink.videos['foobar'][0]['clip'], + self.blink.sync.get_videos() + self.assertEqual(self.blink.sync.videos['foobar'][0]['clip'], '/new/test.mp4') - self.assertEqual(self.blink.videos['foobar'][0]['thumb'], + self.assertEqual(self.blink.sync.videos['foobar'][0]['thumb'], '/test/thumb') - @mock.patch('blinkpy.blinkpy._request') - @mock.patch('blinkpy.blinkpy.Blink._video_request') - def test_get_cameras(self, vid_req, req): + @mock.patch('blinkpy.sync_module.BlinkSyncModule.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): """Test camera extraction.""" + refresh.return_value = True req.return_value = {'devices': [self.config]} vid_req.return_value = [ { @@ -72,10 +81,10 @@ class TestBlinkFunctions(unittest.TestCase): 'thumbnail': '/new' } ] - self.blink.get_cameras() - self.assertTrue('foobar' in self.blink.cameras) + self.blink.sync.get_cameras() + self.assertTrue('foobar' in self.blink.sync.cameras) - @mock.patch('blinkpy.blinkpy._request') + @mock.patch('blinkpy.camera.http_req') def test_image_refresh(self, req): """Test image refresh function.""" req.return_value = {'devices': [self.config]} @@ -83,17 +92,18 @@ class TestBlinkFunctions(unittest.TestCase): self.assertEqual(image, 'https://rest.test.{}/test.jpg'.format(BLINK_URL)) - @mock.patch('blinkpy.blinkpy._request') + @mock.patch('blinkpy.sync_module.http_req') def test_video_count(self, req): """Test video count function.""" req.return_value = {'count': 1} - self.assertEqual(self.blink.video_count, 1) + self.assertEqual(self.blink.sync.video_count, 1) - @mock.patch('blinkpy.blinkpy._request') - @mock.patch('blinkpy.blinkpy.Blink._video_request') + @pytest.mark.skip(reason="Need to mock Blink class, not methods") + @mock.patch('blinkpy.sync_module.http_req') + @mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request') def test_refresh(self, vid_req, req): """Test blinkpy refresh function.""" - self.blink.cameras = {'foobar': self.camera} + self.blink.sync.cameras = {'foobar': self.camera} req.return_value = {'devices': [self.config]} vid_req.return_value = [ { @@ -102,8 +112,9 @@ class TestBlinkFunctions(unittest.TestCase): 'thumbnail': '/new' } ] - self.blink.refresh() - test_camera = self.blink.cameras['foobar'] + self.blink.last_refresh = -1 + self.blink.sync.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, @@ -111,16 +122,16 @@ class TestBlinkFunctions(unittest.TestCase): def test_set_links(self): """Test the link set method.""" - self.blink.cameras = {'foobar': self.camera} + self.blink.sync.cameras = {'foobar': self.camera} self.blink.network_id = 9999 - self.blink.set_links() + self.blink.sync.set_links() net_url = "{}/{}".format(self.blink.urls.network_url, 9999) self.assertEqual(self.camera.image_link, "{}/camera/1111/thumbnail".format(net_url)) self.assertEqual(self.camera.arm_link, "{}/camera/1111/".format(net_url)) - @mock.patch('blinkpy.blinkpy._request') + @mock.patch('blinkpy.blinkpy.http_req') def test_backup_url(self, req): """Test backup login method.""" req.side_effect = [ From 3e23374b32ccfa7a08f336f08759d2e378700dbb Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Wed, 26 Sep 2018 11:09:51 -0400 Subject: [PATCH 4/9] Modified requests so cameras only talk to sync --- blinkpy/blinkpy.py | 29 ++++++++++++++----- blinkpy/camera.py | 28 +++++++----------- blinkpy/sync_module.py | 53 +++++++++++++++++++++++------------ tests/test_blink_functions.py | 1 + 4 files changed, 68 insertions(+), 43 deletions(-) diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index 74c3f8c..cf4c25b 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -40,6 +40,8 @@ class Blink(): self._auth_header = None self._host = None self._events = [] + self._last_summary = None + self._last_events = None self.network_id = None self.account_id = None self.urls = None @@ -67,7 +69,7 @@ class Blink(): self.get_auth_token() self.get_ids() - self.sync = BlinkSyncModule(self, self._auth_header, self.urls) + self.sync = BlinkSyncModule(self, self._auth_header) self.sync.get_videos() if self.sync.video_count > 0: self.sync.get_cameras() @@ -142,7 +144,10 @@ class Blink(): """Get events on server.""" url = "{}/{}".format(self.urls.event_url, self.network_id) headers = self._auth_header - return http_req(self, url=url, headers=headers, reqtype='get') + 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') + return self._last_events def summary_request(self): """Get blink summary.""" @@ -150,15 +155,25 @@ class Blink(): headers = self._auth_header if headers is None: raise BlinkException(ERROR.AUTH_TOKEN) - return http_req(self, url=url, headers=headers, reqtype='get') + if self.check_if_ok_to_update() or self._last_summary is None: + self._last_summary = http_req( + self, url=url, headers=headers, reqtype='get') + return self._last_summary def refresh(self, force_cache=False): - """Check if it is ok to refresh.""" + """Perform a system refresh.""" + if self.last_refresh is None: + force_cache = True + if self.check_if_ok_to_update(): + _LOGGER.debug("Attempting refresh of cameras.") + self.sync.refresh(force_cache=force_cache) + + def check_if_ok_to_update(self): + """Check if it is ok to perform an http request.""" current_time = int(time.time()) last_refresh = self.last_refresh if last_refresh is None: last_refresh = 0 - force_cache = True if current_time >= (last_refresh + self.refresh_rate): - _LOGGER.debug("Attempting refresh of cameras.") - self.sync.refresh(force_cache=force_cache) + return True + return False diff --git a/blinkpy/camera.py b/blinkpy/camera.py index 5cc314c..2cc7916 100644 --- a/blinkpy/camera.py +++ b/blinkpy/camera.py @@ -3,7 +3,6 @@ from shutil import copyfileobj import logging from requests.exceptions import RequestException -from blinkpy.helpers.util import http_req _LOGGER = logging.getLogger(__name__) @@ -15,9 +14,8 @@ class BlinkCamera(): def __init__(self, config, sync): """Initiailize BlinkCamera.""" - self.blink = sync.blink self.sync = sync - self.urls = self.blink.urls + self.urls = self.sync.urls self.id = str(config['device_id']) # pylint: disable=invalid-name self.name = config['name'] self._status = config['active'] @@ -58,7 +56,7 @@ class BlinkCamera(): 'notifications': self.notifications, 'motion_detected': self.motion_detected, 'wifi_strength': self.wifi_strength, - 'network_id': self.blink.network_id, + 'network_id': self.sync.network_id, 'last_record': self.last_record } return attributes @@ -109,18 +107,15 @@ class BlinkCamera(): def snap_picture(self): """Take a picture with camera to create a new thumbnail.""" - http_req(self.blink, url=self.image_link, - headers=self.header, reqtype='post') + self.sync.http_post(self.image_link) def set_motion_detect(self, enable): """Set motion detection.""" url = self.arm_link if enable: - http_req(self.blink, url=url + 'enable', - headers=self.header, reqtype='post') + self.sync.http_post("{}{}".format(url, 'enable')) else: - http_req(self.blink, url=url + 'disable', - headers=self.header, reqtype='post') + self.sync.http_post("{}{}".format(url, 'disable')) def update(self, values, force_cache=False, skip_cache=False): """Update camera information.""" @@ -175,19 +170,16 @@ class BlinkCamera(): if not skip_cache: if update_cached_image or force_cache: - self._cached_image = http_req( - self.blink, url=self.image_refresh(), headers=self.header, - reqtype='get', stream=True, json_resp=False) + self._cached_image = self.sync.http_get( + self.image_refresh(), stream=True, json=False) if (self.clip is None) or self.motion_detected or force_cache: - self._cached_video = http_req( - self.blink, url=self.clip, headers=self.header, - reqtype='get', stream=True, json_resp=False) + self._cached_video = self.sync.http_get( + self.clip, stream=True, json=False) def image_refresh(self): """Refresh current thumbnail.""" url = self.urls.home_url - response = http_req(self.blink, url=url, headers=self.header, - reqtype='get')['devices'] + response = self.sync.http_get(url)['devices'] for element in response: try: if str(element['device_id']) == self.id: diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index e56a86f..ca03524 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -13,7 +13,7 @@ _LOGGER = logging.getLogger(__name__) class BlinkSyncModule(): """Class to initialize sync module.""" - def __init__(self, blink, header, urls): + def __init__(self, blink, header, urls=None): """Initialize Blink sync module.""" self.blink = blink self._auth_header = header @@ -24,12 +24,21 @@ class BlinkSyncModule(): self._events = [] self.cameras = CaseInsensitiveDict({}) self._idlookup = {} - self.urls = urls self._video_count = 0 self._all_videos = {} self._summary = None self.record_dates = dict() + @property + def urls(self): + """Return device urls.""" + return self.blink.urls + + @property + def network_id(self): + """Return the network id.""" + return self.blink.network_id + @property def camera_thumbs(self): """Return camera thumbnails.""" @@ -49,9 +58,7 @@ class BlinkSyncModule(): def video_count(self): """Return number of videos on server.""" url = "{}/count".format(self.urls.video_url) - headers = self._auth_header - self._video_count = http_req(self.blink, url=url, headers=headers, - reqtype='get')['count'] + self._video_count = self.http_get(url)['count'] return self._video_count @property @@ -87,15 +94,14 @@ class BlinkSyncModule(): else: value_to_append = 'disarm' url = "{}/{}/{}".format(self.urls.network_url, - self.blink.network_id, + self.network_id, value_to_append) - http_req(self.blink, url=url, headers=self._auth_header, - reqtype='post') + self.http_post(url) def refresh(self, force_cache=False): """Get all blink cameras and pulls their most recent status.""" self._summary = self._summary_request() - self._events = self.blink.events_request() + self._events = self._events_request() response = self.summary['devices'] self.get_videos() for name in self.cameras: @@ -174,7 +180,7 @@ class BlinkSyncModule(): for name in self.cameras: camera = self.cameras[name] network_id_url = "{}/{}".format(self.urls.network_url, - self.blink.network_id) + self.network_id) image_url = "{}/camera/{}/thumbnail".format(network_id_url, camera.id) arm_url = "{}/camera/{}/".format(network_id_url, @@ -187,23 +193,34 @@ class BlinkSyncModule(): """Request a summary from blink.""" return self.blink.summary_request + def _events_request(self): + """Request a list of events from blink.""" + return self.blink.events_request + def _video_request(self, page=0): """Perform a request for videos.""" url = "{}/page/{}".format(self.urls.video_url, page) - headers = self._auth_header - return http_req(self.blink, url=url, headers=headers, reqtype='get') + return self.http_get(url) def _status_request(self): """Get syncmodule status.""" url = "{}/{}/syncmodules".format(self.urls.network_url, - self.blink.network_id) - headers = self._auth_header - return http_req(self.blink, url=url, headers=headers, reqtype='get') + self.network_id) + return self.http_get(url) def camera_config_request(self, camera_id): """Retrieve more info about Blink config.""" url = "{}/network/{}/camera/{}/config".format(self.urls.base_url, - self.blink.network_id, + self.network_id, str(camera_id)) - headers = self._auth_header - return http_req(self.blink, url=url, headers=headers, reqtype='get') + return self.http_get(url) + + def http_get(self, url, stream=False, json=True): + """Perform a get request.""" + return http_req(self.blink, url=url, headers=self._auth_header, + reqtype='get', stream=stream, json_resp=json) + + def http_post(self, url): + """Perform a post request.""" + return http_req(self.blink, url=url, headers=self._auth_header, + reqtype='post') diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index 1b46584..716b60e 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -84,6 +84,7 @@ class TestBlinkFunctions(unittest.TestCase): self.blink.sync.get_cameras() self.assertTrue('foobar' in self.blink.sync.cameras) + @pytest.mark.skip(reason="Need to mock sync class.") @mock.patch('blinkpy.camera.http_req') def test_image_refresh(self, req): """Test image refresh function.""" From a570413e66b6c758a8dd98485b108f22fdb6b941 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Wed, 26 Sep 2018 12:11:41 -0400 Subject: [PATCH 5/9] Fixed broken tests...probably should re-write them --- blinkpy/sync_module.py | 4 +-- tests/test_blink_functions.py | 61 ++++++++++++++++++++++++----------- 2 files changed, 45 insertions(+), 20 deletions(-) diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index ca03524..6235c1c 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -191,11 +191,11 @@ class BlinkSyncModule(): def _summary_request(self): """Request a summary from blink.""" - return self.blink.summary_request + return self.blink.summary_request() def _events_request(self): """Request a list of events from blink.""" - return self.blink.events_request + return self.blink.events_request() def _video_request(self, page=0): """Perform a request for videos.""" diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index 716b60e..44b77f0 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -1,8 +1,10 @@ """Tests camera and system functions.""" +import time + import unittest from unittest import mock -import pytest +# import pytest from blinkpy import blinkpy from blinkpy.sync_module import BlinkSyncModule @@ -14,6 +16,28 @@ USERNAME = 'foobar' PASSWORD = 'deadbeef' +class MockSyncModule(BlinkSyncModule): + """Mock http requests from sync module.""" + + def __init__(self, blink, header): + """Create mock sync module instance.""" + super().__init__(blink, header) + self.blink = blink + self.header = header + self.return_value = None + self.return_value2 = None + + def http_get(self, url, stream=False, json=True): + """Mock get request.""" + if stream and self.return_value2 is not None: + return self.return_value2 + return self.return_value + + def http_post(self, url): + """Mock post request.""" + return self.return_value + + class TestBlinkFunctions(unittest.TestCase): """Test Blink and BlinkCamera functions in blinkpy.""" @@ -40,8 +64,8 @@ class TestBlinkFunctions(unittest.TestCase): 'region_id': 'test', 'device_type': 'camera' } - self.blink.sync = BlinkSyncModule( - self.blink, self.blink._auth_header, self.blink.urls) + self.blink.sync = MockSyncModule( + self.blink, self.blink._auth_header) self.camera = BlinkCamera(self.config, self.blink.sync) @@ -51,10 +75,9 @@ class TestBlinkFunctions(unittest.TestCase): self.config = {} self.camera = None - @mock.patch('blinkpy.sync_module.http_req') - def test_get_videos(self, req): + def test_get_videos(self): """Test video access.""" - req.return_value = [ + self.blink.sync.return_value = [ { 'camera_name': 'foobar', 'address': '/new/test.mp4', @@ -84,36 +107,38 @@ class TestBlinkFunctions(unittest.TestCase): self.blink.sync.get_cameras() self.assertTrue('foobar' in self.blink.sync.cameras) - @pytest.mark.skip(reason="Need to mock sync class.") - @mock.patch('blinkpy.camera.http_req') - def test_image_refresh(self, req): + def test_image_refresh(self): """Test image refresh function.""" - req.return_value = {'devices': [self.config]} + self.blink.sync.return_value = {'devices': [self.config]} image = self.camera.image_refresh() self.assertEqual(image, 'https://rest.test.{}/test.jpg'.format(BLINK_URL)) - @mock.patch('blinkpy.sync_module.http_req') - def test_video_count(self, req): + def test_video_count(self): """Test video count function.""" - req.return_value = {'count': 1} + self.blink.sync.return_value = {'count': 1} self.assertEqual(self.blink.sync.video_count, 1) - @pytest.mark.skip(reason="Need to mock Blink class, not methods") - @mock.patch('blinkpy.sync_module.http_req') + @mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request') @mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request') def test_refresh(self, vid_req, req): """Test blinkpy refresh function.""" + req.return_value = {'foo': 'bar'} self.blink.sync.cameras = {'foobar': self.camera} - req.return_value = {'devices': [self.config]} + self.blink.sync.return_value = {'devices': [{'foo': 'bar'}]} + # pylint: disable=protected-access + self.blink._last_summary = {'devices': [self.config]} + # pylint: disable=protected-access + self.blink._last_events = {'foo': 'bar'} vid_req.return_value = [ { 'camera_name': 'foobar', 'address': '/new.mp4', - 'thumbnail': '/new' + 'thumbnail': '/new', } ] - self.blink.last_refresh = -1 + self.blink.last_refresh = int(time.time()) + self.blink.refresh_rate = 100000 self.blink.sync.refresh() test_camera = self.blink.sync.cameras['foobar'] self.assertEqual(test_camera.clip, From c3c558e9342317628585120be96bfee0f4816eb8 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Wed, 26 Sep 2018 12:24:45 -0400 Subject: [PATCH 6/9] Fix broken refresh throttle --- blinkpy/blinkpy.py | 1 + 1 file changed, 1 insertion(+) diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index cf4c25b..e603f08 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -175,5 +175,6 @@ class Blink(): if last_refresh is None: last_refresh = 0 if current_time >= (last_refresh + self.refresh_rate): + self.last_refresh = current_time return True return False From 9d5ae9552a970e5b3eeeaff2c55e3abaa5c61303 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Wed, 26 Sep 2018 12:45:29 -0400 Subject: [PATCH 7/9] Check if first init, if so force cache update --- blinkpy/blinkpy.py | 4 +--- blinkpy/sync_module.py | 4 +++- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index e603f08..7a9a4ec 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -162,9 +162,7 @@ class Blink(): def refresh(self, force_cache=False): """Perform a system refresh.""" - if self.last_refresh is None: - force_cache = True - if self.check_if_ok_to_update(): + if self.check_if_ok_to_update() or force_cache: _LOGGER.debug("Attempting refresh of cameras.") self.sync.refresh(force_cache=force_cache) diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 6235c1c..7b28c1d 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -28,6 +28,7 @@ class BlinkSyncModule(): self._all_videos = {} self._summary = None self.record_dates = dict() + self.first_init = True @property def urls(self): @@ -173,7 +174,8 @@ class BlinkSyncModule(): device = BlinkCamera(element, self) self.cameras[device.name] = device self._idlookup[device.id] = device.name - self.blink.refresh() + self.blink.refresh(force_cache=self.first_init) + self.first_init = False def set_links(self): """Set access links and required headers for each camera in system.""" From 451d72594630c9beaa4f649b1e13a1f567cd43fa Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Wed, 26 Sep 2018 13:02:48 -0400 Subject: [PATCH 8/9] Update functionality in README --- README.rst | 38 +++++++++++++++++++++++++++++--------- 1 file changed, 29 insertions(+), 9 deletions(-) diff --git a/README.rst b/README.rst index 52b765d..ef11a44 100644 --- a/README.rst +++ b/README.rst @@ -38,29 +38,49 @@ This library was built with the intention of allowing easy communication with Bl Usage ========= -The simplest way to use this package from a terminal is to call ``Blink.start()`` which will prompt for your Blink username and password and then log you in. Alternatively, you can instantiate the Blink class with a username and password, and call ``Blink.start()`` to login and setup without prompt, as shown below. +The simplest way to use this package from a terminal is to call ``Blink.start()`` which will prompt for your Blink username and password and then log you in. Alternatively, you can instantiate the Blink class with a username and password, and call ``Blink.start()`` to login and setup without prompt, as shown below. In addition, http requests are throttled internally via use of the ``Blink.refresh_rate`` variable, which can be set at initialization and defaults to 30 seconds. .. code:: python - import blinkpy - blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD') + from blinkpy import blinkpy + blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD', refresh_rate=30) blink.start() If you would like to log in without setting up the cameras or system, you can simply call the ``Blink.login()`` function which will prompt for a username and password and then authenticate with the server. This is useful if you want to avoid use of the ``start()`` function which simply acts as a wrapper for more targeted API methods. -The cameras are of a BlinkCamera class, of which the following parameters can be used (the code below creates a Blink object and iterates through each camera found) +Cameras are instantiated as individual ``BlinkCamera`` classes within a ``BlinkSyncModule`` instance. Note: currently the API only supports one sync module, but multiple sync modules are planned to be supported in the future. + +The below code will display cameras and their available attributes: .. code:: python - import blinkpy - + from blinkpy import blinkpy + blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD') blink.start() - - for name, camera in blink.cameras.items(): - print(name) # Name of the camera + + for name, camera in blink.sync.cameras.items(): + print(name) # Name of the camera print(camera.attributes) # Print available attributes of camera +The most recent images and videos can be accessed as a bytes-object via internal variables. These can be updated with calls to ``Blink.refresh()`` but will only make a request if motion has been detected or other changes have been found. This can be overridden with the ``force_cache`` flag, but this should be used for debugging only since it overrides the internal request throttling. + +.. code:: python + + camera = blink.sync.camera['SOME CAMERA NAME'] + blink.refresh(force_cache=True) # force a cache update USE WITH CAUTION + camera.image_from_cache.raw # bytes-like image object (jpg) + camera.video_from_cache.raw # bytes-like video object (mp4) + +The ``blinkpy`` api also allows for saving images and videos to a file and snapping a new picture from the camera remotely: + +.. code:: python + + camera = blink.sync.camera['SOME CAMERA NAME'] + camera.snap_picture() # Take a new picture with the camera + blink.refresh() # Get new information from server + camera.image_to_file('/local/path/for/image.jpg') + camera.video_to_file('/local/path/for/video.mp4') .. |Build Status| image:: https://travis-ci.org/fronzbot/blinkpy.svg?branch=dev :target: https://travis-ci.org/fronzbot/blinkpy From 1b1235210eaa14c7b09e14c67c7a3efa657ae5b2 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Thu, 27 Sep 2018 11:07:31 -0400 Subject: [PATCH 9/9] Split up tests, added cache test --- tests/test_blink_functions.py | 37 ------ ...{test_blink_cameras.py => test_cameras.py} | 84 +++++++++++--- tests/test_sync_module.py | 107 ++++++++++++++++++ 3 files changed, 177 insertions(+), 51 deletions(-) rename tests/{test_blink_cameras.py => test_cameras.py} (63%) create mode 100644 tests/test_sync_module.py diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index 44b77f0..5a32d12 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -75,38 +75,6 @@ class TestBlinkFunctions(unittest.TestCase): self.config = {} self.camera = None - def test_get_videos(self): - """Test video access.""" - self.blink.sync.return_value = [ - { - 'camera_name': 'foobar', - 'address': '/new/test.mp4', - 'thumbnail': '/test/thumb' - } - ] - self.blink.sync.get_videos() - self.assertEqual(self.blink.sync.videos['foobar'][0]['clip'], - '/new/test.mp4') - self.assertEqual(self.blink.sync.videos['foobar'][0]['thumb'], - '/test/thumb') - - @mock.patch('blinkpy.sync_module.BlinkSyncModule.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): - """Test camera extraction.""" - refresh.return_value = True - req.return_value = {'devices': [self.config]} - vid_req.return_value = [ - { - 'camera_name': 'foobar', - 'address': '/new.mp4', - 'thumbnail': '/new' - } - ] - self.blink.sync.get_cameras() - self.assertTrue('foobar' in self.blink.sync.cameras) - def test_image_refresh(self): """Test image refresh function.""" self.blink.sync.return_value = {'devices': [self.config]} @@ -114,11 +82,6 @@ class TestBlinkFunctions(unittest.TestCase): self.assertEqual(image, 'https://rest.test.{}/test.jpg'.format(BLINK_URL)) - def test_video_count(self): - """Test video count function.""" - self.blink.sync.return_value = {'count': 1} - self.assertEqual(self.blink.sync.video_count, 1) - @mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request') @mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request') def test_refresh(self, vid_req, req): diff --git a/tests/test_blink_cameras.py b/tests/test_cameras.py similarity index 63% rename from tests/test_blink_cameras.py rename to tests/test_cameras.py index 068de45..8eb5f7f 100644 --- a/tests/test_blink_cameras.py +++ b/tests/test_cameras.py @@ -50,9 +50,13 @@ class TestBlinkCameraSetup(unittest.TestCase): 'region_id': 'test' } + header = { + 'Host': 'abc.zxc', + 'TOKEN_AUTH': mresp.LOGIN_RESPONSE['authtoken']['authtoken'] + } self.blink.urls = BlinkURLHandler('test') self.blink.network_id = '0000' - self.blink.sync = BlinkSyncModule(self.blink, dict(), self.blink.urls) + self.sync = BlinkSyncModule(self.blink, header, self.blink.urls) def tearDown(self): """Clean up after test.""" @@ -68,12 +72,12 @@ class TestBlinkCameraSetup(unittest.TestCase): """Tests all property set/recall.""" self.blink.urls = BlinkURLHandler('test') - self.blink.sync.cameras = { - 'foobar': BlinkCamera(self.camera_config, self.blink.sync) + self.sync.cameras = { + 'foobar': BlinkCamera(self.camera_config, self.sync) } - for name in self.blink.sync.cameras: - camera = self.blink.sync.cameras[name] + for name in self.sync.cameras: + camera = self.sync.cameras[name] camera.update(self.camera_config, skip_cache=True) self.assertEqual(camera.id, '1111') self.assertEqual(camera.name, 'foobar') @@ -102,8 +106,8 @@ class TestBlinkCameraSetup(unittest.TestCase): camera_config['temp'] = 60 camera_config['battery'] = 0 camera_config['notifications'] = 4 - for name in self.blink.sync.cameras: - camera = self.blink.sync.cameras[name] + for name in self.sync.cameras: + camera = self.sync.cameras[name] camera.update(camera_config, skip_cache=True) self.assertEqual(camera.armed, True) self.assertEqual( @@ -124,9 +128,9 @@ class TestBlinkCameraSetup(unittest.TestCase): def test_camera_case(self): """Tests camera case sensitivity.""" - camera_object = BlinkCamera(self.camera_config, self.blink.sync) - self.blink.sync.cameras['foobar'] = camera_object - self.assertEqual(camera_object, self.blink.sync.cameras['fOoBaR']) + camera_object = BlinkCamera(self.camera_config, self.sync) + self.sync.cameras['foobar'] = camera_object + self.assertEqual(camera_object, self.sync.cameras['fOoBaR']) @mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request', return_value=CAMERA_CFG) @@ -134,12 +138,12 @@ class TestBlinkCameraSetup(unittest.TestCase): """Tests camera attributes.""" self.blink.urls = BlinkURLHandler('test') - self.blink.sync.cameras = { - 'foobar': BlinkCamera(self.camera_config, self.blink.sync) + self.sync.cameras = { + 'foobar': BlinkCamera(self.camera_config, self.sync) } - for name in self.blink.sync.cameras: - camera = self.blink.sync.cameras[name] + for name in self.sync.cameras: + camera = self.sync.cameras[name] camera.update(self.camera_config, skip_cache=True) camera_attr = camera.attributes self.assertEqual(camera_attr['device_id'], '1111') @@ -160,3 +164,55 @@ class TestBlinkCameraSetup(unittest.TestCase): self.assertEqual(camera_attr['network_id'], '0000') self.assertEqual(camera_attr['motion_enabled'], True) 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): + """Tests camera cache.""" + update_vals = { + 'name': 'foobar', + 'active': 'disabled', + 'video': '/clip.mp4', + 'thumbnail': '/image', + 'battery': 3, + 'notifications': 1, + } + self.sync.cameras = { + 'foobar': BlinkCamera(self.camera_config, self.sync) + } + + test_image = 'https://rest.test.immedia-semi.com/image.jpg' + test_clip = 'https://rest.test.immedia-semi.com/clip.mp4' + + for name, camera in self.sync.cameras.items(): + # Check that no cache returns None + self.assertEqual(camera.name, name) + self.assertEqual(camera.image_from_cache, None) + self.assertEqual(camera.video_from_cache, None) + + # Now, call an update with a new thumbnail to see if we update + self.sync.records = [] + # pylint: disable=protected-access + camera.update(update_vals) + self.assertEqual(camera.thumbnail, test_image) + self.assertEqual(camera.image_from_cache.status_code, 200) + + # Now update the clip + self.sync.record_dates = {camera.name: ['7', '1', '4', '3']} + self.assertEqual(camera.last_record, list()) + camera.update(update_vals) + self.assertEqual(camera.clip, test_clip) + self.assertEqual(camera.last_record, list('7')) + # First update should be false + self.assertEqual(camera.motion_detected, False) + self.sync.record_dates[camera.name].append('88') + camera.update(update_vals) + self.assertEqual(camera.last_record, ['88', '7']) + self.assertEqual(camera.motion_detected, True) + self.assertEqual(camera.video_from_cache.status_code, 200) + # Next update shouldn't change records, and motion_dected=False + camera.update(update_vals) + self.assertEqual(camera.motion_detected, False) + self.assertEqual(camera.video_from_cache.status_code, 200) diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py new file mode 100644 index 0000000..1f7587d --- /dev/null +++ b/tests/test_sync_module.py @@ -0,0 +1,107 @@ +"""Tests camera and system functions.""" +import unittest +from unittest import mock + +from blinkpy import blinkpy +from blinkpy.sync_module import BlinkSyncModule +from blinkpy.camera import BlinkCamera + +USERNAME = 'foobar' +PASSWORD = 'deadbeef' + + +class MockSyncModule(BlinkSyncModule): + """Mock http requests from sync module.""" + + def __init__(self, blink, header): + """Create mock sync module instance.""" + super().__init__(blink, header) + self.blink = blink + self.header = header + self.return_value = None + self.return_value2 = None + + def http_get(self, url, stream=False, json=True): + """Mock get request.""" + if stream and self.return_value2 is not None: + return self.return_value2 + return self.return_value + + def http_post(self, url): + """Mock post request.""" + return self.return_value + + +class TestBlinkSyncModule(unittest.TestCase): + """Test BlinkSyncModule functions in blinkpy.""" + + def setUp(self): + """Set up Blink module.""" + self.blink = blinkpy.Blink(username=USERNAME, + password=PASSWORD) + # pylint: disable=protected-access + self.blink._auth_header = { + 'Host': 'test.url.tld', + 'TOKEN_AUTH': 'foobar123' + } + self.blink.urls = blinkpy.BlinkURLHandler('test') + self.config = { + 'device_id': 1111, + 'name': 'foobar', + 'armed': False, + 'active': 'disabled', + 'thumbnail': '/test', + 'video': '/test.mp4', + 'temp': 80, + 'battery': 3, + 'notifications': 2, + 'region_id': 'test', + 'device_type': 'camera' + } + self.blink.sync = MockSyncModule( + self.blink, self.blink._auth_header) + + self.camera = BlinkCamera(self.config, self.blink.sync) + + def tearDown(self): + """Clean up after test.""" + self.blink = None + self.config = {} + self.camera = None + + def test_get_videos(self): + """Test video access.""" + self.blink.sync.return_value = [ + { + 'camera_name': 'foobar', + 'address': '/new/test.mp4', + 'thumbnail': '/test/thumb' + } + ] + self.blink.sync.get_videos() + self.assertEqual(self.blink.sync.videos['foobar'][0]['clip'], + '/new/test.mp4') + self.assertEqual(self.blink.sync.videos['foobar'][0]['thumb'], + '/test/thumb') + + @mock.patch('blinkpy.sync_module.BlinkSyncModule.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): + """Test camera extraction.""" + refresh.return_value = True + req.return_value = {'devices': [self.config]} + vid_req.return_value = [ + { + 'camera_name': 'foobar', + 'address': '/new.mp4', + 'thumbnail': '/new' + } + ] + self.blink.sync.get_cameras() + self.assertTrue('foobar' in self.blink.sync.cameras) + + def test_video_count(self): + """Test video count function.""" + self.blink.sync.return_value = {'count': 1} + self.assertEqual(self.blink.sync.video_count, 1)