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

Merge pull request #100 from fronzbot/dev

Browse Source 
Release 0.10.0
This commit is contained in:
Kevin Fronczak
2018-10-16 17:43:21 -04:00
committed by GitHub
parent 07e1759819 e747fb1010
commit 4e6957a772
16 changed files with 705 additions and 656 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+4 -1
View File
@@ -4,7 +4,7 @@ matrix:
include:
- python: "3.6"
env: TOXENV=lint
- python: "3.5"
- python: "3.5.3"
env: TOXENV=py35
- python: "3.6"
env: TOXENV=py36
@@ -13,6 +13,9 @@ matrix:
- python: "3.7"
env: TOXENV=py37
dist: xenial
- python: "3.8-dev"
env: TOXENV=py38
dist: xenial
install: pip install -U tox coveralls
language: python
CHANGES.rst
+37 -1
View File
@@ -3,8 +3,44 @@ Changelog
A list of changes between each release
0.9.0.dev (Development Version)
0.10.0 (2018-10-16)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- Moved all API calls to own module for easier maintainability
- Added network ids to sync module and cameras to allow for multi-network use
- Removed dependency on video existance prior to camera setup (fixes `#93 <https://github.com/fronzbot/blinkpy/issues/#93>`_)
- Camera wifi_strength now reported in wifi "bars" rather than dBm due to API endpoint change
- Use homescreen thumbnail as fallback in case it's not in the camera endpoint
- Removed "armed" and "status" attributes from camera (status of camera only reported by "motion_enabled" now)
- Added serial number attributes to sync module and cameras
- Check network_id from login response and verify that network is onboarded (fixes `#90 <https://github.com/fronzbot/#90>`_)
- Check if retrieved clip is "None" prior to storing in cache
0.9.0 (2018-09-27)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- Complete code refactoring to enable future multi-sync module support
- Add image and video caching to the cameras
- Add internal throttling of system refresh
- Use session for http requests
**Breaking change:**
- Cameras now accessed through sync module Blink.sync.cameras
0.8.1 (2018-09-24)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- Update requirements_test.txt
- Update linter versions
- Fix pylint warnings
- Remove object from class declarations
- Remove useless returns from functions
- Fix pylint errors
- change if comparison to fix (consider-using-in)
- Disabled no else-if-return check
- Fix useless-import-alias
- Disable no-else-return
- Fix motion detection
- Use an array of recent video clips to determine if motion has been detected.
- Reset the value every system refresh
0.8.0 (2018-05-21)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
blinkpy/api.py
+159
View File
@@ -0,0 +1,159 @@
"""Implements known blink API calls."""
import logging
from json import dumps
import blinkpy.helpers.errors as ERROR
from blinkpy.helpers.util import http_req, BlinkException
from blinkpy.helpers.constants import DEFAULT_URL
_LOGGER = logging.getLogger(__name__)
def request_login(blink, url, username, password):
"""Login request."""
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')
def request_networks(blink):
"""Request network information."""
url = "{}/networks".format(blink.urls.base_url)
return http_get(blink, url)
def request_syncmodule(blink, network):
"""Request sync module info."""
url = "{}/network/{}/syncmodules".format(blink.urls.base_url, network)
return http_get(blink, url)
def request_system_arm(blink, network):
"""Arm system."""
url = "{}/network/{}/arm".format(blink.urls.base_url, network)
return http_post(blink, url)
def request_system_disarm(blink, network):
"""Disarm system."""
url = "{}/network/{}/disarm".format(blink.urls.base_url, network)
return http_post(blink, url)
def request_command_status(blink, network, command_id):
"""Request command status."""
url = "{}/network/{}/command_id/{}".format(blink.urls.base_url,
network,
command_id)
return http_get(blink, url)
def request_homescreen(blink):
"""Request homescreen info."""
url = "{}/homescreen".format(blink.urls.base_url)
return http_get(blink, url)
def request_sync_events(blink, network):
"""Request events from sync module."""
url = "{}/events/network/{}".format(blink.urls.base_url, network)
return http_get(blink, url)
def request_new_image(blink, network, camera_id):
"""Request to capture new thumbnail for camera."""
url = "{}/network/{}/camera/{}/thumbnail".format(blink.urls.base_url,
network,
camera_id)
return http_post(blink, url)
def request_new_video(blink, network, camera_id):
"""Request to capture new video clip."""
url = "{}/network/{}/camera/{}/clip".format(blink.urls.base_url,
network,
camera_id)
return http_post(blink, url)
def request_video_count(blink, headers):
"""Request total video count."""
url = "{}/api/v2/videos/count".format(blink.urls.base_url)
return http_get(blink, url)
def request_videos(blink, page=0):
"""Perform a request for videos."""
url = "{}/api/v2/videos/page/{}".format(blink.urls.base_url, page)
return http_get(blink, url)
def request_cameras(blink, network):
"""Request all camera information."""
url = "{}/network/{}/cameras".format(blink.urls.base_url, network)
return http_get(blink, url)
def request_camera_info(blink, network, camera_id):
"""Request camera info for one camera."""
url = "{}/network/{}/camera/{}".format(blink.urls.base_url,
network,
camera_id)
return http_get(blink, url)
def request_camera_sensors(blink, network, camera_id):
"""Request camera sensor info for one camera."""
url = "{}/network/{}/camera/{}/signals".format(blink.urls.base_url,
network,
camera_id)
return http_get(blink, url)
def request_motion_detection_enable(blink, network, camera_id):
"""Enable motion detection for a camera."""
url = "{}/network/{}/camera/{}/enable".format(blink.urls.base_url,
network,
camera_id)
return http_post(blink, url)
def request_motion_detection_disable(blink, network, camera_id):
"""Disable motion detection for a camera."""
url = "{}/network/{}/camera/{}/disable".format(blink.urls.base_url,
network,
camera_id)
return http_post(blink, url)
def http_get(blink, url, stream=False, json=True):
"""
Perform an http get request.
:param url: URL to perform get request.
:param stream: Stream response? True/FALSE
:param json: Return json response? TRUE/False
"""
if blink.auth_header is None:
raise BlinkException(ERROR.AUTH_TOKEN)
return http_req(blink, url=url, headers=blink.auth_header,
reqtype='get', stream=stream, json_resp=json)
def http_post(blink, url):
"""
Perform an http post request.
:param url: URL to perfom post request.
"""
if blink.auth_header is None:
raise BlinkException(ERROR.AUTH_TOKEN)
return http_req(blink, url=url, headers=blink.auth_header, reqtype='post')
blinkpy/blinkpy.py
+39 -65
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
"""
blinkpy by Kevin Fronczak - A Blink camera Python library.
@@ -13,16 +12,16 @@ I am in no way affiliated with Blink, nor Immedia Inc.
"""
import time
import json
import getpass
import logging
import blinkpy.helpers.errors as ERROR
from blinkpy import api
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.helpers.util import (
http_req, create_session, BlinkURLHandler,
BlinkException, BlinkAuthenticationException)
create_session, BlinkURLHandler,
BlinkAuthenticationException)
from blinkpy.helpers.constants import (
DEFAULT_URL, BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL)
BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL)
REFRESH_RATE = 30
@@ -47,9 +46,6 @@ class Blink():
self._token = None
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
@@ -59,12 +55,13 @@ class Blink():
self.last_refresh = None
self.refresh_rate = refresh_rate
self.session = None
self.networks = []
self._login_url = LOGIN_URL
@property
def events(self):
"""Get all events on server."""
return self._events
def auth_header(self):
"""Return the authentication header."""
return self._auth_header
def start(self):
"""
@@ -79,12 +76,8 @@ class Blink():
self.get_auth_token()
self.get_ids()
self.sync = BlinkSyncModule(self, self._auth_header)
self.sync.get_videos()
if self.sync.video_count > 0:
self.sync.get_cameras()
self.sync.set_links()
self._events = self.events_request()
self.sync = BlinkSyncModule(self)
self.sync.start()
def login(self):
"""Prompt user for username and password."""
@@ -103,16 +96,13 @@ class Blink():
if not isinstance(self._password, str):
raise BlinkAuthenticationException(ERROR.PASSWORD)
headers = {'Host': DEFAULT_URL,
'Content-Type': 'application/json'}
data = json.dumps({
"email": self._username,
"password": self._password,
"client_specifier": "iPhone 9.2 | 2.2 | 222"
})
login_url = LOGIN_URL
self.session = create_session()
response = http_req(self, url=self._login_url, headers=headers,
data=data, json_resp=False, reqtype='post')
response = api.request_login(self,
login_url,
self._username,
self._password)
if response.status_code == 200:
response = response.json()
(self.region_id, self.region), = response['region'].items()
@@ -122,56 +112,42 @@ class Blink():
"when authenticating, "
"trying new url"), response.status_code
)
self._login_url = LOGIN_BACKUP_URL
response = http_req(self, url=self._login_url, headers=headers,
data=data, reqtype='post')
login_url = LOGIN_BACKUP_URL
response = api.request_login(self,
login_url,
self._username,
self._password)
self.region_id = 'piri'
self.region = "UNKNOWN"
self._host = "{}.{}".format(self.region_id, BLINK_URL)
self._token = response['authtoken']['authtoken']
self._auth_header = {'Host': self._host,
'TOKEN_AUTH': self._token}
self.networks = response['networks']
self.urls = BlinkURLHandler(self.region_id)
self._login_url = login_url
return self._auth_header
def get_ids(self):
"""Set the network ID and Account ID."""
response = self._network_request()
self.network_id = str(response['networks'][0]['id'])
self.account_id = str(response['networks'][0]['account_id'])
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 http_req(self, url=url, headers=headers, reqtype='get')
def events_request(self, skip_throttle=False):
"""Get events on server."""
url = "{}/{}".format(self.urls.event_url, self.network_id)
headers = self._auth_header
if self.check_if_ok_to_update() or skip_throttle:
self._last_events = http_req(self, url=url,
headers=headers,
reqtype='get')
return self._last_events
def summary_request(self, skip_throttle=False):
"""Get blink summary."""
url = self.urls.home_url
headers = self._auth_header
if headers is None:
raise BlinkException(ERROR.AUTH_TOKEN)
if self.check_if_ok_to_update() or skip_throttle:
self._last_summary = http_req(
self, url=url, headers=headers, reqtype='get')
return self._last_summary
response = api.request_networks(self)
# Look for only onboarded network, flag warning if multiple
# since it's unexpected
all_networks = []
for network, status in self.networks.items():
if status['onboarded']:
all_networks.append(network)
self.network_id = all_networks.pop(0)
for resp in response['networks']:
if str(resp['id']) == self.network_id:
self.account_id = resp['account_id']
if all_networks:
_LOGGER.error(("More than one unboarded network. "
"Platform may not work as intended. "
"Please open an issue on "
"https://github.com/fronzbot/blinkpy."))
def refresh(self, force_cache=False):
"""
@@ -181,8 +157,6 @@ class Blink():
"""
if self.check_if_ok_to_update() or force_cache:
_LOGGER.debug("Attempting refresh of cameras.")
self._last_events = self.events_request(skip_throttle=True)
self._last_summary = self.summary_request(skip_throttle=True)
self.sync.refresh(force_cache=force_cache)
def check_if_ok_to_update(self):
blinkpy/camera.py
+87 -100
View File
@@ -2,7 +2,7 @@
from shutil import copyfileobj
import logging
from requests.exceptions import RequestException
from blinkpy import api
_LOGGER = logging.getLogger(__name__)
@@ -12,30 +12,22 @@ MAX_CLIPS = 5
class BlinkCamera():
"""Class to initialize individual camera."""
def __init__(self, config, sync):
def __init__(self, sync):
"""Initiailize BlinkCamera."""
self.sync = sync
self.urls = self.sync.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.name = None
self.camera_id = None
self.network_id = None
self.thumbnail = None
self.serial = None
self.motion_enabled = None
self.battery_voltage = None
self.clip = None
self.temperature = None
self.battery_state = None
self.motion_detected = None
self.wifi_strength = None
self.camera_config = dict()
self.motion_enabled = None
self.last_record = list()
self.last_record = []
self._cached_image = None
self._cached_video = None
@@ -44,16 +36,14 @@ class BlinkCamera():
"""Return dictionary of all camera attributes."""
attributes = {
'name': self.name,
'device_id': self.id,
'status': self._status,
'armed': self.armed,
'camera_id': self.camera_id,
'serial': self.serial,
'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.sync.network_id,
@@ -61,31 +51,11 @@ class BlinkCamera():
}
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."""
@@ -107,50 +77,77 @@ class BlinkCamera():
def snap_picture(self):
"""Take a picture with camera to create a new thumbnail."""
self.sync.http_post(self.image_link)
return api.request_new_image(self.sync.blink,
self.network_id,
self.camera_id)
def set_motion_detect(self, enable):
"""Set motion detection."""
url = self.arm_link
if enable:
self.sync.http_post("{}{}".format(url, '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)
def update(self, config, force_cache=False):
"""Update camera info."""
self.name = config['name']
self.camera_id = str(config['camera_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']
# Check if thumbnail exists in config, if not try to
# get it from the homescreen info in teh sync module
# otherwise set it to None and log an error
new_thumbnail = None
if config['thumbnail']:
thumb_addr = config['thumbnail']
else:
self.sync.http_post("{}{}".format(url, 'disable'))
thumb_addr = self.get_thumb_from_homescreen()
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']
if thumb_addr is not None:
new_thumbnail = "{}{}.jpg".format(self.sync.urls.base_url,
thumb_addr)
# Check if a new motion clip has been recorded
# check_for_motion_method sets motion_detected variable
self.check_for_motion()
clip_addr = None
if self.last_record:
clip_addr = self.sync.all_clips[self.name][self.last_record[0]]
self.clip = "{}{}".format(self.sync.urls.base_url,
clip_addr)
# If the thumbnail or clip have changed, update the cache
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_config_request(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)
update_cached_video = False
if self._cached_video is None or self.motion_detected:
update_cached_video = True
# Check if the most recent clip is included in the last_record list
# and that the last_record list is populated
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)
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)
def check_for_motion(self):
"""Check if motion detected.."""
try:
records = sorted(self.sync.record_dates[self.name])
new_clip = records.pop()
@@ -168,30 +165,6 @@ class BlinkCamera():
_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 = 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 = self.sync.http_get(
self.clip, stream=True, json=False)
def image_refresh(self):
"""Refresh current thumbnail."""
url = self.urls.home_url
response = self.sync.http_get(url)['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.
@@ -216,3 +189,17 @@ class BlinkCamera():
response = self._cached_video
with open(path, 'wb') as vidfile:
copyfileobj(response.raw, vidfile)
def get_thumb_from_homescreen(self):
"""Retrieve thumbnail from homescreen."""
for device in self.sync.homescreen['devices']:
try:
device_type = device['device_type']
device_name = device['name']
device_thumb = device['thumbnail']
if device_type == 'camera' and device_name == self.name:
return device_thumb
except KeyError:
pass
_LOGGER.error("Could not find thumbnail for camera %s", self.name)
return None
blinkpy/helpers/constants.py
+5 -3
View File
@@ -3,12 +3,12 @@
import os
MAJOR_VERSION = 0
MINOR_VERSION = 9
MINOR_VERSION = 10
PATCH_VERSION = 0
__version__ = '{}.{}.{}'.format(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION)
REQUIRED_PYTHON_VER = (3, 4, 2)
REQUIRED_PYTHON_VER = (3, 5, 3)
PROJECT_NAME = 'blinkpy'
PROJECT_PACKAGE_NAME = 'blinkpy'
@@ -30,7 +30,9 @@ PROJECT_CLASSIFIERS = [
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Topic :: Home Automation'
]
blinkpy/helpers/util.py
+10 -4
View File
@@ -1,9 +1,9 @@
"""Useful functions for blinkpy."""
import logging
from requests import Request, Session
import blinkpy.helpers.errors as ERROR
from requests import Request, Session, exceptions
from blinkpy.helpers.constants import BLINK_URL
import blinkpy.helpers.errors as ERROR
_LOGGER = logging.getLogger(__name__)
@@ -19,7 +19,6 @@ 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
@@ -42,13 +41,20 @@ def http_req(blink, url='http://example.com', data=None, headers=None,
elif reqtype == 'get':
req = Request('GET', url, headers=headers)
else:
_LOGGER.error("Invalid request type: %s", reqtype)
raise BlinkException(ERROR.REQUEST)
prepped = req.prepare()
response = blink.session.send(prepped, stream=stream)
try:
response = blink.session.send(prepped, stream=stream)
except (exceptions.ConnectionError, exceptions.Timeout):
_LOGGER.error("Cannot connect to server. Possible outage.")
return None
if json_resp and 'code' in response.json():
if is_retry:
_LOGGER.error("Cannot authenticate with server.")
raise BlinkAuthenticationException(
(response.json()['code'], response.json()['message']))
else:
blinkpy/sync_module.py
+87 -166
View File
@@ -3,8 +3,8 @@
import logging
from requests.structures import CaseInsensitiveDict
from blinkpy import api
from blinkpy.camera import BlinkCamera
from blinkpy.helpers.util import http_req
from blinkpy.helpers.constants import ONLINE
_LOGGER = logging.getLogger(__name__)
@@ -13,117 +13,111 @@ _LOGGER = logging.getLogger(__name__)
class BlinkSyncModule():
"""Class to initialize sync module."""
def __init__(self, blink, header, urls=None):
def __init__(self, blink):
"""
Initialize Blink sync module.
:param blink: Blink class instantiation
:param header: Blink authentication header
:param urls: URL Handler instantiation (deprecated)
"""
self.blink = blink
self._auth_header = header
self._auth_header = blink.auth_header
self.network_id = blink.network_id
self.region = blink.region
self.region_id = blink.region_id
self.name = 'sync'
self.serial = None
self.status = None
self.sync_id = None
self.region = None
self.region_id = None
self._host = None
self._events = []
self.host = None
self.summary = None
self.homescreen = None
self.record_dates = {}
self.videos = {}
self.events = []
self.cameras = CaseInsensitiveDict({})
self._idlookup = {}
self._video_count = 0
self._all_videos = {}
self._summary = None
self.record_dates = dict()
self.first_init = True
self.all_clips = {}
@property
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,
}
return attr
@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."""
self.blink.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)
self._video_count = self.http_get(url)['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
return ONLINE[self.status]
@property
def arm(self):
"""Return status of sync module: armed/disarmed."""
return self._summary['network']['armed']
return self.homescreen['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)
self.http_post(url)
return api.request_system_arm(self.blink, self.network_id)
return api.request_system_disarm(self.blink, self.network_id)
def start(self):
"""Initialize the system."""
response = api.request_syncmodule(self.blink, self.network_id)
self.summary = response['syncmodule']
self.name = self.summary['name']
self.sync_id = self.summary['id']
self.network_id = self.summary['network_id']
self.serial = self.summary['serial']
self.status = self.summary['status']
self.events = self.get_events()
self.homescreen = api.request_homescreen(self.blink)
camera_info = self.get_camera_info()
for camera_config in camera_info:
name = camera_config['name']
self.cameras[name] = BlinkCamera(self)
self.videos = self.get_videos()
for camera_config in camera_info:
name = camera_config['name']
if name in self.cameras:
self.cameras[name].update(camera_config, force_cache=True)
def get_events(self):
"""Retrieve events from server."""
response = api.request_sync_events(self.blink, self.network_id)
return response['event']
def get_camera_info(self):
"""Retrieve camera information."""
response = api.request_cameras(self.blink, self.network_id)
return response['devicestatus']
def refresh(self, force_cache=False):
"""Get all blink cameras and pulls their most recent status."""
summary = self._summary_request()
events = self._events_request()
response = 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._summary = summary
self._events = events
self.events = self.get_events()
self.videos = self.get_videos()
self.homescreen = api.request_homescreen(self.blink)
camera_info = self.get_camera_info()
for camera_config in camera_info:
name = camera_config['name']
self.cameras[name].update(camera_config, force_cache=force_cache)
def get_videos(self, start_page=0, end_page=1):
"""
@@ -135,8 +129,9 @@ class BlinkSyncModule():
"""
videos = list()
all_dates = dict()
for page_num in range(start_page, end_page + 1):
this_page = self._video_request(page_num)
this_page = api.request_videos(self.blink, page=page_num)
if not this_page:
break
videos.append(this_page)
@@ -150,18 +145,23 @@ class BlinkSyncModule():
clip_date = clip_addr.split('_')[-6:]
clip_date = '_'.join(clip_date)
clip_date = clip_date.split('.')[0]
try:
self.all_clips[camera_name][clip_date] = clip_addr
except KeyError:
self.all_clips[camera_name] = {clip_date: clip_addr}
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(
self.videos[camera_name].append(
{
'clip': clip_addr,
'thumb': thumb_addr,
}
)
except KeyError:
self._all_videos[camera_name] = [
self.videos[camera_name] = [
{
'clip': clip_addr,
'thumb': thumb_addr,
@@ -169,83 +169,4 @@ class BlinkSyncModule():
]
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.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."""
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 _summary_request(self):
"""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)
return self.http_get(url)
def _status_request(self):
"""Get syncmodule status."""
url = "{}/{}/syncmodules".format(self.urls.network_url,
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.network_id,
str(camera_id))
return self.http_get(url)
def http_get(self, url, stream=False, json=True):
"""
Perform an http get request.
:param url: URL to perform get request
:param stream: Stream response? True/FALSE
:param json: Return json response? TRUE/False
"""
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.
:param url: URL to perform post request
"""
return http_req(self.blink, url=url, headers=self._auth_header,
reqtype='post')
return self.videos
requirements_test.txt
+1 -1
View File
@@ -7,4 +7,4 @@ pytest-cov>=2.3.1
pytest-sugar>=0.9.0
pytest-timeout>=1.0.0
restructuredtext-lint>=1.0.1
pygments>=2.2.0
pygments>=2.2.0
tests/mock_responses.py
+1 -9
View File
@@ -6,15 +6,7 @@ import blinkpy.helpers.constants as const
LOGIN_RESPONSE = {
'region': {'mock': 'Test'},
'networks': {
'summary': {'name': 'TestNetwork'},
'networks': [{
'name': 'TestNetwork',
'account_id': 1111,
'id': 2222,
'active': 'armed',
'armed': True,
'arm_string': 'Armed'
}]
'1234': {'name': 'test', 'onboarded': True}
},
'authtoken': {'authtoken': 'foobar123', 'message': 'auth'}
}
tests/test_api.py
+32
View File
@@ -0,0 +1,32 @@
"""Test various api functions."""
import unittest
from blinkpy import api
from blinkpy.blinkpy import Blink
from blinkpy.helpers.util import create_session
class TestBlinkAPI(unittest.TestCase):
"""Test the blink api module."""
def setUp(self):
"""Initialize the blink module."""
self.blink = Blink()
self.blink.session = create_session()
# pylint: disable=protected-access
self.blink._auth_header = {}
def tearDown(self):
"""Tear down blink module."""
self.blink = None
def test_http_req_connect_error(self):
"""Test http_get error condition."""
expected = ("ERROR:blinkpy.helpers.util:"
"Cannot connect to server. Possible outage.")
with self.assertLogs() as getlog:
api.http_get(self.blink, 'http://notreal.fake')
with self.assertLogs() as postlog:
api.http_post(self.blink, 'http://notreal.fake')
self.assertEqual(getlog.output, [expected])
self.assertEqual(postlog.output, [expected])
tests/test_blink_functions.py
+3 -70
View File
@@ -6,9 +6,7 @@ from requests import Request
from blinkpy import blinkpy
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.camera import BlinkCamera
from blinkpy.helpers.util import create_session
from blinkpy.helpers.constants import BLINK_URL
import tests.mock_responses as mresp
USERNAME = 'foobar'
@@ -52,85 +50,20 @@ class TestBlinkFunctions(unittest.TestCase):
'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.blink.session = create_session()
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_image_refresh(self, mock_sess):
"""Test image refresh function."""
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.BlinkSyncModule.camera_config_request')
@mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request')
def test_refresh(self, vid_req, req, mock_sess):
"""Test blinkpy refresh function."""
req.return_value = {'foo': 'bar'}
self.blink.sync.cameras = {'foobar': self.camera}
self.blink.sync.return_value = {'devices': [{'foo': 'bar'}]}
# pylint: disable=protected-access
summary = {'devices': [self.config]}
# pylint: disable=protected-access
events = {'foo': 'bar'}
vid_req.return_value = [
{
'camera_name': 'foobar',
'address': '/new.mp4',
'thumbnail': '/new',
}
]
with mock.patch('blinkpy.blinkpy.Blink.summary_request',
return_value=summary):
with mock.patch('blinkpy.blinkpy.Blink.events_request',
return_value=events):
self.blink.refresh_rate = 0
self.blink.refresh()
test_camera = self.blink.sync.cameras['foobar']
self.assertEqual(test_camera.clip,
'https://rest.test.{}/new.mp4'.format(BLINK_URL))
self.assertEqual(test_camera.thumbnail,
'https://rest.test.{}/new.jpg'.format(BLINK_URL))
def test_set_links(self, mock_sess):
"""Test the link set method."""
self.blink.sync.cameras = {'foobar': self.camera}
self.blink.network_id = 9999
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.http_req')
@mock.patch('blinkpy.blinkpy.api.http_req')
def test_backup_url(self, req, mock_sess):
"""Test backup login method."""
fake_req = Request('POST', 'http://wrong.url').prepare()
req.side_effect = [
mresp.mocked_session_send(fake_req),
{'authtoken': {'authtoken': 'foobar123'}}
{'authtoken': {'authtoken': 'foobar123'},
'networks': {'1234': {'name': 'foobar', 'onboarded': True}}}
]
self.blink.get_auth_token()
self.assertEqual(self.blink.region_id, 'piri')
tests/test_blink_setup.py → tests/test_blinkpy.py
+37 -12
View File
@@ -8,7 +8,8 @@ any communication related errors at startup.
import unittest
from unittest import mock
from blinkpy import blinkpy
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,
@@ -26,10 +27,12 @@ class TestBlinkSetup(unittest.TestCase):
def setUp(self):
"""Set up Blink module."""
self.blink_no_cred = blinkpy.Blink()
self.blink = blinkpy.Blink(username=USERNAME,
password=PASSWORD)
self.blink.sync = BlinkSyncModule(self.blink, dict(), self.blink.urls)
self.blink_no_cred = Blink()
self.blink = Blink(username=USERNAME,
password=PASSWORD)
self.blink.sync = BlinkSyncModule(self.blink)
self.blink.urls = BlinkURLHandler('test')
self.blink.session = create_session()
def tearDown(self):
"""Clean up after test."""
@@ -59,8 +62,6 @@ class TestBlinkSetup(unittest.TestCase):
self.blink.urls = BlinkURLHandler(region_id)
with self.assertRaises(BlinkException):
self.blink.get_ids()
with self.assertRaises(BlinkException):
self.blink.summary_request()
@mock.patch('blinkpy.blinkpy.getpass.getpass')
def test_manual_login(self, getpwd, mock_sess):
@@ -95,8 +96,32 @@ class TestBlinkSetup(unittest.TestCase):
bad_header = {'Host': self.blink._host, 'TOKEN_AUTH': 'BADTOKEN'}
# pylint: disable=protected-access
self.blink._auth_header = bad_header
# pylint: disable=protected-access
self.assertEqual(self.blink._auth_header, bad_header)
self.blink.summary_request()
# pylint: disable=protected-access
self.assertEqual(self.blink._auth_header, original_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')
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}]
}
self.blink.networks = {'0000': {'onboarded': False},
'5678': {'onboarded': True},
'1234': {'onboarded': False}}
self.blink.get_ids()
self.assertEqual(self.blink.network_id, '5678')
self.assertEqual(self.blink.account_id, 2222)
@mock.patch('blinkpy.blinkpy.time.time')
def test_throttle(self, mock_time, mock_sess):
"""Check throttling functionality."""
now = self.blink.refresh_rate + 1
mock_time.return_value = now
self.assertEqual(self.blink.last_refresh, None)
result = self.blink.check_if_ok_to_update()
self.assertEqual(self.blink.last_refresh, now)
self.assertEqual(result, True)
self.assertEqual(self.blink.check_if_ok_to_update(), False)
self.assertEqual(self.blink.last_refresh, now)
tests/test_cameras.py
+144 -158
View File
@@ -11,8 +11,7 @@ from unittest import mock
from blinkpy import blinkpy
from blinkpy.helpers.util import create_session, BlinkURLHandler
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.camera import BlinkCamera
from blinkpy.helpers.constants import BLINK_URL
from blinkpy.camera import BlinkCamera, MAX_CLIPS
import tests.mock_responses as mresp
USERNAME = 'foobar'
@@ -39,177 +38,164 @@ class TestBlinkCameraSetup(unittest.TestCase):
"""Set up Blink module."""
self.blink = blinkpy.Blink(username=USERNAME,
password=PASSWORD)
self.camera_config = {
'device_id': 1111,
'name': 'foobar',
'armed': False,
'active': 'disarmed',
'thumbnail': '/test/image',
'video': '/test/clip/clip.mp4',
'temp': 70,
'battery': 3,
'notifications': 2,
'region_id': 'test'
}
header = {
'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.network_id = '0000'
self.sync = BlinkSyncModule(self.blink, header, self.blink.urls)
self.blink.sync = BlinkSyncModule(self.blink)
self.camera = BlinkCamera(self.blink.sync)
self.camera.name = 'foobar'
self.blink.sync.cameras['foobar'] = self.camera
def tearDown(self):
"""Clean up after test."""
self.blink = None
@mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request',
return_value=CAMERA_CFG)
def test_camera_properties(self, mock_cfg, mock_sess):
"""Tests all property set/recall."""
self.blink.urls = BlinkURLHandler('test')
def test_check_for_motion(self, mock_sess):
"""Test check for motion function."""
self.assertEqual(self.camera.last_record, [])
self.assertEqual(self.camera.motion_detected, None)
self.camera.sync.record_dates = {'foobar': [1, 3, 2, 4]}
self.camera.check_for_motion()
self.assertEqual(self.camera.last_record, [4])
self.assertEqual(self.camera.motion_detected, False)
self.camera.sync.record_dates = {'foobar': [7, 1, 3, 4]}
self.camera.check_for_motion()
self.assertEqual(self.camera.last_record, [7, 4])
self.assertEqual(self.camera.motion_detected, True)
self.camera.check_for_motion()
self.assertEqual(self.camera.last_record, [7, 4])
self.assertEqual(self.camera.motion_detected, False)
self.sync.cameras = {
'foobar': BlinkCamera(self.camera_config, self.sync)
def test_max_motion_clips(self, mock_sess):
"""Test that we only maintain certain number of records."""
for i in range(0, MAX_CLIPS):
self.camera.last_record.append(i)
self.camera.sync.record_dates['foobar'] = [MAX_CLIPS+2]
self.assertEqual(len(self.camera.last_record), MAX_CLIPS)
self.camera.check_for_motion()
self.assertEqual(self.camera.motion_detected, True)
self.assertEqual(len(self.camera.last_record), MAX_CLIPS)
def test_camera_update(self, mock_sess):
"""Test that we can properly update camera properties."""
config = {
'name': 'new',
'camera_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.sync.all_clips = {'new': {'1': '/test.mp4'}}
mock_sess.side_effect = [
'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.motion_enabled, False)
self.assertEqual(self.camera.battery, 50)
self.assertEqual(self.camera.temperature, 68)
self.assertEqual(self.camera.temperature_c, 20)
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')
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')
self.assertEqual(camera.armed, False)
self.assertEqual(
camera.thumbnail,
"https://rest.test.{}/test/image.jpg".format(BLINK_URL)
)
self.assertEqual(
camera.clip,
"https://rest.test.{}/test/clip/clip.mp4".format(BLINK_URL)
)
self.assertEqual(camera.temperature, 68)
self.assertEqual(camera.temperature_c, 20.0)
self.assertEqual(camera.battery, 50)
self.assertEqual(camera.battery_string, "OK")
self.assertEqual(camera.notifications, 2)
self.assertEqual(camera.region_id, 'test')
self.assertEqual(camera.motion_enabled, True)
self.assertEqual(camera.wifi_strength, -30)
camera_config = self.camera_config
camera_config['active'] = 'armed'
camera_config['thumbnail'] = '/test2/image'
camera_config['video'] = '/test2/clip.mp4'
camera_config['temp'] = 60
camera_config['battery'] = 0
camera_config['notifications'] = 4
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(
camera.thumbnail,
"https://rest.test.{}/test2/image.jpg".format(BLINK_URL)
)
self.assertEqual(
camera.clip,
"https://rest.test.{}/test2/clip.mp4".format(BLINK_URL)
)
self.assertEqual(camera.temperature, 68)
self.assertEqual(camera.battery, 50)
self.assertEqual(camera.battery_string, "Low")
self.assertEqual(camera.notifications, 4)
camera_config['battery'] = -10
camera.update(camera_config, skip_cache=True)
self.assertEqual(camera.battery_string, "Unknown")
def test_camera_case(self, mock_sess):
"""Tests camera case sensitivity."""
camera_object = BlinkCamera(self.camera_config, self.sync)
self.sync.cameras['foobar'] = camera_object
self.assertEqual(camera_object, self.sync.cameras['fOoBaR'])
@mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request',
return_value=CAMERA_CFG)
def test_camera_attributes(self, mock_cfg, mock_sess):
"""Tests camera attributes."""
self.blink.urls = BlinkURLHandler('test')
self.sync.cameras = {
'foobar': BlinkCamera(self.camera_config, self.sync)
def test_thumbnail_not_in_info(self, mock_sess):
"""Test that we grab thumbanil if not in camera_info."""
mock_sess.side_effect = ['foobar', 'barfoo']
self.camera.last_record = ['1']
self.camera.sync.record_dates['new'] = ['1']
self.camera.sync.all_clips = {'new': {'1': '/test.mp4'}}
config = {
'name': 'new',
'camera_id': 1234,
'network_id': 5678,
'serial': '12345678',
'enabled': False,
'battery_voltage': 90,
'battery_state': 'ok',
'temperature': 68,
'wifi_strength': 4,
'thumbnail': '',
}
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')
self.assertEqual(camera_attr['name'], 'foobar')
self.assertEqual(camera_attr['armed'], False)
self.assertEqual(
camera_attr['thumbnail'],
"https://rest.test.{}/test/image.jpg".format(BLINK_URL)
)
self.assertEqual(
camera_attr['video'],
"https://rest.test.{}/test/clip/clip.mp4".format(BLINK_URL)
)
self.assertEqual(camera_attr['temperature'], 68)
self.assertEqual(camera_attr['temperature_c'], 20.0)
self.assertEqual(camera_attr['battery'], 50)
self.assertEqual(camera_attr['notifications'], 2)
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',
return_value='https://fake.url')
def test_camera_cache(self, img_refresh, mock_sess):
"""Tests camera cache."""
update_vals = {
'name': 'foobar',
'active': 'disabled',
'video': '/clip.mp4',
'thumbnail': '/image',
'battery': 3,
'notifications': 1,
self.camera.sync.homescreen = {
'devices': [
{'foo': 'bar'},
{'device_type': 'foobar'},
{'device_type': 'camera',
'name': 'new',
'thumbnail': '/new/thumb'}
]
}
self.sync.cameras = {
'foobar': BlinkCamera(self.camera_config, self.sync)
self.camera.update(config)
self.assertEqual(self.camera.thumbnail,
'https://rest.test.immedia-semi.com/new/thumb.jpg')
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']
self.camera.sync.record_dates['new'] = ['1']
self.camera.sync.all_clips = {'new': {'1': '/test.mp4'}}
config = {
'name': 'new',
'camera_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': []
}
with self.assertLogs() as logrecord:
self.camera.update(config)
self.assertEqual(self.camera.thumbnail, None)
self.assertEqual(
logrecord.output,
["ERROR:blinkpy.camera:Could not find thumbnail for camera new"]
)
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)
def test_no_video_clips(self, mock_sess):
"""Tests that we still proceed with camera setup with no videos."""
mock_sess.return_value = 'foobar'
config = {
'name': 'new',
'camera_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)
tests/test_sync_module.py
+58 -65
View File
@@ -10,28 +10,7 @@ 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
@mock.patch('blinkpy.api.http_req')
class TestBlinkSyncModule(unittest.TestCase):
"""Test BlinkSyncModule functions in blinkpy."""
@@ -45,63 +24,77 @@ class TestBlinkSyncModule(unittest.TestCase):
'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)
self.blink.sync = BlinkSyncModule(self.blink)
self.camera = BlinkCamera(self.blink.sync)
def tearDown(self):
"""Clean up after test."""
self.blink = None
self.config = {}
self.camera = None
def test_get_videos(self):
def test_get_events(self, mock_resp):
"""Test get events function."""
mock_resp.return_value = {'event': True}
self.assertEqual(self.blink.sync.get_events(), True)
def test_get_camera_info(self, mock_resp):
"""Test get camera info function."""
mock_resp.return_value = {'devicestatus': True}
self.assertEqual(self.blink.sync.get_camera_info(), True)
def test_get_videos_one_page(self, mock_resp):
"""Test video access."""
self.blink.sync.return_value = [
mock_resp.return_value = [
{
'camera_name': 'foobar',
'address': '/new/test.mp4',
'address': '/test/clip_1900_01_01_12_00_00AM.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')
expected_videos = {'foobar': [
{'clip': '/test/clip_1900_01_01_12_00_00AM.mp4',
'thumb': '/test/thumb'}]}
expected_records = {'foobar': ['1900_01_01_12_00_00AM']}
expected_clips = {'foobar': {
'1900_01_01_12_00_00AM': '/test/clip_1900_01_01_12_00_00AM.mp4'}}
self.blink.sync.get_videos(start_page=0, end_page=0)
self.assertEqual(self.blink.sync.videos, expected_videos)
self.assertEqual(self.blink.sync.record_dates, expected_records)
self.assertEqual(self.blink.sync.all_clips, expected_clips)
@mock.patch('blinkpy.blinkpy.Blink.refresh')
@mock.patch('blinkpy.sync_module.BlinkSyncModule._summary_request')
@mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request')
def test_get_cameras(self, vid_req, req, refresh):
"""Test camera extraction."""
refresh.return_value = True
req.return_value = {'devices': [self.config]}
vid_req.return_value = [
def test_get_videos_multi_page(self, mock_resp):
"""Test video access with multiple pages."""
mock_resp.return_value = [
{
'camera_name': 'foobar',
'address': '/new.mp4',
'thumbnail': '/new'
'camera_name': 'test',
'address': '/foo/bar_1900_01_01_12_00_00AM.mp4',
'thumbnail': '/foobar'
}
]
self.blink.sync.get_cameras()
self.assertTrue('foobar' in self.blink.sync.cameras)
self.blink.sync.get_videos()
self.assertEqual(mock_resp.call_count, 2)
mock_resp.reset_mock()
self.blink.sync.get_videos(start_page=0, end_page=9)
self.assertEqual(mock_resp.call_count, 10)
def test_video_count(self):
"""Test video count function."""
self.blink.sync.return_value = {'count': 1}
self.assertEqual(self.blink.sync.video_count, 1)
def test_sync_start(self, mock_resp):
"""Test sync start function."""
mock_resp.side_effect = [
{'syncmodule': {
'name': 'test',
'id': 1234,
'network_id': 5678,
'serial': '12345678',
'status': 'foobar'}},
{'event': True},
{},
{'devicestatus': {}},
None,
None
]
self.blink.sync.start()
self.assertEqual(self.blink.sync.name, 'test')
self.assertEqual(self.blink.sync.sync_id, 1234)
self.assertEqual(self.blink.sync.network_id, 5678)
self.assertEqual(self.blink.sync.serial, '12345678')
self.assertEqual(self.blink.sync.status, 'foobar')
tox.ini
+1 -1
View File
@@ -1,5 +1,5 @@
[tox]
envlist = build, py34, py35, py36, py37, lint
envlist = build, py35, py36, py37, py38, lint
skip_missing_interpreters = True
skipsdist = True