diff --git a/app/app.py b/app/app.py index eb7556e..11517a1 100644 --- a/app/app.py +++ b/app/app.py @@ -4,9 +4,9 @@ from datetime import datetime, timedelta from blinkpy import blinkpy -USERNAME = environ.get('USERNAME') -PASSWORD = environ.get('PASSWORD') -TIMEDELTA = timedelta(environ.get('TIMEDELTA', 1)) +USERNAME = environ.get("USERNAME") +PASSWORD = environ.get("PASSWORD") +TIMEDELTA = timedelta(environ.get("TIMEDELTA", 1)) def get_date(): @@ -14,7 +14,7 @@ def get_date(): return (datetime.now() - TIMEDELTA).isoformat() -def download_videos(blink, save_dir='/media'): +def download_videos(blink, save_dir="/media"): """Make request to download videos.""" blink.download_videos(save_dir, since=get_date()) @@ -26,6 +26,6 @@ def start(): return blink -if __name__ == '__main__': +if __name__ == "__main__": BLINK = start() download_videos(BLINK) diff --git a/blinkpy/api.py b/blinkpy/api.py index e4fc2fd..7b1412f 100644 --- a/blinkpy/api.py +++ b/blinkpy/api.py @@ -21,17 +21,23 @@ def request_login(blink, url, username, password, is_retry=False): :param password: Blink password. :param is_retry: Is this part of a re-authorization attempt? """ - headers = { - 'Host': DEFAULT_URL, - 'Content-Type': 'application/json' - } - data = dumps({ - 'email': username, - 'password': password, - 'client_specifier': 'iPhone 9.2 | 2.2 | 222' - }) - return http_req(blink, url=url, headers=headers, data=data, - json_resp=False, reqtype='post', is_retry=is_retry) + headers = {"Host": DEFAULT_URL, "Content-Type": "application/json"} + data = dumps( + { + "email": username, + "password": password, + "client_specifier": "iPhone 9.2 | 2.2 | 222", + } + ) + return http_req( + blink, + url=url, + headers=headers, + data=data, + json_resp=False, + reqtype="post", + is_retry=is_retry, + ) def request_networks(blink): @@ -94,17 +100,16 @@ def request_command_status(blink, network, command_id): :param network: Sync module network id. :param command_id: Command id to check. """ - url = "{}/network/{}/command/{}".format(blink.urls.base_url, - network, - command_id) + url = "{}/network/{}/command/{}".format(blink.urls.base_url, network, command_id) return http_get(blink, url) @Throttle(seconds=MIN_THROTTLE_TIME) def request_homescreen(blink): """Request homescreen info.""" - url = "{}/api/v3/accounts/{}/homescreen".format(blink.urls.base_url, - blink.account_id) + url = "{}/api/v3/accounts/{}/homescreen".format( + blink.urls.base_url, blink.account_id + ) return http_get(blink, url) @@ -129,9 +134,9 @@ def request_new_image(blink, network, camera_id): :param network: Sync module network id. :param camera_id: Camera ID of camera to request new image from. """ - url = "{}/network/{}/camera/{}/thumbnail".format(blink.urls.base_url, - network, - camera_id) + url = "{}/network/{}/camera/{}/thumbnail".format( + blink.urls.base_url, network, camera_id + ) return http_post(blink, url) @@ -144,9 +149,7 @@ def request_new_video(blink, network, camera_id): :param network: Sync module network id. :param camera_id: Camera ID of camera to request new video from. """ - url = "{}/network/{}/camera/{}/clip".format(blink.urls.base_url, - network, - camera_id) + url = "{}/network/{}/camera/{}/clip".format(blink.urls.base_url, network, camera_id) return http_post(blink, url) @@ -167,7 +170,8 @@ def request_videos(blink, time=None, page=0): """ timestamp = get_time(time) url = "{}/api/v1/accounts/{}/media/changed?since={}&page={}".format( - blink.urls.base_url, blink.account_id, timestamp, page) + blink.urls.base_url, blink.account_id, timestamp, page + ) return http_get(blink, url) @@ -190,9 +194,9 @@ def request_camera_info(blink, network, camera_id): :param network: Sync module network id. :param camera_id: Camera ID of camera to request info from. """ - url = "{}/network/{}/camera/{}/config".format(blink.urls.base_url, - network, - camera_id) + url = "{}/network/{}/camera/{}/config".format( + blink.urls.base_url, network, camera_id + ) return http_get(blink, url) @@ -204,9 +208,9 @@ def request_camera_sensors(blink, network, camera_id): :param network: Sync module network id. :param camera_id: Camera ID of camera to request sesnor info from. """ - url = "{}/network/{}/camera/{}/signals".format(blink.urls.base_url, - network, - camera_id) + url = "{}/network/{}/camera/{}/signals".format( + blink.urls.base_url, network, camera_id + ) return http_get(blink, url) @@ -219,9 +223,9 @@ def request_motion_detection_enable(blink, network, camera_id): :param network: Sync module network id. :param camera_id: Camera ID of camera to enable. """ - url = "{}/network/{}/camera/{}/enable".format(blink.urls.base_url, - network, - camera_id) + url = "{}/network/{}/camera/{}/enable".format( + blink.urls.base_url, network, camera_id + ) return http_post(blink, url) @@ -233,9 +237,9 @@ def request_motion_detection_disable(blink, network, camera_id): :param network: Sync module network id. :param camera_id: Camera ID of camera to disable. """ - url = "{}/network/{}/camera/{}/disable".format(blink.urls.base_url, - network, - camera_id) + url = "{}/network/{}/camera/{}/disable".format( + blink.urls.base_url, network, camera_id + ) return http_post(blink, url) @@ -251,9 +255,15 @@ def http_get(blink, url, stream=False, json=True, is_retry=False): if blink.auth_header is None: raise BlinkException(ERROR.AUTH_TOKEN) _LOGGER.debug("Making GET request to %s", url) - return http_req(blink, url=url, headers=blink.auth_header, - reqtype='get', stream=stream, json_resp=json, - is_retry=is_retry) + return http_req( + blink, + url=url, + headers=blink.auth_header, + reqtype="get", + stream=stream, + json_resp=json, + is_retry=is_retry, + ) def http_post(blink, url, is_retry=False): @@ -266,5 +276,6 @@ def http_post(blink, url, is_retry=False): if blink.auth_header is None: raise BlinkException(ERROR.AUTH_TOKEN) _LOGGER.debug("Making POST request to %s", url) - return http_req(blink, url=url, headers=blink.auth_header, - reqtype='post', is_retry=is_retry) + return http_req( + blink, url=url, headers=blink.auth_header, reqtype="post", is_retry=is_retry + ) diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index 8106d16..347abc1 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -28,25 +28,40 @@ from blinkpy import api from blinkpy.sync_module import BlinkSyncModule from blinkpy.helpers import errors as ERROR from blinkpy.helpers.util import ( - create_session, merge_dicts, get_time, BlinkURLHandler, - BlinkAuthenticationException, Throttle) + create_session, + merge_dicts, + get_time, + BlinkURLHandler, + BlinkAuthenticationException, + Throttle, +) from blinkpy.helpers.constants import ( - BLINK_URL, LOGIN_URL, OLD_LOGIN_URL, LOGIN_BACKUP_URL, - DEFAULT_MOTION_INTERVAL, DEFAULT_REFRESH, MIN_THROTTLE_TIME) + BLINK_URL, + LOGIN_URL, + OLD_LOGIN_URL, + LOGIN_BACKUP_URL, + DEFAULT_MOTION_INTERVAL, + DEFAULT_REFRESH, + MIN_THROTTLE_TIME, +) from blinkpy.helpers.constants import __version__ _LOGGER = logging.getLogger(__name__) -class Blink(): +class Blink: """Class to initialize communication.""" - def __init__(self, username=None, password=None, - cred_file=None, - refresh_rate=DEFAULT_REFRESH, - motion_interval=DEFAULT_MOTION_INTERVAL, - legacy_subdomain=False): + def __init__( + self, + username=None, + password=None, + cred_file=None, + refresh_rate=DEFAULT_REFRESH, + motion_interval=DEFAULT_MOTION_INTERVAL, + legacy_subdomain=False, + ): """ Initialize Blink system. @@ -114,10 +129,9 @@ class Blink(): if network_id not in camera_list.keys(): camera_list[network_id] = {} _LOGGER.warning("No cameras found for %s", network_name) - sync_module = BlinkSyncModule(self, - network_name, - network_id, - camera_list[network_id]) + sync_module = BlinkSyncModule( + self, network_name, network_id, camera_list[network_id] + ) sync_module.start() self.sync[network_name] = sync_module self.cameras = self.merge_cameras() @@ -126,18 +140,17 @@ class Blink(): """Prompt user for username and password.""" if self._cred_file is not None and os.path.isfile(self._cred_file): try: - with open(self._cred_file, 'r') as json_file: + with open(self._cred_file, "r") as json_file: creds = json.load(json_file) - self._username = creds['username'] - self._password = creds['password'] + self._username = creds["username"] + self._password = creds["password"] except ValueError: - _LOGGER.error("Improperly formated json file %s.", - self._cred_file, - exc_info=True) + _LOGGER.error( + "Improperly formated json file %s.", self._cred_file, exc_info=True + ) return False except KeyError: - _LOGGER.error("JSON file information incomplete %s.", - exc_info=True) + _LOGGER.error("JSON file information incomplete %s.", exc_info=True) return False else: self._username = input("Username:") @@ -164,11 +177,10 @@ class Blink(): return False self._host = "{}.{}".format(self.region_id, BLINK_URL) - self._token = response['authtoken']['authtoken'] - self.networks = response['networks'] + self._token = response["authtoken"]["authtoken"] + self.networks = response["networks"] - self._auth_header = {'Host': self._host, - 'TOKEN_AUTH': self._token} + self._auth_header = {"Host": self._host, "TOKEN_AUTH": self._token} self.urls = BlinkURLHandler(self.region_id, legacy=self.legacy) return self._auth_header @@ -183,26 +195,23 @@ class Blink(): _LOGGER.info("Attempting login with %s", login_url) - response = api.request_login(self, - login_url, - self._username, - self._password, - is_retry=is_retry) + response = api.request_login( + self, login_url, self._username, self._password, is_retry=is_retry + ) try: if response.status_code != 200: response = self.login_request(is_retry=True) response = response.json() - (self.region_id, self.region), = response['region'].items() + ((self.region_id, self.region),) = response["region"].items() except AttributeError: - _LOGGER.error("Login API endpoint failed with response %s", - response) + _LOGGER.error("Login API endpoint failed with response %s", response) return False except KeyError: _LOGGER.warning("Could not extract region info.") - self.region_id = 'piri' - self.region = 'UNKNOWN' + self.region_id = "piri" + self.region = "UNKNOWN" self._login_url = login_url return response @@ -213,14 +222,14 @@ class Blink(): all_networks = [] network_dict = {} for network, status in self.networks.items(): - if status['onboarded']: - all_networks.append('{}'.format(network)) - network_dict[status['name']] = network + if status["onboarded"]: + all_networks.append("{}".format(network)) + network_dict[status["name"]] = network # For the first onboarded network we find, grab the account id - for resp in response['networks']: - if str(resp['id']) in all_networks: - self.account_id = resp['account_id'] + for resp in response["networks"]: + if str(resp["id"]) in all_networks: + self.account_id = resp["account_id"] break self.network_ids = all_networks @@ -231,11 +240,11 @@ class Blink(): response = api.request_homescreen(self) try: all_cameras = {} - for camera in response['cameras']: - camera_network = str(camera['network_id']) - camera_name = camera['name'] - camera_id = camera['id'] - camera_info = {'name': camera_name, 'id': camera_id} + for camera in response["cameras"]: + camera_network = str(camera["network_id"]) + camera_name = camera["name"] + camera_id = camera["id"] + camera_info = {"name": camera_name, "id": camera_id} if camera_network not in all_cameras: all_cameras[camera_network] = [] @@ -279,8 +288,7 @@ class Blink(): combined = merge_dicts(combined, self.sync[sync].cameras) return combined - def download_videos(self, path, since=None, - camera='all', stop=10, debug=False): + def download_videos(self, path, since=None, camera="all", stop=10, debug=False): """ Download all videos from server since specified time. @@ -310,7 +318,7 @@ class Blink(): response = api.request_videos(self, time=since_epochs, page=page) _LOGGER.debug("Processing page %s", page) try: - result = response['media'] + result = response["media"] if not result: raise IndexError except (KeyError, IndexError): @@ -323,22 +331,20 @@ class Blink(): """Parse downloaded videos.""" for item in result: try: - created_at = item['created_at'] - camera_name = item['device_name'] - is_deleted = item['deleted'] - address = item['media'] + created_at = item["created_at"] + camera_name = item["device_name"] + is_deleted = item["deleted"] + address = item["media"] except KeyError: _LOGGER.info("Missing clip information, skipping...") continue - if camera_name not in camera and 'all' not in camera: + if camera_name not in camera and "all" not in camera: _LOGGER.debug("Skipping videos for %s.", camera_name) continue if is_deleted: - _LOGGER.debug("%s: %s is marked as deleted.", - camera_name, - address) + _LOGGER.debug("%s: %s is marked as deleted.", camera_name, address) continue clip_address = "{}{}".format(self.urls.base_url, address) @@ -351,13 +357,14 @@ class Blink(): _LOGGER.info("%s already exists, skipping...", filename) continue - response = api.http_get(self, url=clip_address, - stream=True, json=False) - with open(filename, 'wb') as vidfile: + response = api.http_get(self, url=clip_address, stream=True, json=False) + with open(filename, "wb") as vidfile: copyfileobj(response.raw, vidfile) _LOGGER.info("Downloaded video to %s", filename) else: - print(("Camera: {}, Timestamp: {}, " - "Address: {}, Filename: {}").format( - camera_name, created_at, address, filename)) + print( + ("Camera: {}, Timestamp: {}, " "Address: {}, Filename: {}").format( + camera_name, created_at, address, filename + ) + ) diff --git a/blinkpy/camera.py b/blinkpy/camera.py index efe9c7b..a2f05be 100644 --- a/blinkpy/camera.py +++ b/blinkpy/camera.py @@ -7,7 +7,7 @@ from blinkpy import api _LOGGER = logging.getLogger(__name__) -class BlinkCamera(): +class BlinkCamera: """Class to initialize individual camera.""" def __init__(self, sync): @@ -34,22 +34,22 @@ class BlinkCamera(): def attributes(self): """Return dictionary of all camera attributes.""" attributes = { - 'name': self.name, - 'camera_id': self.camera_id, - 'serial': self.serial, - 'temperature': self.temperature, - 'temperature_c': self.temperature_c, - 'temperature_calibrated': self.temperature_calibrated, - 'battery': self.battery, - 'battery_voltage': self.battery_voltage, - 'thumbnail': self.thumbnail, - 'video': self.clip, - 'motion_enabled': self.motion_enabled, - 'motion_detected': self.motion_detected, - 'wifi_strength': self.wifi_strength, - 'network_id': self.sync.network_id, - 'sync_module': self.sync.name, - 'last_record': self.last_record + "name": self.name, + "camera_id": self.camera_id, + "serial": self.serial, + "temperature": self.temperature, + "temperature_c": self.temperature_c, + "temperature_calibrated": self.temperature_calibrated, + "battery": self.battery, + "battery_voltage": self.battery_voltage, + "thumbnail": self.thumbnail, + "video": self.clip, + "motion_enabled": self.motion_enabled, + "motion_detected": self.motion_detected, + "wifi_strength": self.wifi_strength, + "network_id": self.sync.network_id, + "sync_module": self.sync.name, + "last_record": self.last_record, } return attributes @@ -79,39 +79,37 @@ class BlinkCamera(): def snap_picture(self): """Take a picture with camera to create a new thumbnail.""" - return api.request_new_image(self.sync.blink, - self.network_id, - self.camera_id) + return api.request_new_image(self.sync.blink, self.network_id, self.camera_id) def set_motion_detect(self, enable): """Set motion detection.""" if enable: - return api.request_motion_detection_enable(self.sync.blink, - self.network_id, - self.camera_id) - return api.request_motion_detection_disable(self.sync.blink, - self.network_id, - self.camera_id) + return api.request_motion_detection_enable( + self.sync.blink, self.network_id, self.camera_id + ) + return api.request_motion_detection_disable( + self.sync.blink, self.network_id, self.camera_id + ) def update(self, config, force_cache=False, **kwargs): """Update camera info.""" # force = kwargs.pop('force', False) - self.name = config['name'] - self.camera_id = str(config['id']) - self.network_id = str(config['network_id']) - self.serial = config['serial'] - self.motion_enabled = config['enabled'] - self.battery_voltage = config['battery_voltage'] - self.battery_state = config['battery_state'] - self.temperature = config['temperature'] - self.wifi_strength = config['wifi_strength'] + self.name = config["name"] + self.camera_id = str(config["id"]) + self.network_id = str(config["network_id"]) + self.serial = config["serial"] + self.motion_enabled = config["enabled"] + self.battery_voltage = config["battery_voltage"] + self.battery_state = config["battery_state"] + self.temperature = config["temperature"] + self.wifi_strength = config["wifi_strength"] # Retrieve calibrated temperature from special endpoint - resp = api.request_camera_sensors(self.sync.blink, - self.network_id, - self.camera_id) + resp = api.request_camera_sensors( + self.sync.blink, self.network_id, self.camera_id + ) try: - self.temperature_calibrated = resp['temp'] + self.temperature_calibrated = resp["temp"] except KeyError: self.temperature_calibrated = self.temperature _LOGGER.warning("Could not retrieve calibrated temperature.") @@ -121,16 +119,15 @@ class BlinkCamera(): # otherwise set it to None and log an error new_thumbnail = None thumb_addr = None - if config['thumbnail']: - thumb_addr = config['thumbnail'] + if config["thumbnail"]: + thumb_addr = config["thumbnail"] else: - _LOGGER.warning("Could not find thumbnail for camera %s", - self.name, - exc_info=True) + _LOGGER.warning( + "Could not find thumbnail for camera %s", self.name, exc_info=True + ) if thumb_addr is not None: - new_thumbnail = "{}{}.jpg".format(self.sync.urls.base_url, - thumb_addr) + new_thumbnail = "{}{}.jpg".format(self.sync.urls.base_url, thumb_addr) try: self.motion_detected = self.sync.motion[self.name] @@ -139,10 +136,9 @@ class BlinkCamera(): clip_addr = None if self.name in self.sync.last_record: - clip_addr = self.sync.last_record[self.name]['clip'] - self.last_record = self.sync.last_record[self.name]['time'] - self.clip = "{}{}".format(self.sync.urls.base_url, - clip_addr) + clip_addr = self.sync.last_record[self.name]["clip"] + self.last_record = self.sync.last_record[self.name]["time"] + self.clip = "{}{}".format(self.sync.urls.base_url, clip_addr) # If the thumbnail or clip have changed, update the cache update_cached_image = False @@ -155,15 +151,13 @@ class BlinkCamera(): update_cached_video = True if new_thumbnail is not None and (update_cached_image or force_cache): - self._cached_image = api.http_get(self.sync.blink, - url=self.thumbnail, - stream=True, - json=False) + self._cached_image = api.http_get( + self.sync.blink, url=self.thumbnail, stream=True, json=False + ) if clip_addr is not None and (update_cached_video or force_cache): - self._cached_video = api.http_get(self.sync.blink, - url=self.clip, - stream=True, - json=False) + self._cached_video = api.http_get( + self.sync.blink, url=self.clip, stream=True, json=False + ) def image_to_file(self, path): """ @@ -174,11 +168,12 @@ class BlinkCamera(): _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: + with open(path, "wb") as imgfile: copyfileobj(response.raw, imgfile) else: - _LOGGER.error("Cannot write image to file, response %s", - response.status_code) + _LOGGER.error( + "Cannot write image to file, response %s", response.status_code + ) def video_to_file(self, path): """Write video to file. @@ -188,8 +183,7 @@ class BlinkCamera(): _LOGGER.debug("Writing video from %s to %s", self.name, path) response = self._cached_video if response is None: - _LOGGER.error("No saved video exist for %s.", - self.name) + _LOGGER.error("No saved video exist for %s.", self.name) return - with open(path, 'wb') as vidfile: + with open(path, "wb") as vidfile: copyfileobj(response.raw, vidfile) diff --git a/blinkpy/helpers/constants.py b/blinkpy/helpers/constants.py index 477fe8b..6cdbaba 100644 --- a/blinkpy/helpers/constants.py +++ b/blinkpy/helpers/constants.py @@ -4,62 +4,63 @@ import os MAJOR_VERSION = 0 MINOR_VERSION = 15 -PATCH_VERSION = '0-rc.0' +PATCH_VERSION = "0-rc.0" -__version__ = '{}.{}.{}'.format(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION) +__version__ = "{}.{}.{}".format(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION) REQUIRED_PYTHON_VER = (3, 5, 3) -PROJECT_NAME = 'blinkpy' -PROJECT_PACKAGE_NAME = 'blinkpy' -PROJECT_LICENSE = 'MIT' -PROJECT_AUTHOR = 'Kevin Fronczak' -PROJECT_COPYRIGHT = ' 2017, {}'.format(PROJECT_AUTHOR) -PROJECT_URL = 'https://github.com/fronzbot/blinkpy' -PROJECT_EMAIL = 'kfronczak@gmail.com' -PROJECT_DESCRIPTION = ('A Blink camera Python library ' - 'running on Python 3.') -PROJECT_LONG_DESCRIPTION = ('blinkpy is an open-source ' - 'unofficial API for the Blink Camera ' - 'system with the intention for easy ' - 'integration into various home ' - 'automation platforms.') -if os.path.exists('README.rst'): - PROJECT_LONG_DESCRIPTION = open('README.rst').read() +PROJECT_NAME = "blinkpy" +PROJECT_PACKAGE_NAME = "blinkpy" +PROJECT_LICENSE = "MIT" +PROJECT_AUTHOR = "Kevin Fronczak" +PROJECT_COPYRIGHT = " 2017, {}".format(PROJECT_AUTHOR) +PROJECT_URL = "https://github.com/fronzbot/blinkpy" +PROJECT_EMAIL = "kfronczak@gmail.com" +PROJECT_DESCRIPTION = "A Blink camera Python library " "running on Python 3." +PROJECT_LONG_DESCRIPTION = ( + "blinkpy is an open-source " + "unofficial API for the Blink Camera " + "system with the intention for easy " + "integration into various home " + "automation platforms." +) +if os.path.exists("README.rst"): + PROJECT_LONG_DESCRIPTION = open("README.rst").read() PROJECT_CLASSIFIERS = [ - 'Intended Audience :: Developers', - 'License :: OSI Approved :: MIT License', - 'Operating System :: OS Independent', - 'Programming Language :: Python :: 3.6', - 'Programming Language :: Python :: 3.7', - 'Programming Language :: Python :: 3.8', - 'Topic :: Home Automation' + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3.6", + "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Topic :: Home Automation", ] -PROJECT_GITHUB_USERNAME = 'fronzbot' -PROJECT_GITHUB_REPOSITORY = 'blinkpy' +PROJECT_GITHUB_USERNAME = "fronzbot" +PROJECT_GITHUB_REPOSITORY = "blinkpy" -PYPI_URL = 'https://pypi.python.org/pypi/{}'.format(PROJECT_PACKAGE_NAME) +PYPI_URL = "https://pypi.python.org/pypi/{}".format(PROJECT_PACKAGE_NAME) -''' +""" URLS -''' -BLINK_URL = 'immedia-semi.com' -DEFAULT_URL = "{}.{}".format('rest-prod', BLINK_URL) +""" +BLINK_URL = "immedia-semi.com" +DEFAULT_URL = "{}.{}".format("rest-prod", BLINK_URL) BASE_URL = "https://{}".format(DEFAULT_URL) LOGIN_URL = "{}/api/v2/login".format(BASE_URL) OLD_LOGIN_URL = "{}/login".format(BASE_URL) -LOGIN_BACKUP_URL = "https://{}.{}/login".format('rest-piri', BLINK_URL) +LOGIN_BACKUP_URL = "https://{}.{}/login".format("rest-piri", BLINK_URL) -''' +""" Dictionaries -''' -ONLINE = {'online': True, 'offline': False} +""" +ONLINE = {"online": True, "offline": False} -''' +""" OTHER -''' -TIMESTAMP_FORMAT = '%Y-%m-%dT%H:%M:%S%z' +""" +TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S%z" DEFAULT_MOTION_INTERVAL = 1 DEFAULT_REFRESH = 30 diff --git a/blinkpy/helpers/errors.py b/blinkpy/helpers/errors.py index b81404b..3f39086 100644 --- a/blinkpy/helpers/errors.py +++ b/blinkpy/helpers/errors.py @@ -4,11 +4,11 @@ USERNAME = (0, "Username must be a string") PASSWORD = (1, "Password must be a string") AUTHENTICATE = ( 2, - "Cannot authenticate since either password or username has not been set" + "Cannot authenticate since either password or username has not been set", ) AUTH_TOKEN = ( 3, - "Authentication header incorrect. Are you sure you received your token?" + "Authentication header incorrect. Are you sure you received your token?", ) REQUEST = (4, "Cannot perform request (get/post type incorrect)") diff --git a/blinkpy/helpers/util.py b/blinkpy/helpers/util.py index cff1187..3d7637d 100644 --- a/blinkpy/helpers/util.py +++ b/blinkpy/helpers/util.py @@ -18,8 +18,7 @@ def time_to_seconds(timestamp): try: dtime = dateutil.parser.isoparse(timestamp) except ValueError: - _LOGGER.error("Incorrect timestamp format for conversion: %s.", - timestamp) + _LOGGER.error("Incorrect timestamp format for conversion: %s.", timestamp) return False return timegm(dtime.timetuple()) @@ -35,8 +34,10 @@ def merge_dicts(dict_a, dict_b): """Merge two dictionaries into one.""" duplicates = [val for val in dict_a if val in dict_b] if duplicates: - _LOGGER.warning(("Duplicates found during merge: %s. " - "Renaming is recommended."), duplicates) + _LOGGER.warning( + ("Duplicates found during merge: %s. " "Renaming is recommended."), + duplicates, + ) return {**dict_a, **dict_b} @@ -59,8 +60,16 @@ def attempt_reauthorization(blink): return headers -def http_req(blink, url='http://example.com', data=None, headers=None, - reqtype='get', stream=False, json_resp=True, is_retry=False): +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. @@ -73,10 +82,10 @@ def http_req(blink, url='http://example.com', data=None, headers=None, :param json_resp: Return JSON response? TRUE/False :param is_retry: Is this a retry attempt? True/FALSE """ - if reqtype == 'post': - req = Request('POST', url, headers=headers, data=data) - elif reqtype == 'get': - req = Request('GET', url, headers=headers) + if reqtype == "post": + req = Request("POST", url, headers=headers, data=data) + elif reqtype == "get": + req = Request("GET", url, headers=headers) else: _LOGGER.error("Invalid request type: %s", reqtype) raise BlinkException(ERROR.REQUEST) @@ -85,10 +94,10 @@ def http_req(blink, url='http://example.com', data=None, headers=None, try: response = blink.session.send(prepped, stream=stream) - if json_resp and 'code' in response.json(): + if json_resp and "code" in response.json(): resp_dict = response.json() - code = resp_dict['code'] - message = resp_dict['message'] + code = resp_dict["code"] + message = resp_dict["message"] if is_retry and code in ERROR.BLINK_ERRORS: _LOGGER.error("Cannot obtain new token for server auth.") return None @@ -96,20 +105,33 @@ def http_req(blink, url='http://example.com', data=None, headers=None, headers = attempt_reauthorization(blink) if not headers: raise exceptions.ConnectionError - return http_req(blink, url=url, data=data, headers=headers, - reqtype=reqtype, stream=stream, - json_resp=json_resp, is_retry=True) + return http_req( + blink, + url=url, + data=data, + headers=headers, + reqtype=reqtype, + stream=stream, + json_resp=json_resp, + is_retry=True, + ) _LOGGER.warning("Response from server: %s - %s", code, message) except (exceptions.ConnectionError, exceptions.Timeout): _LOGGER.info("Cannot connect to server with url %s.", url) if not is_retry: 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) - _LOGGER.error("Endpoint %s failed. Possible issue with Blink servers.", - url) + return http_req( + blink, + url=url, + data=data, + headers=headers, + reqtype=reqtype, + stream=stream, + json_resp=json_resp, + is_retry=True, + ) + _LOGGER.error("Endpoint %s failed. Possible issue with Blink servers.", url) return None if json_resp: @@ -132,14 +154,14 @@ class BlinkAuthenticationException(BlinkException): """Class to throw authentication exception.""" -class BlinkURLHandler(): +class BlinkURLHandler: """Class that handles Blink URLS.""" def __init__(self, region_id, legacy=False): """Initialize the urls.""" - self.subdomain = 'rest-{}'.format(region_id) + self.subdomain = "rest-{}".format(region_id) if legacy: - self.subdomain = 'rest.{}'.format(region_id) + self.subdomain = "rest.{}".format(region_id) self.base_url = "https://{}.{}".format(self.subdomain, BLINK_URL) self.home_url = "{}/homescreen".format(self.base_url) self.event_url = "{}/events/network".format(self.base_url) @@ -149,7 +171,7 @@ class BlinkURLHandler(): _LOGGER.debug("Setting base url to %s.", self.base_url) -class Throttle(): +class Throttle: """Class for throttling api calls.""" def __init__(self, seconds=10): @@ -159,6 +181,7 @@ class Throttle(): def __call__(self, method): """Throttle caller method.""" + def throttle_method(): """Call when method is throttled.""" return None @@ -166,7 +189,7 @@ class Throttle(): @wraps(method) def wrapper(*args, **kwargs): """Wrap that checks for throttling.""" - force = kwargs.pop('force', False) + force = kwargs.pop("force", False) now = int(time.time()) last_call_delta = now - self.last_call if force or last_call_delta > self.throttle_time: diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index ee465c7..ebed95a 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -11,7 +11,7 @@ from blinkpy.helpers.constants import ONLINE _LOGGER = logging.getLogger(__name__) -class BlinkSyncModule(): +class BlinkSyncModule: """Class to initialize sync module.""" def __init__(self, blink, network_name, network_id, camera_list): @@ -43,13 +43,13 @@ class BlinkSyncModule(): def attributes(self): """Return sync attributes.""" attr = { - 'name': self.name, - 'id': self.sync_id, - 'network_id': self.network_id, - 'serial': self.serial, - 'status': self.status, - 'region': self.region, - 'region_id': self.region_id, + "name": self.name, + "id": self.sync_id, + "network_id": self.network_id, + "serial": self.serial, + "status": self.status, + "region": self.region, + "region_id": self.region_id, } return attr @@ -67,7 +67,7 @@ class BlinkSyncModule(): def arm(self): """Return status of sync module: armed/disarmed.""" try: - return self.network_info['network']['armed'] + return self.network_info["network"]["armed"] except (KeyError, TypeError): return None @@ -81,85 +81,74 @@ class BlinkSyncModule(): def start(self): """Initialize the system.""" - response = api.request_syncmodule(self.blink, - self.network_id) + response = api.request_syncmodule(self.blink, self.network_id) try: - self.summary = response['syncmodule'] - self.network_id = self.summary['network_id'] + self.summary = response["syncmodule"] + self.network_id = self.summary["network_id"] except (TypeError, KeyError): - _LOGGER.error(("Could not retrieve sync module information " - "with response: %s"), response, exc_info=True) + _LOGGER.error( + ("Could not retrieve sync module information " "with response: %s"), + response, + exc_info=True, + ) return False try: - self.sync_id = self.summary['id'] - self.serial = self.summary['serial'] - self.status = self.summary['status'] + self.sync_id = self.summary["id"] + self.serial = self.summary["serial"] + self.status = self.summary["status"] except KeyError: - _LOGGER.error("Could not extract some sync module info: %s", - response, - exc_info=True) + _LOGGER.error( + "Could not extract some sync module info: %s", response, exc_info=True + ) - self.network_info = api.request_network_status(self.blink, - self.network_id) + self.network_info = api.request_network_status(self.blink, self.network_id) self.check_new_videos() try: for camera_config in self.camera_list: - if 'name' not in camera_config: + if "name" not in camera_config: break - name = camera_config['name'] + name = camera_config["name"] self.cameras[name] = BlinkCamera(self) self.motion[name] = False - camera_info = self.get_camera_info(camera_config['id']) - self.cameras[name].update(camera_info, - force_cache=True, - force=True) + camera_info = self.get_camera_info(camera_config["id"]) + self.cameras[name].update(camera_info, force_cache=True, force=True) except KeyError: - _LOGGER.error("Could not create cameras instances for %s", - self.name, - exc_info=True) + _LOGGER.error( + "Could not create cameras instances for %s", self.name, exc_info=True + ) return False return True def get_events(self, **kwargs): """Retrieve events from server.""" - force = kwargs.pop('force', False) - response = api.request_sync_events(self.blink, - self.network_id, - force=force) + force = kwargs.pop("force", False) + response = api.request_sync_events(self.blink, self.network_id, force=force) try: - return response['event'] + return response["event"] except (TypeError, KeyError): - _LOGGER.error("Could not extract events: %s", - response, - exc_info=True) + _LOGGER.error("Could not extract events: %s", response, exc_info=True) return False def get_camera_info(self, camera_id): """Retrieve camera information.""" - response = api.request_camera_info(self.blink, - self.network_id, - camera_id) + response = api.request_camera_info(self.blink, self.network_id, camera_id) try: - return response['camera'][0] + return response["camera"][0] except (TypeError, KeyError): - _LOGGER.error("Could not extract camera info: %s", - response, - exc_info=True) + _LOGGER.error("Could not extract camera info: %s", response, exc_info=True) return [] def refresh(self, force_cache=False): """Get all blink cameras and pulls their most recent status.""" - self.network_info = api.request_network_status(self.blink, - self.network_id) + self.network_info = api.request_network_status(self.blink, self.network_id) self.check_new_videos() for camera_name in self.cameras.keys(): camera_id = self.cameras[camera_name].camera_id camera_info = self.get_camera_info(camera_id) - self.cameras[camera_name].update(camera_info, - force_cache=force_cache) + self.cameras[camera_name].update(camera_info, force_cache=force_cache) def check_new_videos(self): """Check if new videos since last refresh.""" @@ -170,27 +159,25 @@ class BlinkSyncModule(): # No need to check for motion. return False - resp = api.request_videos(self.blink, - time=interval, - page=1) + resp = api.request_videos(self.blink, time=interval, page=1) for camera in self.cameras.keys(): self.motion[camera] = False try: - info = resp['media'] + info = resp["media"] except (KeyError, TypeError): _LOGGER.warning("Could not check for motion. Response: %s", resp) return False for entry in info: try: - name = entry['device_name'] - clip = entry['media'] - timestamp = entry['created_at'] + name = entry["device_name"] + clip = entry["media"] + timestamp = entry["created_at"] if self.check_new_video_time(timestamp): self.motion[name] = True and self.arm - self.last_record[name] = {'clip': clip, 'time': timestamp} + self.last_record[name] = {"clip": clip, "time": timestamp} except KeyError: _LOGGER.debug("No new videos since last refresh.") diff --git a/pylintrc b/pylintrc index 85dfdd6..8b9f17b 100644 --- a/pylintrc +++ b/pylintrc @@ -12,9 +12,12 @@ reports=no # unexpected-keyword-arg - doesn't allow for use of **kwargs, which is dumb disable= + format, + bad-continuation, locally-disabled, unused-argument, duplicate-code, + implicit-str-concat, too-many-arguments, too-many-branches, too-many-instance-attributes, diff --git a/requirements_test.txt b/requirements_test.txt index 92a1557..510ce65 100644 --- a/requirements_test.txt +++ b/requirements_test.txt @@ -1,3 +1,4 @@ +black==19.10b0 coverage==5.1 flake8==3.7.9 flake8-docstrings==1.5.0 diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..72b7313 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,18 @@ +[flake8] +exclude = .venv,.git,.tox,docs,venv,bin,lib,deps,build +doctests = True +# To work with Black +max-line-length = 88 + +# E501: line too long +# W503: Line break occurred before a binary operator +# E203: Whitespace before ':' +# D202 No blank lines allowed after function docstring +# W504 line break after binary operator + +ignore = + E501, + W503, + E203, + D202, + W504, diff --git a/tests/mock_responses.py b/tests/mock_responses.py index 90ae81c..14ec206 100644 --- a/tests/mock_responses.py +++ b/tests/mock_responses.py @@ -4,11 +4,9 @@ from blinkpy.helpers.util import BlinkURLHandler import blinkpy.helpers.constants as const LOGIN_RESPONSE = { - 'region': {'mock': 'Test'}, - 'networks': { - '1234': {'name': 'test', 'onboarded': True} - }, - 'authtoken': {'authtoken': 'foobar123', 'message': 'auth'} + "region": {"mock": "Test"}, + "networks": {"1234": {"name": "test", "onboarded": True}}, + "authtoken": {"authtoken": "foobar123", "message": "auth"}, } @@ -37,29 +35,29 @@ def mocked_session_send(*args, **kwargs): 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} + 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'} + elif url == "use_bad_response": + response = {"foo": "bar"} status = 200 - elif url == 'reauth': - response = {'message': 'REAUTH', 'code': 777} + elif url == "reauth": + response = {"message": "REAUTH", "code": 777} status = 777 else: - response = {'test': 'foo'} + response = {"test": "foo"} status = 200 - elif method == 'POST': + 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} + elif url == "http://wrong.url/" or url is None: + response = {"message": "Error", "code": 404} status = 404 else: - response = {'message': 'foo', 'code': 200} + response = {"message": "foo", "code": 200} status = 200 return MockResponse(response, status) diff --git a/tests/test_api.py b/tests/test_api.py index e5f7399..f22daaa 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -21,22 +21,26 @@ class TestBlinkAPI(unittest.TestCase): """Tear down blink module.""" self.blink = None - @mock.patch('blinkpy.blinkpy.Blink.get_auth_token') + @mock.patch("blinkpy.blinkpy.Blink.get_auth_token") def test_http_req_connect_error(self, mock_auth): """Test http_get error condition.""" - mock_auth.return_value = {'foo': 'bar'} - firstlog = ("INFO:blinkpy.helpers.util:" - "Cannot connect to server with url {}.").format( - 'http://notreal.fake') - nextlog = ("INFO:blinkpy.helpers.util:" - "Auth token expired, attempting reauthorization.") - lastlog = ("ERROR:blinkpy.helpers.util:" - "Endpoint {} failed. Possible issue with " - "Blink servers.").format('http://notreal.fake') + mock_auth.return_value = {"foo": "bar"} + firstlog = ( + "INFO:blinkpy.helpers.util:" "Cannot connect to server with url {}." + ).format("http://notreal.fake") + nextlog = ( + "INFO:blinkpy.helpers.util:" + "Auth token expired, attempting reauthorization." + ) + lastlog = ( + "ERROR:blinkpy.helpers.util:" + "Endpoint {} failed. Possible issue with " + "Blink servers." + ).format("http://notreal.fake") expected = [firstlog, nextlog, firstlog, lastlog] with self.assertLogs() as getlog: - api.http_get(self.blink, 'http://notreal.fake') + api.http_get(self.blink, "http://notreal.fake") with self.assertLogs() as postlog: - api.http_post(self.blink, 'http://notreal.fake') + api.http_post(self.blink, "http://notreal.fake") self.assertEqual(getlog.output, expected) self.assertEqual(postlog.output, expected) diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index df4079f..775e1b5 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -8,8 +8,8 @@ from blinkpy.sync_module import BlinkSyncModule from blinkpy.helpers.util import create_session, get_time import tests.mock_responses as mresp -USERNAME = 'foobar' -PASSWORD = 'deadbeef' +USERNAME = "foobar" +PASSWORD = "deadbeef" class MockSyncModule(BlinkSyncModule): @@ -34,69 +34,56 @@ class MockSyncModule(BlinkSyncModule): return self.return_value -@mock.patch('blinkpy.helpers.util.Session.send', - side_effect=mresp.mocked_session_send) +@mock.patch("blinkpy.helpers.util.Session.send", side_effect=mresp.mocked_session_send) class TestBlinkFunctions(unittest.TestCase): """Test Blink and BlinkCamera functions in blinkpy.""" def setUp(self): """Set up Blink module.""" - self.blink = blinkpy.Blink(username=USERNAME, - password=PASSWORD) + self.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.blink._auth_header = {"Host": "test.url.tld", "TOKEN_AUTH": "foobar123"} + self.blink.urls = blinkpy.BlinkURLHandler("test") self.blink.session = create_session() def tearDown(self): """Clean up after test.""" self.blink = None - @mock.patch('blinkpy.blinkpy.api.request_login') + @mock.patch("blinkpy.blinkpy.api.request_login") def test_backup_url(self, req, mock_sess): """Test backup login method.""" json_resp = { - 'authtoken': {'authtoken': 'foobar123'}, - 'networks': {'1234': {'name': 'foobar', 'onboarded': True}} + "authtoken": {"authtoken": "foobar123"}, + "networks": {"1234": {"name": "foobar", "onboarded": True}}, } bad_req = mresp.MockResponse({}, 404) new_req = mresp.MockResponse(json_resp, 200) - req.side_effect = [ - bad_req, - bad_req, - new_req - ] - self.blink.login_urls = ['test1', 'test2', 'test3'] + req.side_effect = [bad_req, bad_req, new_req] + self.blink.login_urls = ["test1", "test2", "test3"] self.blink.login_request() # pylint: disable=protected-access - self.assertEqual(self.blink._login_url, 'test3') + self.assertEqual(self.blink._login_url, "test3") - req.side_effect = [ - bad_req, - new_req, - bad_req - ] - self.blink.login_urls = ['test1', 'test2', 'test3'] + req.side_effect = [bad_req, new_req, bad_req] + self.blink.login_urls = ["test1", "test2", "test3"] self.blink.login_request() # pylint: disable=protected-access - self.assertEqual(self.blink._login_url, 'test2') + self.assertEqual(self.blink._login_url, "test2") def test_merge_cameras(self, mock_sess): """Test merge camera functionality.""" - first_dict = {'foo': 'bar', 'test': 123} - next_dict = {'foobar': 456, 'bar': 'foo'} - self.blink.sync['foo'] = BlinkSyncModule(self.blink, 'foo', 1, []) - self.blink.sync['bar'] = BlinkSyncModule(self.blink, 'bar', 2, []) - self.blink.sync['foo'].cameras = first_dict - self.blink.sync['bar'].cameras = next_dict + first_dict = {"foo": "bar", "test": 123} + next_dict = {"foobar": 456, "bar": "foo"} + self.blink.sync["foo"] = BlinkSyncModule(self.blink, "foo", 1, []) + self.blink.sync["bar"] = BlinkSyncModule(self.blink, "bar", 2, []) + self.blink.sync["foo"].cameras = first_dict + self.blink.sync["bar"].cameras = next_dict result = self.blink.merge_cameras() - expected = {'foo': 'bar', 'test': 123, 'foobar': 456, 'bar': 'foo'} + expected = {"foo": "bar", "test": 123, "foobar": 456, "bar": "foo"} self.assertEqual(expected, result) - @mock.patch('blinkpy.blinkpy.api.request_videos') + @mock.patch("blinkpy.blinkpy.api.request_videos") def test_download_video_exit(self, mock_req, mock_sess): """Test we exit method when provided bad response.""" blink = blinkpy.Blink() @@ -106,63 +93,60 @@ class TestBlinkFunctions(unittest.TestCase): mock_req.return_value = {} formatted_date = get_time(blink.last_refresh) expected_log = [ - "INFO:blinkpy.blinkpy:Retrieving videos since {}".format( - formatted_date), + "INFO:blinkpy.blinkpy:Retrieving videos since {}".format(formatted_date), "DEBUG:blinkpy.blinkpy:Processing page 1", - "INFO:blinkpy.blinkpy:No videos found on page 1. Exiting." + "INFO:blinkpy.blinkpy:No videos found on page 1. Exiting.", ] with self.assertLogs() as dl_log: - blink.download_videos('/tmp') + blink.download_videos("/tmp") self.assertEqual(dl_log.output, expected_log) - @mock.patch('blinkpy.blinkpy.api.request_videos') + @mock.patch("blinkpy.blinkpy.api.request_videos") def test_parse_downloaded_items(self, mock_req, mock_sess): """Test ability to parse downloaded items list.""" blink = blinkpy.Blink() # pylint: disable=protected-access blinkpy._LOGGER.setLevel(logging.DEBUG) generic_entry = { - 'created_at': '1970', - 'device_name': 'foo', - 'deleted': True, - 'media': '/bar.mp4' + "created_at": "1970", + "device_name": "foo", + "deleted": True, + "media": "/bar.mp4", } result = [generic_entry] - mock_req.return_value = {'media': result} + mock_req.return_value = {"media": result} blink.last_refresh = 0 formatted_date = get_time(blink.last_refresh) expected_log = [ - "INFO:blinkpy.blinkpy:Retrieving videos since {}".format( - formatted_date), + "INFO:blinkpy.blinkpy:Retrieving videos since {}".format(formatted_date), "DEBUG:blinkpy.blinkpy:Processing page 1", - "DEBUG:blinkpy.blinkpy:foo: /bar.mp4 is marked as deleted." + "DEBUG:blinkpy.blinkpy:foo: /bar.mp4 is marked as deleted.", ] with self.assertLogs() as dl_log: - blink.download_videos('/tmp', stop=2) + blink.download_videos("/tmp", stop=2) self.assertEqual(dl_log.output, expected_log) - @mock.patch('blinkpy.blinkpy.api.request_videos') + @mock.patch("blinkpy.blinkpy.api.request_videos") def test_parse_camera_not_in_list(self, mock_req, mock_sess): """Test ability to parse downloaded items list.""" blink = blinkpy.Blink() # pylint: disable=protected-access blinkpy._LOGGER.setLevel(logging.DEBUG) generic_entry = { - 'created_at': '1970', - 'device_name': 'foo', - 'deleted': True, - 'media': '/bar.mp4' + "created_at": "1970", + "device_name": "foo", + "deleted": True, + "media": "/bar.mp4", } result = [generic_entry] - mock_req.return_value = {'media': result} + mock_req.return_value = {"media": result} blink.last_refresh = 0 formatted_date = get_time(blink.last_refresh) expected_log = [ - "INFO:blinkpy.blinkpy:Retrieving videos since {}".format( - formatted_date), + "INFO:blinkpy.blinkpy:Retrieving videos since {}".format(formatted_date), "DEBUG:blinkpy.blinkpy:Processing page 1", - "DEBUG:blinkpy.blinkpy:Skipping videos for foo." + "DEBUG:blinkpy.blinkpy:Skipping videos for foo.", ] with self.assertLogs() as dl_log: - blink.download_videos('/tmp', camera='bar', stop=2) + blink.download_videos("/tmp", camera="bar", stop=2) self.assertEqual(dl_log.output, expected_log) diff --git a/tests/test_blinkpy.py b/tests/test_blinkpy.py index 686641c..c76d43e 100644 --- a/tests/test_blinkpy.py +++ b/tests/test_blinkpy.py @@ -12,30 +12,29 @@ from blinkpy import api from blinkpy.blinkpy import Blink from blinkpy.sync_module import BlinkSyncModule from blinkpy.helpers.util import ( - http_req, create_session, BlinkAuthenticationException, - BlinkException, BlinkURLHandler) + http_req, + create_session, + BlinkAuthenticationException, + BlinkException, + BlinkURLHandler, +) from blinkpy.helpers.constants import __version__ import tests.mock_responses as mresp -USERNAME = 'foobar' -PASSWORD = 'deadbeef' +USERNAME = "foobar" +PASSWORD = "deadbeef" -@mock.patch('blinkpy.helpers.util.Session.send', - side_effect=mresp.mocked_session_send) +@mock.patch("blinkpy.helpers.util.Session.send", side_effect=mresp.mocked_session_send) class TestBlinkSetup(unittest.TestCase): """Test the Blink class in blinkpy.""" def setUp(self): """Set up Blink module.""" self.blink_no_cred = Blink() - self.blink = Blink(username=USERNAME, - password=PASSWORD) - self.blink.sync['test'] = BlinkSyncModule(self.blink, - 'test', - '1234', - []) - self.blink.urls = BlinkURLHandler('test') + self.blink = Blink(username=USERNAME, password=PASSWORD) + self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", "1234", []) + self.blink.urls = BlinkURLHandler("test") self.blink.session = create_session() def tearDown(self): @@ -63,122 +62,129 @@ class TestBlinkSetup(unittest.TestCase): 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() + ((region_id, region),) = mresp.LOGIN_RESPONSE["region"].items() self.blink.urls = BlinkURLHandler(region_id) with self.assertRaises(BlinkException): self.blink.get_ids() - @mock.patch('blinkpy.blinkpy.getpass.getpass') + @mock.patch("blinkpy.blinkpy.getpass.getpass") 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): + with mock.patch("builtins.input", return_value=USERNAME): self.assertTrue(self.blink_no_cred.login()) # pylint: disable=protected-access self.assertEqual(self.blink_no_cred._username, USERNAME) # pylint: disable=protected-access self.assertEqual(self.blink_no_cred._password, PASSWORD) - @mock.patch('blinkpy.blinkpy.getpass.getpass') - @mock.patch('blinkpy.blinkpy.Blink.get_auth_token') + @mock.patch("blinkpy.blinkpy.getpass.getpass") + @mock.patch("blinkpy.blinkpy.Blink.get_auth_token") def test_no_cred_file(self, getpwd, getauth, mock_sess): """Check that normal login occurs when cred file doesn't exist.""" # pylint: disable=protected-access - self.blink._cred_file = '/tmp/fake.file' + self.blink._cred_file = "/tmp/fake.file" getpwd.return_value = PASSWORD getauth.return_value = True - with mock.patch('builtins.input', return_value=USERNAME): + with mock.patch("builtins.input", return_value=USERNAME): self.assertTrue(self.blink.login()) def test_exit_on_missing_json(self, mock_sess): """Test that we fail on missing json data.""" # pylint: disable=protected-access - self.blink._cred_file = '/tmp/fake.file' - with mock.patch('os.path.isfile', return_value=True): - with mock.patch('builtins.open', mock.mock_open(read_data="{}")): + self.blink._cred_file = "/tmp/fake.file" + with mock.patch("os.path.isfile", return_value=True): + with mock.patch("builtins.open", mock.mock_open(read_data="{}")): self.assertFalse(self.blink.login()) def test_exit_on_bad_json(self, mock_sess): """Test that we fail on bad json format.""" # pylint: disable=protected-access - self.blink._cred_file = '/tmp/fake.file' - with mock.patch('os.path.isfile', return_value=True): - with mock.patch('builtins.open', mock.mock_open(read_data='{]')): + self.blink._cred_file = "/tmp/fake.file" + with mock.patch("os.path.isfile", return_value=True): + with mock.patch("builtins.open", mock.mock_open(read_data="{]")): self.assertFalse(self.blink.login()) - @mock.patch('blinkpy.blinkpy.json.load') + @mock.patch("blinkpy.blinkpy.json.load") def test_cred_file(self, mockjson, mock_sess): """Test that loading credential file works.""" # pylint: disable=protected-access - self.blink_no_cred._cred_file = '/tmp/fake.file' - mockjson.return_value = {'username': 'foo', 'password': 'bar'} - with mock.patch('os.path.isfile', return_value=True): - with mock.patch('builtins.open', mock.mock_open(read_data='')): + self.blink_no_cred._cred_file = "/tmp/fake.file" + mockjson.return_value = {"username": "foo", "password": "bar"} + with mock.patch("os.path.isfile", return_value=True): + with mock.patch("builtins.open", mock.mock_open(read_data="")): self.assertTrue(self.blink_no_cred.login()) # pylint: disable=protected-access - self.assertEqual(self.blink_no_cred._username, 'foo') + self.assertEqual(self.blink_no_cred._username, "foo") # pylint: disable=protected-access - self.assertEqual(self.blink_no_cred._password, 'bar') + self.assertEqual(self.blink_no_cred._password, "bar") def test_bad_request(self, mock_sess): """Check that we raise an Exception with a bad request.""" self.blink.session = create_session() - explog = ("WARNING:blinkpy.helpers.util:" - "Response from server: 200 - foo") + explog = "WARNING:blinkpy.helpers.util:" "Response from server: 200 - foo" with self.assertRaises(BlinkException): - http_req(self.blink, reqtype='bad') + http_req(self.blink, reqtype="bad") with self.assertLogs() as logrecord: - http_req(self.blink, reqtype='post', is_retry=True) + http_req(self.blink, reqtype="post", is_retry=True) self.assertEqual(logrecord.output, [explog]) 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'] + authtoken = self.blink.get_auth_token()["TOKEN_AUTH"] + expected = mresp.LOGIN_RESPONSE["authtoken"]["authtoken"] self.assertEqual(authtoken, expected) 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 - bad_header = {'Host': self.blink._host, 'TOKEN_AUTH': 'BADTOKEN'} + bad_header = {"Host": self.blink._host, "TOKEN_AUTH": "BADTOKEN"} # pylint: disable=protected-access self.blink._auth_header = bad_header self.assertEqual(self.blink.auth_header, bad_header) api.request_homescreen(self.blink) self.assertEqual(self.blink.auth_header, original_header) - @mock.patch('blinkpy.api.request_networks') + @mock.patch("blinkpy.api.request_networks") def test_multiple_networks(self, mock_net, mock_sess): """Check that we handle multiple networks appropriately.""" mock_net.return_value = { - 'networks': [{'id': 1234, 'account_id': 1111}, - {'id': 5678, 'account_id': 2222}] + "networks": [ + {"id": 1234, "account_id": 1111}, + {"id": 5678, "account_id": 2222}, + ] + } + self.blink.networks = { + "0000": {"onboarded": False, "name": "foo"}, + "5678": {"onboarded": True, "name": "bar"}, + "1234": {"onboarded": False, "name": "test"}, } - self.blink.networks = {'0000': {'onboarded': False, 'name': 'foo'}, - '5678': {'onboarded': True, 'name': 'bar'}, - '1234': {'onboarded': False, 'name': 'test'}} self.blink.get_ids() - self.assertTrue('5678' in self.blink.network_ids) + self.assertTrue("5678" in self.blink.network_ids) self.assertEqual(self.blink.account_id, 2222) - @mock.patch('blinkpy.api.request_networks') + @mock.patch("blinkpy.api.request_networks") def test_multiple_onboarded_networks(self, mock_net, mock_sess): """Check that we handle multiple networks appropriately.""" mock_net.return_value = { - 'networks': [{'id': 0000, 'account_id': 2222}, - {'id': 5678, 'account_id': 1111}] + "networks": [ + {"id": 0000, "account_id": 2222}, + {"id": 5678, "account_id": 1111}, + ] + } + self.blink.networks = { + "0000": {"onboarded": False, "name": "foo"}, + "5678": {"onboarded": True, "name": "bar"}, + "1234": {"onboarded": True, "name": "test"}, } - self.blink.networks = {'0000': {'onboarded': False, 'name': 'foo'}, - '5678': {'onboarded': True, 'name': 'bar'}, - '1234': {'onboarded': True, 'name': 'test'}} self.blink.get_ids() - self.assertTrue('5678' in self.blink.network_ids) - self.assertTrue('1234' in self.blink.network_ids) + self.assertTrue("5678" in self.blink.network_ids) + self.assertTrue("1234" in self.blink.network_ids) self.assertEqual(self.blink.account_id, 1111) - @mock.patch('blinkpy.blinkpy.time.time') + @mock.patch("blinkpy.blinkpy.time.time") def test_throttle(self, mock_time, mock_sess): """Check throttling functionality.""" now = self.blink.refresh_rate + 1 @@ -186,8 +192,9 @@ class TestBlinkSetup(unittest.TestCase): self.assertEqual(self.blink.last_refresh, None) self.assertEqual(self.blink.check_if_ok_to_update(), True) self.assertEqual(self.blink.last_refresh, None) - with mock.patch('blinkpy.sync_module.BlinkSyncModule.refresh', - return_value=True): + with mock.patch( + "blinkpy.sync_module.BlinkSyncModule.refresh", return_value=True + ): self.blink.refresh() self.assertEqual(self.blink.last_refresh, now) @@ -196,29 +203,35 @@ class TestBlinkSetup(unittest.TestCase): def test_sync_case_insensitive_dict(self, mock_sess): """Check that we can access sync modules ignoring case.""" - self.assertEqual(self.blink.sync['test'].name, 'test') - self.assertEqual(self.blink.sync['TEST'].name, 'test') + self.assertEqual(self.blink.sync["test"].name, "test") + self.assertEqual(self.blink.sync["TEST"].name, "test") - @mock.patch('blinkpy.api.request_login') + @mock.patch("blinkpy.api.request_login") def test_unexpected_login(self, mock_login, mock_sess): """Check that we appropriately handle unexpected login info.""" mock_login.return_value = None self.assertFalse(self.blink.get_auth_token()) - @mock.patch('blinkpy.api.request_homescreen') + @mock.patch("blinkpy.api.request_homescreen") def test_get_cameras(self, mock_home, mock_sess): """Check retrieval of camera information.""" mock_home.return_value = { - 'cameras': [{'name': 'foo', 'network_id': 1234, 'id': 5678}, - {'name': 'bar', 'network_id': 1234, 'id': 5679}, - {'name': 'test', 'network_id': 4321, 'id': 0000}] + "cameras": [ + {"name": "foo", "network_id": 1234, "id": 5678}, + {"name": "bar", "network_id": 1234, "id": 5679}, + {"name": "test", "network_id": 4321, "id": 0000}, + ] } result = self.blink.get_cameras() - self.assertEqual(result, {'1234': [{'name': 'foo', 'id': 5678}, - {'name': 'bar', 'id': 5679}], - '4321': [{'name': 'test', 'id': 0000}]}) + self.assertEqual( + result, + { + "1234": [{"name": "foo", "id": 5678}, {"name": "bar", "id": 5679}], + "4321": [{"name": "test", "id": 0000}], + }, + ) - @mock.patch('blinkpy.api.request_homescreen') + @mock.patch("blinkpy.api.request_homescreen") def test_get_cameras_failure(self, mock_home, mock_sess): """Check that on failure we initialize empty info and move on.""" mock_home.return_value = {} diff --git a/tests/test_cameras.py b/tests/test_cameras.py index d1f18ce..ca2fbbf 100644 --- a/tests/test_cameras.py +++ b/tests/test_cameras.py @@ -14,42 +14,40 @@ from blinkpy.sync_module import BlinkSyncModule from blinkpy.camera import BlinkCamera import tests.mock_responses as mresp -USERNAME = 'foobar' -PASSWORD = 'deadbeef' +USERNAME = "foobar" +PASSWORD = "deadbeef" CAMERA_CFG = { - 'camera': [ + "camera": [ { - 'battery_voltage': 90, - 'motion_alert': True, - 'wifi_strength': -30, - 'temperature': 68 + "battery_voltage": 90, + "motion_alert": True, + "wifi_strength": -30, + "temperature": 68, } ] } -@mock.patch('blinkpy.helpers.util.Session.send', - side_effect=mresp.mocked_session_send) +@mock.patch("blinkpy.helpers.util.Session.send", side_effect=mresp.mocked_session_send) class TestBlinkCameraSetup(unittest.TestCase): """Test the Blink class in blinkpy.""" def setUp(self): """Set up Blink module.""" - self.blink = blinkpy.Blink(username=USERNAME, - password=PASSWORD) + self.blink = blinkpy.Blink(username=USERNAME, password=PASSWORD) header = { - 'Host': 'abc.zxc', - 'TOKEN_AUTH': mresp.LOGIN_RESPONSE['authtoken']['authtoken'] + "Host": "abc.zxc", + "TOKEN_AUTH": mresp.LOGIN_RESPONSE["authtoken"]["authtoken"], } # pylint: disable=protected-access self.blink._auth_header = header self.blink.session = create_session() - self.blink.urls = BlinkURLHandler('test') - self.blink.sync['test'] = BlinkSyncModule(self.blink, 'test', 1234, []) - self.camera = BlinkCamera(self.blink.sync['test']) - self.camera.name = 'foobar' - self.blink.sync['test'].cameras['foobar'] = self.camera + self.blink.urls = BlinkURLHandler("test") + self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", 1234, []) + self.camera = BlinkCamera(self.blink.sync["test"]) + self.camera.name = "foobar" + self.blink.sync["test"].cameras["foobar"] = self.camera def tearDown(self): """Clean up after test.""" @@ -58,98 +56,99 @@ class TestBlinkCameraSetup(unittest.TestCase): def test_camera_update(self, mock_sess): """Test that we can properly update camera properties.""" config = { - 'name': 'new', - 'id': 1234, - 'network_id': 5678, - 'serial': '12345678', - 'enabled': False, - 'battery_voltage': 90, - 'battery_state': 'ok', - 'temperature': 68, - 'wifi_strength': 4, - 'thumbnail': '/thumb', + "name": "new", + "id": 1234, + "network_id": 5678, + "serial": "12345678", + "enabled": False, + "battery_voltage": 90, + "battery_state": "ok", + "temperature": 68, + "wifi_strength": 4, + "thumbnail": "/thumb", } - self.camera.last_record = ['1'] + self.camera.last_record = ["1"] self.camera.sync.last_record = { - 'new': { - 'clip': '/test.mp4', - 'time': '1970-01-01T00:00:00' - } + "new": {"clip": "/test.mp4", "time": "1970-01-01T00:00:00"} } mock_sess.side_effect = [ - mresp.MockResponse({'temp': 71}, 200), - 'test', - 'foobar' + mresp.MockResponse({"temp": 71}, 200), + "test", + "foobar", ] self.camera.update(config) - self.assertEqual(self.camera.name, 'new') - self.assertEqual(self.camera.camera_id, '1234') - self.assertEqual(self.camera.network_id, '5678') - self.assertEqual(self.camera.serial, '12345678') + self.assertEqual(self.camera.name, "new") + self.assertEqual(self.camera.camera_id, "1234") + self.assertEqual(self.camera.network_id, "5678") + self.assertEqual(self.camera.serial, "12345678") self.assertEqual(self.camera.motion_enabled, False) - self.assertEqual(self.camera.battery, 'ok') + self.assertEqual(self.camera.battery, "ok") self.assertEqual(self.camera.temperature, 68) self.assertEqual(self.camera.temperature_c, 20) self.assertEqual(self.camera.temperature_calibrated, 71) self.assertEqual(self.camera.wifi_strength, 4) - self.assertEqual(self.camera.thumbnail, - 'https://rest-test.immedia-semi.com/thumb.jpg') - self.assertEqual(self.camera.clip, - 'https://rest-test.immedia-semi.com/test.mp4') - self.assertEqual(self.camera.image_from_cache, 'test') - self.assertEqual(self.camera.video_from_cache, 'foobar') + self.assertEqual( + self.camera.thumbnail, "https://rest-test.immedia-semi.com/thumb.jpg" + ) + self.assertEqual( + self.camera.clip, "https://rest-test.immedia-semi.com/test.mp4" + ) + self.assertEqual(self.camera.image_from_cache, "test") + self.assertEqual(self.camera.video_from_cache, "foobar") def test_no_thumbnails(self, mock_sess): """Tests that thumbnail is 'None' if none found.""" - mock_sess.return_value = 'foobar' - self.camera.last_record = ['1'] + mock_sess.return_value = "foobar" + self.camera.last_record = ["1"] config = { - 'name': 'new', - 'id': 1234, - 'network_id': 5678, - 'serial': '12345678', - 'enabled': False, - 'battery_voltage': 90, - 'battery_state': 'ok', - 'temperature': 68, - 'wifi_strength': 4, - 'thumbnail': '', - } - self.camera.sync.homescreen = { - 'devices': [] + "name": "new", + "id": 1234, + "network_id": 5678, + "serial": "12345678", + "enabled": False, + "battery_voltage": 90, + "battery_state": "ok", + "temperature": 68, + "wifi_strength": 4, + "thumbnail": "", } + self.camera.sync.homescreen = {"devices": []} self.assertEqual(self.camera.temperature_calibrated, None) with self.assertLogs() as logrecord: self.camera.update(config, force=True) self.assertEqual(self.camera.thumbnail, None) - self.assertEqual(self.camera.last_record, ['1']) + self.assertEqual(self.camera.last_record, ["1"]) self.assertEqual(self.camera.temperature_calibrated, 68) self.assertEqual( logrecord.output, - [("WARNING:blinkpy.camera:Could not retrieve calibrated " - "temperature."), - ("WARNING:blinkpy.camera:Could not find thumbnail for camera new" - "\nNoneType: None")] + [ + ( + "WARNING:blinkpy.camera:Could not retrieve calibrated " + "temperature." + ), + ( + "WARNING:blinkpy.camera:Could not find thumbnail for camera new" + "\nNoneType: None" + ), + ], ) def test_no_video_clips(self, mock_sess): """Tests that we still proceed with camera setup with no videos.""" - mock_sess.return_value = 'foobar' + mock_sess.return_value = "foobar" config = { - 'name': 'new', - 'id': 1234, - 'network_id': 5678, - 'serial': '12345678', - 'enabled': False, - 'battery_voltage': 90, - 'battery_state': 'ok', - 'temperature': 68, - 'wifi_strength': 4, - 'thumbnail': '/foobar', - } - self.camera.sync.homescreen = { - 'devices': [] + "name": "new", + "id": 1234, + "network_id": 5678, + "serial": "12345678", + "enabled": False, + "battery_voltage": 90, + "battery_state": "ok", + "temperature": 68, + "wifi_strength": 4, + "thumbnail": "/foobar", } + self.camera.sync.homescreen = {"devices": []} self.camera.update(config, force_cache=True) self.assertEqual(self.camera.clip, None) self.assertEqual(self.camera.video_from_cache, None) diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py index 266b773..5a8912f 100644 --- a/tests/test_sync_module.py +++ b/tests/test_sync_module.py @@ -6,44 +6,41 @@ from blinkpy import blinkpy from blinkpy.sync_module import BlinkSyncModule from blinkpy.camera import BlinkCamera -USERNAME = 'foobar' -PASSWORD = 'deadbeef' +USERNAME = "foobar" +PASSWORD = "deadbeef" -@mock.patch('blinkpy.api.http_req') +@mock.patch("blinkpy.api.http_req") class TestBlinkSyncModule(unittest.TestCase): """Test BlinkSyncModule functions in blinkpy.""" def setUp(self): """Set up Blink module.""" - self.blink = blinkpy.Blink(username=USERNAME, - password=PASSWORD, - motion_interval=0) + self.blink = blinkpy.Blink( + username=USERNAME, password=PASSWORD, motion_interval=0 + ) # pylint: disable=protected-access - self.blink._auth_header = { - 'Host': 'test.url.tld', - 'TOKEN_AUTH': 'foobar123' - } + self.blink._auth_header = {"Host": "test.url.tld", "TOKEN_AUTH": "foobar123"} self.blink.last_refresh = 0 - self.blink.urls = blinkpy.BlinkURLHandler('test') - self.blink.sync['test'] = BlinkSyncModule(self.blink, - 'test', - '1234', - []) + self.blink.urls = blinkpy.BlinkURLHandler("test") + self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", "1234", []) self.camera = BlinkCamera(self.blink.sync) self.mock_start = [ - {'syncmodule': { - 'id': 1234, - 'network_id': 5678, - 'serial': '12345678', - 'status': 'foobar'}}, - {'event': True}, + { + "syncmodule": { + "id": 1234, + "network_id": 5678, + "serial": "12345678", + "status": "foobar", + } + }, + {"event": True}, {}, {}, None, - {'devicestatus': {}}, + {"devicestatus": {}}, ] - self.blink.sync['test'].network_info = {'network': {'armed': True}} + self.blink.sync["test"].network_info = {"network": {"armed": True}} def tearDown(self): """Clean up after test.""" @@ -53,173 +50,177 @@ class TestBlinkSyncModule(unittest.TestCase): def test_get_events(self, mock_resp): """Test get events function.""" - mock_resp.return_value = {'event': True} - self.assertEqual(self.blink.sync['test'].get_events(), True) + mock_resp.return_value = {"event": True} + self.assertEqual(self.blink.sync["test"].get_events(), True) def test_get_camera_info(self, mock_resp): """Test get camera info function.""" - mock_resp.return_value = {'camera': ['foobar']} - self.assertEqual(self.blink.sync['test'].get_camera_info('1234'), - 'foobar') + mock_resp.return_value = {"camera": ["foobar"]} + self.assertEqual(self.blink.sync["test"].get_camera_info("1234"), "foobar") def test_check_new_videos_startup(self, mock_resp): """Test that check_new_videos does not block startup.""" - sync_module = self.blink.sync['test'] + sync_module = self.blink.sync["test"] self.blink.last_refresh = None self.assertFalse(sync_module.check_new_videos()) def test_check_new_videos(self, mock_resp): """Test recent video response.""" mock_resp.return_value = { - 'media': [{ - 'device_name': 'foo', - 'media': '/foo/bar.mp4', - 'created_at': '1990-01-01T00:00:00+00:00' - }] + "media": [ + { + "device_name": "foo", + "media": "/foo/bar.mp4", + "created_at": "1990-01-01T00:00:00+00:00", + } + ] } - sync_module = self.blink.sync['test'] - sync_module.cameras = {'foo': None} + sync_module = self.blink.sync["test"] + sync_module.cameras = {"foo": None} sync_module.blink.last_refresh = 0 self.assertEqual(sync_module.motion, {}) self.assertTrue(sync_module.check_new_videos()) - self.assertEqual(sync_module.last_record['foo'], - {'clip': '/foo/bar.mp4', - 'time': '1990-01-01T00:00:00+00:00'}) - self.assertEqual(sync_module.motion, {'foo': True}) - mock_resp.return_value = {'media': []} + self.assertEqual( + sync_module.last_record["foo"], + {"clip": "/foo/bar.mp4", "time": "1990-01-01T00:00:00+00:00"}, + ) + self.assertEqual(sync_module.motion, {"foo": True}) + mock_resp.return_value = {"media": []} self.assertTrue(sync_module.check_new_videos()) - self.assertEqual(sync_module.motion, {'foo': False}) - self.assertEqual(sync_module.last_record['foo'], - {'clip': '/foo/bar.mp4', - 'time': '1990-01-01T00:00:00+00:00'}) + self.assertEqual(sync_module.motion, {"foo": False}) + self.assertEqual( + sync_module.last_record["foo"], + {"clip": "/foo/bar.mp4", "time": "1990-01-01T00:00:00+00:00"}, + ) def test_check_new_videos_old_date(self, mock_resp): """Test videos return response with old date.""" mock_resp.return_value = { - 'media': [{ - 'device_name': 'foo', - 'media': '/foo/bar.mp4', - 'created_at': '1970-01-01T00:00:00+00:00' - }] + "media": [ + { + "device_name": "foo", + "media": "/foo/bar.mp4", + "created_at": "1970-01-01T00:00:00+00:00", + } + ] } - sync_module = self.blink.sync['test'] - sync_module.cameras = {'foo': None} + sync_module = self.blink.sync["test"] + sync_module.cameras = {"foo": None} sync_module.blink.last_refresh = 1000 self.assertTrue(sync_module.check_new_videos()) - self.assertEqual(sync_module.motion, {'foo': False}) + self.assertEqual(sync_module.motion, {"foo": False}) def test_check_no_motion_if_not_armed(self, mock_resp): """Test that motion detection is not set if module unarmed.""" mock_resp.return_value = { - 'media': [{ - 'device_name': 'foo', - 'media': '/foo/bar.mp4', - 'created_at': '1990-01-01T00:00:00+00:00' - }] + "media": [ + { + "device_name": "foo", + "media": "/foo/bar.mp4", + "created_at": "1990-01-01T00:00:00+00:00", + } + ] } - sync_module = self.blink.sync['test'] - sync_module.cameras = {'foo': None} + sync_module = self.blink.sync["test"] + sync_module.cameras = {"foo": None} sync_module.blink.last_refresh = 1000 self.assertTrue(sync_module.check_new_videos()) - self.assertEqual(sync_module.motion, {'foo': True}) - sync_module.network_info = {'network': {'armed': False}} + self.assertEqual(sync_module.motion, {"foo": True}) + sync_module.network_info = {"network": {"armed": False}} self.assertTrue(sync_module.check_new_videos()) - self.assertEqual(sync_module.motion, {'foo': False}) + self.assertEqual(sync_module.motion, {"foo": False}) def test_check_multiple_videos(self, mock_resp): """Test motion found even with multiple videos.""" mock_resp.return_value = { - 'media': [ + "media": [ { - 'device_name': 'foo', - 'media': '/foo/bar.mp4', - 'created_at': '1970-01-01T00:00:00+00:00' + "device_name": "foo", + "media": "/foo/bar.mp4", + "created_at": "1970-01-01T00:00:00+00:00", }, { - 'device_name': 'foo', - 'media': '/bar/foo.mp4', - 'created_at': '1990-01-01T00:00:00+00:00' + "device_name": "foo", + "media": "/bar/foo.mp4", + "created_at": "1990-01-01T00:00:00+00:00", }, { - 'device_name': 'foo', - 'media': '/foobar.mp4', - 'created_at': '1970-01-01T00:00:01+00:00' - } + "device_name": "foo", + "media": "/foobar.mp4", + "created_at": "1970-01-01T00:00:01+00:00", + }, ] } - sync_module = self.blink.sync['test'] - sync_module.cameras = {'foo': None} + sync_module = self.blink.sync["test"] + sync_module.cameras = {"foo": None} sync_module.blink.last_refresh = 1000 self.assertTrue(sync_module.check_new_videos()) - self.assertEqual(sync_module.motion, {'foo': True}) + self.assertEqual(sync_module.motion, {"foo": True}) expected_result = { - 'foo': { - 'clip': '/bar/foo.mp4', - 'time': '1990-01-01T00:00:00+00:00' - } + "foo": {"clip": "/bar/foo.mp4", "time": "1990-01-01T00:00:00+00:00"} } self.assertEqual(sync_module.last_record, expected_result) def test_check_new_videos_failed(self, mock_resp): """Test method when response is unexpected.""" - mock_resp.side_effect = [None, 'just a string', {}] - sync_module = self.blink.sync['test'] - sync_module.cameras = {'foo': None} + mock_resp.side_effect = [None, "just a string", {}] + sync_module = self.blink.sync["test"] + sync_module.cameras = {"foo": None} - sync_module.motion['foo'] = True + sync_module.motion["foo"] = True self.assertFalse(sync_module.check_new_videos()) - self.assertFalse(sync_module.motion['foo']) + self.assertFalse(sync_module.motion["foo"]) - sync_module.motion['foo'] = True + sync_module.motion["foo"] = True self.assertFalse(sync_module.check_new_videos()) - self.assertFalse(sync_module.motion['foo']) + self.assertFalse(sync_module.motion["foo"]) - sync_module.motion['foo'] = True + sync_module.motion["foo"] = True self.assertFalse(sync_module.check_new_videos()) - self.assertFalse(sync_module.motion['foo']) + self.assertFalse(sync_module.motion["foo"]) def test_sync_start(self, mock_resp): """Test sync start function.""" mock_resp.side_effect = self.mock_start - self.blink.sync['test'].start() - self.assertEqual(self.blink.sync['test'].name, 'test') - self.assertEqual(self.blink.sync['test'].sync_id, 1234) - self.assertEqual(self.blink.sync['test'].network_id, 5678) - self.assertEqual(self.blink.sync['test'].serial, '12345678') - self.assertEqual(self.blink.sync['test'].status, 'foobar') + self.blink.sync["test"].start() + self.assertEqual(self.blink.sync["test"].name, "test") + self.assertEqual(self.blink.sync["test"].sync_id, 1234) + self.assertEqual(self.blink.sync["test"].network_id, 5678) + self.assertEqual(self.blink.sync["test"].serial, "12345678") + self.assertEqual(self.blink.sync["test"].status, "foobar") def test_unexpected_summary(self, mock_resp): """Test unexpected summary response.""" self.mock_start[0] = None mock_resp.side_effect = self.mock_start - self.assertFalse(self.blink.sync['test'].start()) + self.assertFalse(self.blink.sync["test"].start()) def test_summary_with_no_network_id(self, mock_resp): """Test handling of bad summary.""" - self.mock_start[0]['syncmodule'] = None + self.mock_start[0]["syncmodule"] = None mock_resp.side_effect = self.mock_start - self.assertFalse(self.blink.sync['test'].start()) + self.assertFalse(self.blink.sync["test"].start()) def test_summary_with_only_network_id(self, mock_resp): """Test handling of sparse summary.""" - self.mock_start[0]['syncmodule'] = {'network_id': 8675309} + self.mock_start[0]["syncmodule"] = {"network_id": 8675309} mock_resp.side_effect = self.mock_start - self.blink.sync['test'].start() - self.assertEqual(self.blink.sync['test'].network_id, 8675309) + self.blink.sync["test"].start() + self.assertEqual(self.blink.sync["test"].network_id, 8675309) def test_unexpected_camera_info(self, mock_resp): """Test unexpected camera info response.""" - self.blink.sync['test'].cameras['foo'] = None + self.blink.sync["test"].cameras["foo"] = None self.mock_start[5] = None mock_resp.side_effect = self.mock_start - self.blink.sync['test'].start() - self.assertEqual(self.blink.sync['test'].cameras, {'foo': None}) + self.blink.sync["test"].start() + self.assertEqual(self.blink.sync["test"].cameras, {"foo": None}) def test_missing_camera_info(self, mock_resp): """Test missing key from camera info response.""" - self.blink.sync['test'].cameras['foo'] = None + self.blink.sync["test"].cameras["foo"] = None self.mock_start[5] = {} - self.blink.sync['test'].start() - self.assertEqual(self.blink.sync['test'].cameras, {'foo': None}) + self.blink.sync["test"].start() + self.assertEqual(self.blink.sync["test"].cameras, {"foo": None}) diff --git a/tests/test_util.py b/tests/test_util.py index d5c1a49..f048c35 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -43,17 +43,18 @@ class TestUtil(unittest.TestCase): self.assertEqual(2, len(calls)) # Fake time as 4 seconds from now - with mock.patch('time.time', return_value=now_plus_four): + with mock.patch("time.time", return_value=now_plus_four): test_throttle() self.assertEqual(2, len(calls)) # Fake time as 6 seconds from now - with mock.patch('time.time', return_value=now_plus_six): + with mock.patch("time.time", return_value=now_plus_six): test_throttle() self.assertEqual(3, len(calls)) def test_throttle_per_instance(self): """Test that throttle is done once per instance of class.""" + class Tester: """A tester class for throttling.""" @@ -68,6 +69,7 @@ class TestUtil(unittest.TestCase): def test_throttle_on_two_methods(self): """Test that throttle works for multiple methods.""" + class Tester: """A tester class for throttling.""" @@ -91,24 +93,24 @@ class TestUtil(unittest.TestCase): self.assertEqual(tester.test1(), None) self.assertEqual(tester.test2(), None) - with mock.patch('time.time', return_value=now_plus_4): + with mock.patch("time.time", return_value=now_plus_4): self.assertEqual(tester.test1(), True) self.assertEqual(tester.test2(), None) - with mock.patch('time.time', return_value=now_plus_6): + with mock.patch("time.time", return_value=now_plus_6): self.assertEqual(tester.test1(), None) self.assertEqual(tester.test2(), True) def test_legacy_subdomains(self): """Test that subdomain can be set to legacy mode.""" - urls = BlinkURLHandler('test') - self.assertEqual(urls.subdomain, 'rest-test') - urls = BlinkURLHandler('test', legacy=True) - self.assertEqual(urls.subdomain, 'rest.test') + urls = BlinkURLHandler("test") + self.assertEqual(urls.subdomain, "rest-test") + urls = BlinkURLHandler("test", legacy=True) + self.assertEqual(urls.subdomain, "rest.test") def test_time_to_seconds(self): """Test time to seconds conversion.""" - correct_time = '1970-01-01T00:00:05+00:00' - wrong_time = '1/1/1970 00:00:03' + correct_time = "1970-01-01T00:00:05+00:00" + wrong_time = "1/1/1970 00:00:03" self.assertEqual(time_to_seconds(correct_time), 5) self.assertFalse(time_to_seconds(wrong_time)) diff --git a/tox.ini b/tox.ini index 437ed1a..c9b8f42 100644 --- a/tox.ini +++ b/tox.ini @@ -23,7 +23,7 @@ deps = -r{toxinidir}/requirements.txt -r{toxinidir}/requirements_test.txt -[testenv:lint] +[testenv:pylint] deps = -r{toxinidir}/requirements.txt -r{toxinidir}/requirements_test.txt @@ -31,8 +31,16 @@ basepython = python3 ignore_errors = True commands = pylint --rcfile={toxinidir}/pylintrc blinkpy tests app + +[testenv:lint] +deps = + -r{toxinidir}/requirements.txt + -r{toxinidir}/requirements_test.txt +basepython = python3 +commands = flake8 blinkpy tests app pydocstyle blinkpy tests app + black --check --diff blinkpy tests app rst-lint README.rst rst-lint CHANGES.rst