Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4e6957a772 | ||
|
|
e747fb1010 | ||
|
|
9d110cdd72 | ||
|
|
6fa0eacff3 | ||
|
|
7ce924f52e | ||
|
|
91b3c2a64b | ||
|
|
97579400b1 | ||
|
|
63b01dde14 | ||
|
|
d9b996ab02 | ||
|
|
3af105edf5 | ||
|
|
5057592b0d | ||
|
|
13bb819e07 | ||
|
|
eefa9252e5 | ||
|
|
c5b84ac357 | ||
|
|
787256ce66 | ||
|
|
2f1877267f | ||
|
|
9c34fc9281 | ||
|
|
1b9bba7c3d | ||
|
|
123867ecd8 | ||
|
|
26d900897b | ||
|
|
f7a5b5cc03 | ||
|
|
7fd822caf2 | ||
|
|
01c6f0b12b | ||
|
|
62c89a925e | ||
|
|
07e1759819 | ||
|
|
8fac6c5bc0 | ||
|
|
bd4307af58 | ||
|
|
bd6e3bb4e3 | ||
|
|
c2630a43e1 | ||
|
|
2e6fb97cc5 | ||
|
|
674dae4440 | ||
|
|
a1082812bb | ||
|
|
9d838fe4eb | ||
|
|
fe0aef69e4 | ||
|
|
1b1235210e | ||
|
|
451d725946 | ||
|
|
9d5ae9552a | ||
|
|
c3c558e934 | ||
|
|
a570413e66 | ||
|
|
3e23374b32 | ||
|
|
1c6a5eb6c6 | ||
|
|
40e9e4bd6c | ||
|
|
e1ff851068 | ||
|
|
8a4e507f22 | ||
|
|
cd7b5bfabc | ||
|
|
346882f1ff | ||
|
|
5bc2466da2 | ||
|
|
672f59ab4e | ||
|
|
ae26a9e371 | ||
|
|
37fc047174 | ||
|
|
6343965bbc | ||
|
|
411da60bb2 | ||
|
|
72c36d8381 | ||
|
|
b0974c1ab2 | ||
|
|
a76eac085a | ||
|
|
5319714a37 | ||
|
|
5d9105655f | ||
|
|
b98c03b96b | ||
|
|
ec703ae4b0 | ||
|
|
9ce09b4761 | ||
|
|
6d002fc8b0 | ||
|
|
7228340677 | ||
|
|
9335a37feb | ||
|
|
371348533f | ||
|
|
26f6defe60 | ||
|
|
52bc4af1f0 | ||
|
|
8c94305804 | ||
|
|
0eb57698bd | ||
|
|
7b0db80edd | ||
|
|
0e63e97a57 | ||
|
|
e4a21964d2 | ||
|
|
39ad37f23b | ||
|
|
34bc61e1c3 | ||
|
|
e6a7c367b3 | ||
|
|
66afd33b79 | ||
|
|
cf617d720e | ||
|
|
4f2eca08e0 | ||
|
|
2972780c18 | ||
|
|
9c3f3f2f98 | ||
|
|
0d83a7fcb1 | ||
|
|
80f4ca7c4f | ||
|
|
96e7fce304 | ||
|
|
13a358c07c | ||
|
|
389745533c | ||
|
|
65dd547a32 | ||
|
|
91a877da79 | ||
|
|
d357215e0e | ||
|
|
ff47dc4731 | ||
|
|
fedea74256 | ||
|
|
66a6701a66 | ||
|
|
5bdfd6da52 | ||
|
|
a8f90400b2 | ||
|
|
3348386a4c | ||
|
|
1b53c4a0dd | ||
|
|
24158638ef | ||
|
|
901cbfc9d1 | ||
|
|
9f85068a0d | ||
|
|
90b83544f4 | ||
|
|
1d8e8e53d9 | ||
|
|
46a0a433ec | ||
|
|
d38d30471a | ||
|
|
5aad8ea641 | ||
|
|
74f4bd7763 | ||
|
|
51dd69a51f | ||
|
|
950e7d4971 | ||
|
|
623b694c78 | ||
|
|
7162ea24cc | ||
|
|
968b77b8bd | ||
|
|
6e00b126d1 | ||
|
|
003ceb0944 | ||
|
|
ce6dec50cd | ||
|
|
87643ddcab | ||
|
|
017d8aa6c0 | ||
|
|
87b83465b3 | ||
|
|
a1900f0130 | ||
|
|
cf9ecff529 | ||
|
|
db3dee81e9 | ||
|
|
798032be08 | ||
|
|
d6834dca06 | ||
|
|
933cd7229a | ||
|
|
8e51a86971 | ||
|
|
8ab44041eb | ||
|
|
79191d757a | ||
|
|
7d01f54d73 | ||
|
|
63c0e11cc9 | ||
|
|
7347d03f93 | ||
|
|
fffb10d7e6 | ||
|
|
91b3de1444 | ||
|
|
8340f2f78a | ||
|
|
e65840d269 | ||
|
|
c245bc24a6 | ||
|
|
71ebc3dc8a | ||
|
|
7ade321dd3 | ||
|
|
19c0114eea | ||
|
|
968c0cf8fc | ||
|
|
f8d4874f2f | ||
|
|
261d6293dd | ||
|
|
988265fd9f | ||
|
|
ba03894e94 | ||
|
|
556c28d4b9 | ||
|
|
9d2a8c6b69 | ||
|
|
89b517f47e | ||
|
|
150016f543 | ||
|
|
5df599be15 | ||
|
|
0fd4b4b274 | ||
|
|
dd744053f4 | ||
|
|
be68c0d714 | ||
|
|
2a4b0d3f4b | ||
|
|
46dd002e5d | ||
|
|
04bc0ace3f | ||
|
|
fa6cc49343 | ||
|
|
231d191556 | ||
|
|
b6c8141042 | ||
|
|
ae209a37f7 | ||
|
|
0d07ae1b4c |
@@ -1,3 +1,4 @@
|
||||
.pytest_cache/*
|
||||
.cache/*
|
||||
.tox/*
|
||||
__pycache__/*
|
||||
@@ -8,3 +9,4 @@ htmlcov/*
|
||||
dist/*
|
||||
.sh
|
||||
build/*
|
||||
docs/_build
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
image: python
|
||||
|
||||
stages:
|
||||
- test
|
||||
|
||||
before_script:
|
||||
- curl -O https://bootstrap.pypa.io/get-pip.py
|
||||
- python get-pip.py
|
||||
- pip install tox
|
||||
|
||||
python35:
|
||||
image: python:3.5
|
||||
stage: test
|
||||
script: tox -e py35
|
||||
|
||||
python36:
|
||||
image: python:3.6
|
||||
stage: test
|
||||
script: tox -e py36
|
||||
|
||||
lint:
|
||||
image: python:3.6
|
||||
stage: test
|
||||
script: tox -e lint
|
||||
|
||||
@@ -1,2 +1,7 @@
|
||||
python:
|
||||
enabled: true
|
||||
ignore_file: setup.py
|
||||
ignored_directories:
|
||||
- docs
|
||||
|
||||
fail_on_violations: true
|
||||
|
||||
+11
-6
@@ -1,16 +1,21 @@
|
||||
sudo: required
|
||||
matrix:
|
||||
fast_finish: true
|
||||
include:
|
||||
- python: "3.4.2"
|
||||
env: TOXENV=py34
|
||||
- python: "3.4.2"
|
||||
- python: "3.6"
|
||||
env: TOXENV=lint
|
||||
- python: "3.5"
|
||||
- python: "3.5.3"
|
||||
env: TOXENV=py35
|
||||
- python: "3.6"
|
||||
env: TOXENV=py36
|
||||
- python: "3.6-dev"
|
||||
env: TOXENV=py36
|
||||
- python: "3.6"
|
||||
env: TOXENV=build
|
||||
- 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
|
||||
|
||||
+64
@@ -3,6 +3,70 @@ Changelog
|
||||
|
||||
A list of changes between each release
|
||||
|
||||
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)
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
- Added support for battery voltage level (fixes `#64 <https://github.com/fronzbot/blinkpy/issues/64>`_)
|
||||
- Added motion detection per camera
|
||||
- Added fully accessible camera configuration dict
|
||||
- Added celcius property to camera (fixes `#60 <https://github.com/fronzbot/blinkpy/issues/60>`_)
|
||||
|
||||
0.7.1 (2018-05-09)
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
- Fixed pip 10 import issue during setup (`@fronzbot <https://github.com/fronzbot/blinkpy/pull/61>`_)
|
||||
|
||||
0.7.0 (2018-02-08)
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
- Fixed style errors for bumped pydocstring and pylint versions
|
||||
- Changed Blink.cameras dictionary to be case-insensitive (fixes `#35 <https://github.com/fronzbot/blinkpy/issues/35>`_)
|
||||
- Changed api endpoint for video extraction (fixes `#35 <https://github.com/fronzbot/blinkpy/issues/35>`_ and `#41 <https://github.com/fronzbot/blinkpy/issues/41>`_)
|
||||
- Removed last_motion() function from Blink class
|
||||
- Refactored code for better organization
|
||||
- Moved some request calls out of @property methods (enables future CLI support)
|
||||
- Renamed get_summary() method to summary and changed to @property
|
||||
- Added ability to download most recent video clip
|
||||
- Improved camera arm/disarm handling (`@b10m <https://github.com/fronzbot/blinkpy/pull/50>`_)
|
||||
- Added authentication to ``login()`` function and deprecated ``setup_system()`` in favor of ``start()``
|
||||
- Added ``attributes`` dictionary to camera object
|
||||
|
||||
0.6.0 (2017-05-12)
|
||||
^^^^^^^^^^^^^^^^^^
|
||||
- Removed redundent properties that only called hidden variables
|
||||
|
||||
+54
-89
@@ -1,6 +1,7 @@
|
||||
blinkpy |Build Status| |Coverage Status|
|
||||
============
|
||||
blinkpy |Build Status| |Coverage Status| |Docs| |PyPi Version|
|
||||
================================================================
|
||||
A Python library for the Blink Camera system
|
||||
Only compatible with Python 3+
|
||||
|
||||
Disclaimer:
|
||||
~~~~~~~~~~~~~~~
|
||||
@@ -17,111 +18,75 @@ Installation
|
||||
================
|
||||
``pip3 install blinkpy``
|
||||
|
||||
Installing Development Version
|
||||
==================================
|
||||
To install the current development version, perform the following steps. Note that the following will create a blinkpy directory in your home area:
|
||||
|
||||
.. code:: bash
|
||||
|
||||
$ cd ~
|
||||
$ git clone https://github.com/fronzbot/blinkpy.git
|
||||
$ cd blinkpy
|
||||
$ rm -rf build dist
|
||||
$ python3 setup.py bdist_wheel
|
||||
$ pip3 install --upgrade dist/*.whl
|
||||
|
||||
|
||||
Purpose
|
||||
===========
|
||||
This library was built with the intention of allowing easy communication with Blink camera systems, specifically so I can add a module into homeassistant https://home-assistant.io
|
||||
|
||||
Usage
|
||||
=========
|
||||
In terms of usage, you just need to instantiate the module with a username and password
|
||||
The simplest way to use this package from a terminal is to call ``Blink.start()`` which will prompt for your Blink username and password and then log you in. Alternatively, you can instantiate the Blink class with a username and password, and call ``Blink.start()`` to login and setup without prompt, as shown below. In addition, http requests are throttled internally via use of the ``Blink.refresh_rate`` variable, which can be set at initialization and defaults to 30 seconds.
|
||||
|
||||
.. code:: python
|
||||
|
||||
import blinkpy
|
||||
from blinkpy import blinkpy
|
||||
blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD', refresh_rate=30)
|
||||
blink.start()
|
||||
|
||||
If you would like to log in without setting up the cameras or system, you can simply call the ``Blink.login()`` function which will prompt for a username and password and then authenticate with the server. This is useful if you want to avoid use of the ``start()`` function which simply acts as a wrapper for more targeted API methods.
|
||||
|
||||
Cameras are instantiated as individual ``BlinkCamera`` classes within a ``BlinkSyncModule`` instance. Note: currently the API only supports one sync module, but multiple sync modules are planned to be supported in the future.
|
||||
|
||||
The below code will display cameras and their available attributes:
|
||||
|
||||
.. code:: python
|
||||
|
||||
from blinkpy import blinkpy
|
||||
|
||||
blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD')
|
||||
blink.start()
|
||||
|
||||
for name, camera in blink.sync.cameras.items():
|
||||
print(name) # Name of the camera
|
||||
print(camera.attributes) # Print available attributes of camera
|
||||
|
||||
If you leave out either of those parameters, you need to call the login function which will prompt for your username and password
|
||||
The most recent images and videos can be accessed as a bytes-object via internal variables. These can be updated with calls to ``Blink.refresh()`` but will only make a request if motion has been detected or other changes have been found. This can be overridden with the ``force_cache`` flag, but this should be used for debugging only since it overrides the internal request throttling.
|
||||
|
||||
.. code:: python
|
||||
|
||||
import blinkpy
|
||||
blink = blinkpy.Blink()
|
||||
blink.login()
|
||||
|
||||
|
||||
Once the login information is entered, you can run the `setup_system()` function which will attempt to authenticate with Blink servers using your username and password, obtain network ids, and create a list of cameras.
|
||||
The cameras are of a BlinkCamera class, of which the following parameters can be used (the code below creates a Blink object and iterates through each camera found)
|
||||
|
||||
.. code:: python
|
||||
|
||||
import blinkpy
|
||||
|
||||
blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD')
|
||||
blink.setup_system()
|
||||
|
||||
for camera in blink.cameras:
|
||||
print(camera.name) # Name of the camera
|
||||
print(camera.id) # Integer id of the camera (assigned by Blink)
|
||||
print(camera.armed) # Whether the device is armed/disarmed (ie. detecting motion)
|
||||
print(camera.clip) # Link to last motion clip captured
|
||||
print(camera.thumbnail) # Link to current camera thumbnail
|
||||
print(camera.temperature) # Current camera temperature (not super accurate, but might be useful for someone)
|
||||
print(camera.battery) # Current battery level... I think the value ranges from 0-3, but not quite sure yet.
|
||||
print(camera.battery_string) # Gives battery level as a string ("OK" or "Low"). Returns "Unknown" if value is... well, unknown
|
||||
print(camera.notifications) # Number of unread notifications (ie. motion alerts that haven't been viewed)
|
||||
print(camera.motion) # Dictionary containing values for keys ['video', 'image', 'time']
|
||||
# which correspond to last motion recorded, thumbnail of last motion, and timestamp of last motion
|
||||
camera = blink.sync.camera['SOME CAMERA NAME']
|
||||
blink.refresh(force_cache=True) # force a cache update USE WITH CAUTION
|
||||
camera.image_from_cache.raw # bytes-like image object (jpg)
|
||||
camera.video_from_cache.raw # bytes-like video object (mp4)
|
||||
|
||||
|
||||
Class Descriptions
|
||||
===================
|
||||
.. code:: python
|
||||
|
||||
class Blink()
|
||||
|
||||
* ``Blink.cameras`` Returns a dictionary of ``BlinkCamera`` objects where the key corresponds to the camera name and the value is the actual BlinkCamera object.
|
||||
* ``Blink.network_id`` Returns the current network id.
|
||||
* ``Blink.account_id`` Returns the account id.
|
||||
* ``Blink.events`` Returns a list of events recorded by blink. This information will contain links to any motion caught by an armed camera..
|
||||
* ``Blink.online`` Returns online status of sync module (True = online, False = offline).
|
||||
* ``Blink.last_motion()`` Finds last motion information for each camera and stores it in the ``BlinkCamera.motion`` field.
|
||||
* ``Blink.arm`` Set to True to arm, False to disarm. Can be used to see the status of the system as well.
|
||||
* ``Blink.refresh()`` Forces a refresh of all camera information.
|
||||
* ``Blink.get_summary()`` Returns json formatted summary of the system.
|
||||
* ``Blink.get_cameras()`` Finds all cameras in the system and creates ``BlinkCamera`` objects to represent them.
|
||||
* ``Blink.set_links()`` Gives each BlinkCamera object the links needed to find recent images and videos.
|
||||
* ``Blink.login()`` Prompts user for login information.
|
||||
* ``Blink.get_auth_token()`` Uses login information to retrieve authorization token from Blink for further communication.
|
||||
* ``Blink.get_ids()`` Retrieves the network_id and account_id from Blink in order to access video and image pages on their server.
|
||||
* ``Blink.setup_system()`` A wrapper script that calls:
|
||||
.. code:: python
|
||||
|
||||
Blink.get_auth_token()
|
||||
Blink.get_ids()
|
||||
Blink.get_camers()
|
||||
Blink.set_links()
|
||||
The ``blinkpy`` api also allows for saving images and videos to a file and snapping a new picture from the camera remotely:
|
||||
|
||||
.. code:: python
|
||||
|
||||
class BlinkCamera(config, urls)
|
||||
|
||||
The ``BlinkCamera`` class expects to receive:
|
||||
camera = blink.sync.camera['SOME CAMERA NAME']
|
||||
camera.snap_picture() # Take a new picture with the camera
|
||||
blink.refresh() # Get new information from server
|
||||
camera.image_to_file('/local/path/for/image.jpg')
|
||||
camera.video_to_file('/local/path/for/video.mp4')
|
||||
|
||||
* A dictionary ``config`` that contains the camera name, device id, armed status, thumbnail url, camera temperature, camery battery level, number of notifications, and region id
|
||||
* A ``BlinkURLHandler`` object that contains all the links necessary for communication.
|
||||
|
||||
Ultimately, this class is just a wrapper for each individual camera in order to make communication with individual cameras less clunky. The following properties/methods are availiable (in addition to the ones mentioned earlier):
|
||||
|
||||
* ``BlinkCamera.snap_picture()`` Takes an image with the camera and saves it as the new thumbnail. The ``Blink.refresh()`` method should be called after this if you want to store the new thumbnail link.
|
||||
* ``BlinkCamera.set_motion_detect(enable=True/False)`` Sending True to this function will enable motion detection for the camera. Setting to False will disable motion detection.
|
||||
* ``BlinkCamera.image_to_file(path)`` This will write the current thumbnail to the location indicated in 'path'
|
||||
* ``BlinkCamera.image_refresh()`` Refreshes the current thumbnail.
|
||||
|
||||
.. code:: python
|
||||
|
||||
class BlinkURLHandler(region_id)
|
||||
|
||||
The ``BlinkURLHandler`` class expects to be initialized with the region id found in the ``Blink.get_auth_token()`` function. The class will then create the necessary links required for various communication.
|
||||
|
||||
.. |Build Status| image:: https://travis-ci.org/fronzbot/blinkpy.svg?branch=master
|
||||
.. |Build Status| image:: https://travis-ci.org/fronzbot/blinkpy.svg?branch=dev
|
||||
:target: https://travis-ci.org/fronzbot/blinkpy
|
||||
.. |Coverage Status| image:: https://coveralls.io/repos/github/fronzbot/blinkpy/badge.svg?branch=master
|
||||
:target: https://coveralls.io/github/fronzbot/blinkpy?branch=master
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
.. |Coverage Status| image:: https://coveralls.io/repos/github/fronzbot/blinkpy/badge.svg?branch=dev
|
||||
:target: https://coveralls.io/github/fronzbot/blinkpy?branch=dev
|
||||
.. |PyPi Version| image:: https://img.shields.io/pypi/v/blinkpy.svg
|
||||
:target: https://pypi.python.org/pypi/blinkpy
|
||||
.. |Docs| image:: https://readthedocs.org/projects/blinkpy/badge/?version=latest
|
||||
:target: http://blinkpy.readthedocs.io/en/latest/?badge=latest
|
||||
|
||||
@@ -1,5 +0,0 @@
|
||||
"""Init file for blinkpy."""
|
||||
from blinkpy.blinkpy import Blink
|
||||
from blinkpy.blinkpy import BlinkAuthenticationException
|
||||
from blinkpy.blinkpy import BlinkException
|
||||
from blinkpy.blinkpy import BlinkURLHandler
|
||||
-367
@@ -1,367 +0,0 @@
|
||||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
blinkpy by Kevin Fronczak - A Blink camera Python library.
|
||||
|
||||
https://github.com/fronzbot/blinkpy
|
||||
Original protocol hacking by MattTW :
|
||||
https://github.com/MattTW/BlinkMonitorProtocol
|
||||
Published under the MIT license - See LICENSE file for more details.
|
||||
"Blink Wire-Free HS Home Monitoring & Alert Systems" is a trademark
|
||||
owned by Immedia Inc., see www.blinkforhome.com for more information.
|
||||
I am in no way affiliated with Blink, nor Immedia Inc.
|
||||
"""
|
||||
|
||||
import json
|
||||
import getpass
|
||||
from shutil import copyfileobj
|
||||
import requests
|
||||
import helpers.errors as ERROR
|
||||
from helpers.constants import (BLINK_URL, LOGIN_URL,
|
||||
LOGIN_BACKUP_URL,
|
||||
DEFAULT_URL, ONLINE)
|
||||
|
||||
|
||||
def _attempt_reauthorization(blink):
|
||||
"""Attempt to refresh auth token and links."""
|
||||
headers = blink.get_auth_token()
|
||||
blink.set_links()
|
||||
return headers
|
||||
|
||||
|
||||
def _request(blink, url='http://google.com', data=None, headers=None,
|
||||
reqtype='get', stream=False, json_resp=True, is_retry=False):
|
||||
"""Wrapper function for request."""
|
||||
if reqtype == 'post':
|
||||
response = requests.post(url, headers=headers,
|
||||
data=data)
|
||||
elif reqtype == 'get':
|
||||
response = requests.get(url, headers=headers,
|
||||
stream=stream)
|
||||
else:
|
||||
raise BlinkException(ERROR.REQUEST)
|
||||
|
||||
if json_resp and 'code' in response.json():
|
||||
if is_retry:
|
||||
raise BlinkAuthenticationException(
|
||||
(response.json()['code'], response.json()['message']))
|
||||
else:
|
||||
headers = _attempt_reauthorization(blink)
|
||||
return _request(blink, url=url, data=data, headers=headers,
|
||||
reqtype=reqtype, stream=stream,
|
||||
json_resp=json_resp, is_retry=True)
|
||||
|
||||
if json_resp:
|
||||
return response.json()
|
||||
else:
|
||||
return response
|
||||
|
||||
|
||||
# pylint: disable=super-init-not-called
|
||||
class BlinkException(Exception):
|
||||
"""Class to throw general blink exception."""
|
||||
|
||||
def __init__(self, errcode):
|
||||
"""Initialize BlinkException."""
|
||||
self.errid = errcode[0]
|
||||
self.message = errcode[1]
|
||||
|
||||
|
||||
class BlinkAuthenticationException(BlinkException):
|
||||
"""Class to throw authentication exception."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class BlinkURLHandler(object):
|
||||
"""Class that handles Blink URLS."""
|
||||
|
||||
def __init__(self, region_id):
|
||||
"""Initialize the urls."""
|
||||
self.base_url = 'https://' + region_id + '.' + BLINK_URL
|
||||
self.home_url = self.base_url + '/homescreen'
|
||||
self.event_url = self.base_url + '/events/network/'
|
||||
self.network_url = self.base_url + '/network/'
|
||||
self.networks_url = self.base_url + '/networks'
|
||||
|
||||
|
||||
class BlinkCamera(object):
|
||||
"""Class to initialize individual camera."""
|
||||
|
||||
def __init__(self, config, blink):
|
||||
"""Initiailize BlinkCamera."""
|
||||
self.blink = blink
|
||||
self.urls = self.blink.urls
|
||||
self.id = str(config['device_id']) # pylint: disable=invalid-name
|
||||
self.name = config['name']
|
||||
self._status = config['armed']
|
||||
self.thumbnail = self.urls.base_url + config['thumbnail'] + '.jpg'
|
||||
self.clip = self.urls.base_url + config['thumbnail'] + '.mp4'
|
||||
self.temperature = config['temp']
|
||||
self.battery = config['battery']
|
||||
self.notifications = config['notifications']
|
||||
self.motion = {}
|
||||
self.header = None
|
||||
self.image_link = None
|
||||
self.arm_link = None
|
||||
self.region_id = config['region_id']
|
||||
|
||||
@property
|
||||
def armed(self):
|
||||
"""Return camera arm status."""
|
||||
return self._status
|
||||
|
||||
@property
|
||||
def battery_string(self):
|
||||
"""Return string indicating battery status."""
|
||||
if self.battery > 1 and self.battery <= 3:
|
||||
return "OK"
|
||||
elif self.battery >= 0:
|
||||
return "Low"
|
||||
else:
|
||||
return "Unknown"
|
||||
|
||||
def snap_picture(self):
|
||||
"""Take a picture with camera to create a new thumbnail."""
|
||||
_request(self.blink, url=self.image_link,
|
||||
headers=self.header, reqtype='post')
|
||||
|
||||
def set_motion_detect(self, enable):
|
||||
"""Set motion detection."""
|
||||
url = self.arm_link
|
||||
if enable:
|
||||
_request(self.blink, url=url + 'enable',
|
||||
headers=self.header, reqtype='post')
|
||||
else:
|
||||
_request(self.blink, url=url + 'disable',
|
||||
headers=self.header, reqtype='post')
|
||||
|
||||
def update(self, values):
|
||||
"""Update camera information."""
|
||||
self.name = values['name']
|
||||
self._status = values['armed']
|
||||
self.thumbnail = self.urls.base_url + values['thumbnail'] + '.jpg'
|
||||
self.clip = self.urls.base_url + values['thumbnail'] + '.mp4'
|
||||
self.temperature = values['temp']
|
||||
self.battery = values['battery']
|
||||
self.notifications = values['notifications']
|
||||
|
||||
def image_refresh(self):
|
||||
"""Refresh current thumbnail."""
|
||||
url = self.urls.home_url
|
||||
response = _request(self.blink, url=url, headers=self.header,
|
||||
reqtype='get')['devices']
|
||||
for element in response:
|
||||
try:
|
||||
if str(element['device_id']) == self.id:
|
||||
self.thumbnail = (self.urls.base_url +
|
||||
element['thumbnail'] + '.jpg')
|
||||
return self.thumbnail
|
||||
except KeyError:
|
||||
pass
|
||||
return None
|
||||
|
||||
def image_to_file(self, path):
|
||||
"""Write image to file."""
|
||||
thumb = self.image_refresh()
|
||||
response = _request(self.blink, url=thumb, headers=self.header,
|
||||
reqtype='get', stream=True, json_resp=False)
|
||||
if response.status_code == 200:
|
||||
with open(path, 'wb') as imgfile:
|
||||
copyfileobj(response.raw, imgfile)
|
||||
|
||||
|
||||
class Blink(object):
|
||||
"""Class to initialize communication and sync module."""
|
||||
|
||||
def __init__(self, username=None, password=None):
|
||||
"""Initialize Blink system."""
|
||||
self._username = username
|
||||
self._password = password
|
||||
self._token = None
|
||||
self._auth_header = None
|
||||
self.network_id = None
|
||||
self.account_id = None
|
||||
self.region = None
|
||||
self.region_id = None
|
||||
self._host = None
|
||||
self._events = []
|
||||
self.cameras = {}
|
||||
self._idlookup = {}
|
||||
self.urls = None
|
||||
|
||||
@property
|
||||
def camera_thumbs(self):
|
||||
"""Return camera thumbnails."""
|
||||
self.refresh()
|
||||
data = {}
|
||||
for name, camera in self.cameras.items():
|
||||
data[name] = camera.thumbnail
|
||||
|
||||
return data
|
||||
|
||||
@property
|
||||
def id_table(self):
|
||||
"""Return id/camera pairs."""
|
||||
return self._idlookup
|
||||
|
||||
@property
|
||||
def events(self):
|
||||
"""Get all events on server."""
|
||||
url = self.urls.event_url + self.network_id
|
||||
headers = self._auth_header
|
||||
self._events = _request(self, url=url, headers=headers,
|
||||
reqtype='get')['event']
|
||||
return self._events
|
||||
|
||||
@property
|
||||
def online(self):
|
||||
"""Return boolean system online status."""
|
||||
url = self.urls.network_url + self.network_id + '/syncmodules'
|
||||
headers = self._auth_header
|
||||
return ONLINE[_request(self, url=url, headers=headers,
|
||||
reqtype='get')['syncmodule']['status']]
|
||||
|
||||
def last_motion(self):
|
||||
"""Find last motion of each camera."""
|
||||
recent = self.events
|
||||
for element in recent:
|
||||
try:
|
||||
camera_id = str(element['camera_id'])
|
||||
camera_name = self.id_table[camera_id]
|
||||
camera = self.cameras[camera_name]
|
||||
if element['type'] == 'motion':
|
||||
url = self.urls.base_url + element['video_url']
|
||||
camera.motion = {'video': url,
|
||||
'image': url[:-3] + 'jpg',
|
||||
'time': element['created_at']}
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
@property
|
||||
def arm(self):
|
||||
"""Return status of sync module: armed/disarmed."""
|
||||
return self.get_summary()['network']['armed']
|
||||
|
||||
@arm.setter
|
||||
def arm(self, value):
|
||||
"""Arm or disarm system."""
|
||||
if value:
|
||||
value_to_append = 'arm'
|
||||
else:
|
||||
value_to_append = 'disarm'
|
||||
url = self.urls.network_url + self.network_id + '/' + value_to_append
|
||||
_request(self, url=url, headers=self._auth_header, reqtype='post')
|
||||
|
||||
def refresh(self):
|
||||
"""Get all blink cameras and pulls their most recent status."""
|
||||
response = self.get_summary()['devices']
|
||||
|
||||
for name in self.cameras:
|
||||
camera = self.cameras[name]
|
||||
for element in response:
|
||||
try:
|
||||
if str(element['device_id']) == camera.id:
|
||||
camera.update(element)
|
||||
except KeyError:
|
||||
pass
|
||||
return None
|
||||
|
||||
def get_summary(self):
|
||||
"""Get a full summary of device information."""
|
||||
url = self.urls.home_url
|
||||
headers = self._auth_header
|
||||
|
||||
if self._auth_header is None:
|
||||
raise BlinkException(ERROR.AUTH_TOKEN)
|
||||
|
||||
return _request(self, url=url, headers=headers, reqtype='get')
|
||||
|
||||
def get_cameras(self):
|
||||
"""Find and creates cameras."""
|
||||
response = self.get_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
|
||||
device = BlinkCamera(element, self)
|
||||
self.cameras[device.name] = device
|
||||
self._idlookup[device.id] = device.name
|
||||
|
||||
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 = self.urls.network_url + self.network_id
|
||||
image_url = network_id_url + '/camera/' + camera.id + '/thumbnail'
|
||||
arm_url = network_id_url + '/camera/' + camera.id + '/'
|
||||
camera.image_link = image_url
|
||||
camera.arm_link = arm_url
|
||||
camera.header = self._auth_header
|
||||
|
||||
def setup_system(self):
|
||||
"""
|
||||
Wrapper for various setup functions.
|
||||
|
||||
Method logs in and sets auth token, urls, and ids for future requests.
|
||||
"""
|
||||
if self._username is None or self._password is None:
|
||||
raise BlinkAuthenticationException(ERROR.AUTHENTICATE)
|
||||
|
||||
self.get_auth_token()
|
||||
self.get_ids()
|
||||
self.get_cameras()
|
||||
self.set_links()
|
||||
|
||||
def login(self):
|
||||
"""Prompt user for username and password."""
|
||||
self._username = input("Username:")
|
||||
self._password = getpass.getpass("Password:")
|
||||
|
||||
def get_auth_token(self):
|
||||
"""Retrieve the authentication token from Blink."""
|
||||
if not isinstance(self._username, str):
|
||||
raise BlinkAuthenticationException(ERROR.USERNAME)
|
||||
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"
|
||||
})
|
||||
response = _request(self, url=LOGIN_URL, headers=headers,
|
||||
data=data, json_resp=False, reqtype='post')
|
||||
if response.status_code is 200:
|
||||
response = response.json()
|
||||
(self.region_id, self.region), = response['region'].items()
|
||||
else:
|
||||
response = _request(self, url=LOGIN_BACKUP_URL, headers=headers,
|
||||
data=data, reqtype='post')
|
||||
self.region_id = 'rest.piri'
|
||||
self.region = "UNKNOWN"
|
||||
|
||||
self._host = self.region_id + '.' + BLINK_URL
|
||||
self._token = response['authtoken']['authtoken']
|
||||
|
||||
self._auth_header = {'Host': self._host,
|
||||
'TOKEN_AUTH': self._token}
|
||||
|
||||
self.urls = BlinkURLHandler(self.region_id)
|
||||
|
||||
return self._auth_header
|
||||
|
||||
def get_ids(self):
|
||||
"""Set the network ID and Account ID."""
|
||||
url = self.urls.networks_url
|
||||
headers = self._auth_header
|
||||
|
||||
if self._auth_header is None:
|
||||
raise BlinkException(ERROR.AUTH_TOKEN)
|
||||
|
||||
response = _request(self, url=url, headers=headers, reqtype='get')
|
||||
self.network_id = str(response['networks'][0]['id'])
|
||||
self.account_id = str(response['networks'][0]['account_id'])
|
||||
@@ -0,0 +1 @@
|
||||
"""Init file for blinkpy."""
|
||||
+159
@@ -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')
|
||||
@@ -0,0 +1,171 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
blinkpy by Kevin Fronczak - A Blink camera Python library.
|
||||
|
||||
https://github.com/fronzbot/blinkpy
|
||||
Original protocol hacking by MattTW :
|
||||
https://github.com/MattTW/BlinkMonitorProtocol
|
||||
Published under the MIT license - See LICENSE file for more details.
|
||||
"Blink Wire-Free HS Home Monitoring & Alert Systems" is a trademark
|
||||
owned by Immedia Inc., see www.blinkforhome.com for more information.
|
||||
I am in no way affiliated with Blink, nor Immedia Inc.
|
||||
"""
|
||||
|
||||
import time
|
||||
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 (
|
||||
create_session, BlinkURLHandler,
|
||||
BlinkAuthenticationException)
|
||||
from blinkpy.helpers.constants import (
|
||||
BLINK_URL, LOGIN_URL, LOGIN_BACKUP_URL)
|
||||
|
||||
REFRESH_RATE = 30
|
||||
|
||||
_LOGGER = logging.getLogger('blinkpy')
|
||||
|
||||
|
||||
class Blink():
|
||||
"""Class to initialize communication."""
|
||||
|
||||
def __init__(self, username=None, password=None,
|
||||
refresh_rate=REFRESH_RATE):
|
||||
"""
|
||||
Initialize Blink system.
|
||||
|
||||
:param username: Blink username (usually email address)
|
||||
:param password: Blink password
|
||||
:param refresh_rate: Refresh rate of blink information.
|
||||
Defaults to 15 (seconds)
|
||||
"""
|
||||
self._username = username
|
||||
self._password = password
|
||||
self._token = None
|
||||
self._auth_header = None
|
||||
self._host = None
|
||||
self.network_id = None
|
||||
self.account_id = None
|
||||
self.urls = None
|
||||
self.sync = None
|
||||
self.region = None
|
||||
self.region_id = None
|
||||
self.last_refresh = None
|
||||
self.refresh_rate = refresh_rate
|
||||
self.session = None
|
||||
self.networks = []
|
||||
self._login_url = LOGIN_URL
|
||||
|
||||
@property
|
||||
def auth_header(self):
|
||||
"""Return the authentication header."""
|
||||
return self._auth_header
|
||||
|
||||
def start(self):
|
||||
"""
|
||||
Perform full system setup.
|
||||
|
||||
Method logs in and sets auth token, urls, and ids for future requests.
|
||||
Essentially this is just a wrapper function for ease of use.
|
||||
"""
|
||||
if self._username is None or self._password is None:
|
||||
self.login()
|
||||
else:
|
||||
self.get_auth_token()
|
||||
|
||||
self.get_ids()
|
||||
self.sync = BlinkSyncModule(self)
|
||||
self.sync.start()
|
||||
|
||||
def login(self):
|
||||
"""Prompt user for username and password."""
|
||||
self._username = input("Username:")
|
||||
self._password = getpass.getpass("Password:")
|
||||
if self.get_auth_token():
|
||||
_LOGGER.info("Login successful!")
|
||||
return True
|
||||
_LOGGER.warning("Unable to login with %s.", self._username)
|
||||
return False
|
||||
|
||||
def get_auth_token(self):
|
||||
"""Retrieve the authentication token from Blink."""
|
||||
if not isinstance(self._username, str):
|
||||
raise BlinkAuthenticationException(ERROR.USERNAME)
|
||||
if not isinstance(self._password, str):
|
||||
raise BlinkAuthenticationException(ERROR.PASSWORD)
|
||||
|
||||
login_url = LOGIN_URL
|
||||
self.session = create_session()
|
||||
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()
|
||||
else:
|
||||
_LOGGER.debug(
|
||||
("Received response code %s "
|
||||
"when authenticating, "
|
||||
"trying new url"), response.status_code
|
||||
)
|
||||
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 = 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):
|
||||
"""
|
||||
Perform a system refresh.
|
||||
|
||||
:param force_cache: Force an update of the camera cache
|
||||
"""
|
||||
if self.check_if_ok_to_update() or force_cache:
|
||||
_LOGGER.debug("Attempting refresh of cameras.")
|
||||
self.sync.refresh(force_cache=force_cache)
|
||||
|
||||
def check_if_ok_to_update(self):
|
||||
"""Check if it is ok to perform an http request."""
|
||||
current_time = int(time.time())
|
||||
last_refresh = self.last_refresh
|
||||
if last_refresh is None:
|
||||
last_refresh = 0
|
||||
if current_time >= (last_refresh + self.refresh_rate):
|
||||
self.last_refresh = current_time
|
||||
return True
|
||||
return False
|
||||
@@ -0,0 +1,205 @@
|
||||
"""Defines Blink cameras."""
|
||||
|
||||
from shutil import copyfileobj
|
||||
import logging
|
||||
from blinkpy import api
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
MAX_CLIPS = 5
|
||||
|
||||
|
||||
class BlinkCamera():
|
||||
"""Class to initialize individual camera."""
|
||||
|
||||
def __init__(self, sync):
|
||||
"""Initiailize BlinkCamera."""
|
||||
self.sync = sync
|
||||
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.last_record = []
|
||||
self._cached_image = None
|
||||
self._cached_video = None
|
||||
|
||||
@property
|
||||
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,
|
||||
'battery': self.battery,
|
||||
'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,
|
||||
'last_record': self.last_record
|
||||
}
|
||||
return attributes
|
||||
|
||||
@property
|
||||
def battery(self):
|
||||
"""Return battery level as percentage."""
|
||||
return round(self.battery_voltage / 180 * 100)
|
||||
|
||||
@property
|
||||
def temperature_c(self):
|
||||
"""Return temperature in celcius."""
|
||||
return round((self.temperature - 32) / 9.0 * 5.0, 1)
|
||||
|
||||
@property
|
||||
def image_from_cache(self):
|
||||
"""Return the most recently cached image."""
|
||||
if self._cached_image:
|
||||
return self._cached_image
|
||||
return None
|
||||
|
||||
@property
|
||||
def video_from_cache(self):
|
||||
"""Return the most recently cached video."""
|
||||
if self._cached_video:
|
||||
return self._cached_video
|
||||
return None
|
||||
|
||||
def snap_picture(self):
|
||||
"""Take a picture with camera to create a new thumbnail."""
|
||||
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)
|
||||
|
||||
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:
|
||||
thumb_addr = self.get_thumb_from_homescreen()
|
||||
|
||||
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
|
||||
|
||||
update_cached_video = False
|
||||
if self._cached_video is None or self.motion_detected:
|
||||
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)
|
||||
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()
|
||||
if new_clip not in self.last_record and self.last_record:
|
||||
self.motion_detected = True
|
||||
self.last_record.insert(0, new_clip)
|
||||
if len(self.last_record) > MAX_CLIPS:
|
||||
self.last_record.pop()
|
||||
elif not self.last_record:
|
||||
self.last_record.insert(0, new_clip)
|
||||
self.motion_detected = False
|
||||
else:
|
||||
self.motion_detected = False
|
||||
except KeyError:
|
||||
_LOGGER.warning("Could not extract clip info from camera %s",
|
||||
self.name)
|
||||
|
||||
def image_to_file(self, path):
|
||||
"""
|
||||
Write image to file.
|
||||
|
||||
:param path: Path to write file
|
||||
"""
|
||||
_LOGGER.debug("Writing image from %s to %s", self.name, path)
|
||||
response = self._cached_image
|
||||
if response.status_code == 200:
|
||||
with open(path, 'wb') as imgfile:
|
||||
copyfileobj(response.raw, imgfile)
|
||||
else:
|
||||
_LOGGER.error("Cannot write image to file, response %s",
|
||||
response.status_code)
|
||||
|
||||
def video_to_file(self, path):
|
||||
"""Write video to file.
|
||||
|
||||
:param path: Path to write file
|
||||
"""
|
||||
_LOGGER.debug("Writing video from %s to %s", self.name, path)
|
||||
response = self._cached_video
|
||||
with open(path, 'wb') as vidfile:
|
||||
copyfileobj(response.raw, vidfile)
|
||||
|
||||
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
|
||||
@@ -0,0 +1 @@
|
||||
"""Init file for blinkpy helper functions."""
|
||||
@@ -1,16 +1,14 @@
|
||||
'''
|
||||
constants.py
|
||||
Generates constants for use in blinkpy
|
||||
'''
|
||||
"""Generates constants for use in blinkpy."""
|
||||
|
||||
import os
|
||||
|
||||
MAJOR_VERSION = 0
|
||||
MINOR_VERSION = 6
|
||||
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'
|
||||
@@ -32,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'
|
||||
]
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
"""Module to define error types."""
|
||||
|
||||
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"
|
||||
)
|
||||
AUTH_TOKEN = (
|
||||
3,
|
||||
"Authentication header incorrect. Are you sure you received your token?"
|
||||
)
|
||||
REQUEST = (4, "Cannot perform request (get/post type incorrect)")
|
||||
@@ -0,0 +1,99 @@
|
||||
"""Useful functions for blinkpy."""
|
||||
|
||||
import logging
|
||||
from requests import Request, Session, exceptions
|
||||
from blinkpy.helpers.constants import BLINK_URL
|
||||
import blinkpy.helpers.errors as ERROR
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def create_session():
|
||||
"""Create a session for blink communication."""
|
||||
sess = Session()
|
||||
return sess
|
||||
|
||||
|
||||
def attempt_reauthorization(blink):
|
||||
"""Attempt to refresh auth token and links."""
|
||||
_LOGGER.debug("Auth token expired, attempting reauthorization.")
|
||||
headers = blink.get_auth_token()
|
||||
return headers
|
||||
|
||||
|
||||
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.
|
||||
|
||||
:param blink: Blink instance
|
||||
:param url: URL to perform request
|
||||
:param data: Data to send (default: None)
|
||||
:param headers: Headers to send (default: None)
|
||||
:param reqtype: Can be 'get' or 'post' (default: 'get')
|
||||
:param stream: Stream response? True/FALSE
|
||||
: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)
|
||||
else:
|
||||
_LOGGER.error("Invalid request type: %s", reqtype)
|
||||
raise BlinkException(ERROR.REQUEST)
|
||||
|
||||
prepped = req.prepare()
|
||||
|
||||
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:
|
||||
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)
|
||||
|
||||
if json_resp:
|
||||
return response.json()
|
||||
|
||||
return response
|
||||
|
||||
|
||||
class BlinkException(Exception):
|
||||
"""Class to throw general blink exception."""
|
||||
|
||||
def __init__(self, errcode):
|
||||
"""Initialize BlinkException."""
|
||||
super().__init__()
|
||||
self.errid = errcode[0]
|
||||
self.message = errcode[1]
|
||||
|
||||
|
||||
class BlinkAuthenticationException(BlinkException):
|
||||
"""Class to throw authentication exception."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class BlinkURLHandler():
|
||||
"""Class that handles Blink URLS."""
|
||||
|
||||
def __init__(self, region_id):
|
||||
"""Initialize the urls."""
|
||||
self.base_url = "https://rest.{}.{}".format(region_id, BLINK_URL)
|
||||
self.home_url = "{}/homescreen".format(self.base_url)
|
||||
self.event_url = "{}/events/network".format(self.base_url)
|
||||
self.network_url = "{}/network".format(self.base_url)
|
||||
self.networks_url = "{}/networks".format(self.base_url)
|
||||
self.video_url = "{}/api/v2/videos".format(self.base_url)
|
||||
_LOGGER.debug("Setting base url to %s.", self.base_url)
|
||||
@@ -0,0 +1,172 @@
|
||||
"""Defines a sync module for Blink."""
|
||||
|
||||
import logging
|
||||
|
||||
from requests.structures import CaseInsensitiveDict
|
||||
from blinkpy import api
|
||||
from blinkpy.camera import BlinkCamera
|
||||
from blinkpy.helpers.constants import ONLINE
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BlinkSyncModule():
|
||||
"""Class to initialize sync module."""
|
||||
|
||||
def __init__(self, blink):
|
||||
"""
|
||||
Initialize Blink sync module.
|
||||
|
||||
:param blink: Blink class instantiation
|
||||
"""
|
||||
self.blink = blink
|
||||
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.host = None
|
||||
self.summary = None
|
||||
self.homescreen = None
|
||||
self.record_dates = {}
|
||||
self.videos = {}
|
||||
self.events = []
|
||||
self.cameras = CaseInsensitiveDict({})
|
||||
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 online(self):
|
||||
"""Return boolean system online status."""
|
||||
return ONLINE[self.status]
|
||||
|
||||
@property
|
||||
def arm(self):
|
||||
"""Return status of sync module: armed/disarmed."""
|
||||
return self.homescreen['network']['armed']
|
||||
|
||||
@arm.setter
|
||||
def arm(self, value):
|
||||
"""Arm or disarm system."""
|
||||
if value:
|
||||
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."""
|
||||
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):
|
||||
"""
|
||||
Retrieve last recorded videos per camera.
|
||||
|
||||
:param start_page: Page to start reading from on blink servers
|
||||
(defaults to 0)
|
||||
:param end_page: Page to stop reading from (defaults to 1)
|
||||
"""
|
||||
videos = list()
|
||||
all_dates = dict()
|
||||
|
||||
for page_num in range(start_page, end_page + 1):
|
||||
this_page = api.request_videos(self.blink, page=page_num)
|
||||
if not this_page:
|
||||
break
|
||||
videos.append(this_page)
|
||||
|
||||
for page in videos:
|
||||
_LOGGER.debug("Retrieved video page %s", page)
|
||||
for entry in page:
|
||||
camera_name = entry['camera_name']
|
||||
clip_addr = entry['address']
|
||||
thumb_addr = entry['thumbnail']
|
||||
clip_date = clip_addr.split('_')[-6:]
|
||||
clip_date = '_'.join(clip_date)
|
||||
clip_date = clip_date.split('.')[0]
|
||||
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.videos[camera_name].append(
|
||||
{
|
||||
'clip': clip_addr,
|
||||
'thumb': thumb_addr,
|
||||
}
|
||||
)
|
||||
except KeyError:
|
||||
self.videos[camera_name] = [
|
||||
{
|
||||
'clip': clip_addr,
|
||||
'thumb': thumb_addr,
|
||||
}
|
||||
]
|
||||
self.record_dates = all_dates
|
||||
|
||||
return self.videos
|
||||
@@ -0,0 +1,20 @@
|
||||
# Minimal makefile for Sphinx documentation
|
||||
#
|
||||
|
||||
# You can set these variables from the command line.
|
||||
SPHINXOPTS =
|
||||
SPHINXBUILD = sphinx-build
|
||||
SPHINXPROJ = blinkpy
|
||||
SOURCEDIR = .
|
||||
BUILDDIR = _build
|
||||
|
||||
# Put it first so that "make" without argument is like "make help".
|
||||
help:
|
||||
@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
|
||||
|
||||
.PHONY: help Makefile
|
||||
|
||||
# Catch-all target: route all unknown targets to Sphinx using the new
|
||||
# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS).
|
||||
%: Makefile
|
||||
@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
|
||||
Symlink
+1
@@ -0,0 +1 @@
|
||||
../README.rst
|
||||
@@ -0,0 +1,30 @@
|
||||
.. _core_module:
|
||||
|
||||
:mod:`blinkpy`
|
||||
----------------------
|
||||
|
||||
.. automodule:: blinkpy.blinkpy
|
||||
|
||||
.. automodule:: blinkpy.sync_module
|
||||
|
||||
.. automodule:: blinkpy.camera
|
||||
|
||||
.. automodule:: blinkpy.helpers.util
|
||||
|
||||
.. autoclass:: blinkpy.blinkpy.Blink
|
||||
:members:
|
||||
|
||||
.. autoclass:: blinkpy.sync_module.BlinkSyncModule
|
||||
:members:
|
||||
|
||||
.. autoclass:: blinkpy.camera.BlinkCamera
|
||||
:members:
|
||||
|
||||
.. autoclass:: blinkpy.helpers.util.BlinkURLHandler
|
||||
:members:
|
||||
|
||||
.. autofunction:: blinkpy.helpers.util.create_session
|
||||
|
||||
.. autofunction:: blinkpy.helpers.util.http_req
|
||||
|
||||
.. autofunction:: blinkpy.helpers.util.attempt_reauthorization
|
||||
+177
@@ -0,0 +1,177 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# blinkpy documentation build configuration file, created by
|
||||
# sphinx-quickstart on Sat Jan 20 22:20:16 2018.
|
||||
#
|
||||
# This file is execfile()d with the current directory set to its
|
||||
# containing dir.
|
||||
#
|
||||
# Note that not all possible configuration values are present in this
|
||||
# autogenerated file.
|
||||
#
|
||||
# All configuration values have a default; values that are commented out
|
||||
# serve to show the default.
|
||||
|
||||
# If extensions (or modules to document with autodoc) are in another directory,
|
||||
# add these directories to sys.path here. If the directory is relative to the
|
||||
# documentation root, use os.path.abspath to make it absolute, like shown here.
|
||||
#
|
||||
# import os
|
||||
# import sys
|
||||
# sys.path.insert(0, os.path.abspath('.'))
|
||||
|
||||
|
||||
# -- General configuration ------------------------------------------------
|
||||
|
||||
# If your documentation needs a minimal Sphinx version, state it here.
|
||||
#
|
||||
# needs_sphinx = '1.0'
|
||||
|
||||
# Add any Sphinx extension module names here, as strings. They can be
|
||||
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
|
||||
# ones.
|
||||
import sphinx_rtd_theme
|
||||
from blinkpy.helpers.constants import __version__
|
||||
html_theme = "sphinx_rtd_theme"
|
||||
html_theme_path = [sphinx_rtd_theme.get_html_theme_path()]
|
||||
|
||||
extensions = ['sphinx.ext.autodoc',
|
||||
'sphinx.ext.doctest',
|
||||
'sphinx.ext.mathjax',
|
||||
'sphinx.ext.viewcode']
|
||||
|
||||
# Add any paths that contain templates here, relative to this directory.
|
||||
templates_path = ['_templates']
|
||||
|
||||
# The suffix(es) of source filenames.
|
||||
# You can specify multiple suffix as a list of string:
|
||||
#
|
||||
# source_suffix = ['.rst', '.md']
|
||||
source_suffix = '.rst'
|
||||
|
||||
# The master toctree document.
|
||||
master_doc = 'index'
|
||||
|
||||
# General information about the project.
|
||||
project = 'blinkpy'
|
||||
copyright = '2018, Kevin Fronczak'
|
||||
author = 'Kevin Fronczak'
|
||||
|
||||
# The version info for the project you're documenting, acts as replacement for
|
||||
# |version| and |release|, also used in various other places throughout the
|
||||
# built documents.
|
||||
#
|
||||
# The short X.Y version.
|
||||
version = __version__
|
||||
# The full version, including alpha/beta/rc tags.
|
||||
release = __version__
|
||||
|
||||
# The language for content autogenerated by Sphinx. Refer to documentation
|
||||
# for a list of supported languages.
|
||||
#
|
||||
# This is also used if you do content translation via gettext catalogs.
|
||||
# Usually you set "language" from the command line for these cases.
|
||||
language = None
|
||||
|
||||
# List of patterns, relative to source directory, that match files and
|
||||
# directories to ignore when looking for source files.
|
||||
# This patterns also effect to html_static_path and html_extra_path
|
||||
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
|
||||
|
||||
# The name of the Pygments (syntax highlighting) style to use.
|
||||
pygments_style = 'sphinx'
|
||||
|
||||
# If true, `todo` and `todoList` produce output, else they produce nothing.
|
||||
todo_include_todos = False
|
||||
|
||||
|
||||
# -- Options for HTML output ----------------------------------------------
|
||||
|
||||
# The theme to use for HTML and HTML Help pages. See the documentation for
|
||||
# a list of builtin themes.
|
||||
#
|
||||
#html_theme = 'alabaster'
|
||||
|
||||
# Theme options are theme-specific and customize the look and feel of a theme
|
||||
# further. For a list of options available for each theme, see the
|
||||
# documentation.
|
||||
#
|
||||
# html_theme_options = {}
|
||||
|
||||
# Add any paths that contain custom static files (such as style sheets) here,
|
||||
# relative to this directory. They are copied after the builtin static files,
|
||||
# so a file named "default.css" will overwrite the builtin "default.css".
|
||||
html_static_path = ['_static']
|
||||
|
||||
# Custom sidebar templates, must be a dictionary that maps document names
|
||||
# to template names.
|
||||
#
|
||||
# This is required for the alabaster theme
|
||||
# refs: http://alabaster.readthedocs.io/en/latest/installation.html#sidebars
|
||||
html_sidebars = {
|
||||
'**': [
|
||||
'relations.html', # needs 'show_related': True theme option to display
|
||||
'searchbox.html',
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
# -- Options for HTMLHelp output ------------------------------------------
|
||||
|
||||
# Output file base name for HTML help builder.
|
||||
htmlhelp_basename = 'blinkpydoc'
|
||||
|
||||
|
||||
# -- Options for LaTeX output ---------------------------------------------
|
||||
|
||||
latex_elements = {
|
||||
# The paper size ('letterpaper' or 'a4paper').
|
||||
#
|
||||
# 'papersize': 'letterpaper',
|
||||
|
||||
# The font size ('10pt', '11pt' or '12pt').
|
||||
#
|
||||
# 'pointsize': '10pt',
|
||||
|
||||
# Additional stuff for the LaTeX preamble.
|
||||
#
|
||||
# 'preamble': '',
|
||||
|
||||
# Latex figure (float) alignment
|
||||
#
|
||||
# 'figure_align': 'htbp',
|
||||
}
|
||||
|
||||
# Grouping the document tree into LaTeX files. List of tuples
|
||||
# (source start file, target name, title,
|
||||
# author, documentclass [howto, manual, or own class]).
|
||||
latex_documents = [
|
||||
(master_doc, 'blinkpy.tex', 'blinkpy Documentation',
|
||||
'Kevin Fronczak', 'manual'),
|
||||
]
|
||||
|
||||
|
||||
# -- Options for manual page output ---------------------------------------
|
||||
|
||||
# One entry per manual page. List of tuples
|
||||
# (source start file, name, description, authors, manual section).
|
||||
man_pages = [
|
||||
(master_doc, 'blinkpy', 'blinkpy Documentation',
|
||||
[author], 1)
|
||||
]
|
||||
|
||||
|
||||
# -- Options for Texinfo output -------------------------------------------
|
||||
|
||||
# Grouping the document tree into Texinfo files. List of tuples
|
||||
# (source start file, target name, title, author,
|
||||
# dir menu entry, description, category)
|
||||
texinfo_documents = [
|
||||
(master_doc, 'blinkpy', 'blinkpy Documentation',
|
||||
author, 'blinkpy', 'One line description of project.',
|
||||
'Miscellaneous'),
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
.. blinkpy documentation master file, created by
|
||||
sphinx-quickstart on Sat Jan 20 22:20:16 2018.
|
||||
You can adapt this file completely to your liking, but it should at least
|
||||
contain the root `toctree` directive.
|
||||
|
||||
Welcome to blinkpy's documentation!
|
||||
===================================
|
||||
|
||||
.. toctree::
|
||||
:maxdepth: 2
|
||||
:caption: Contents:
|
||||
:glob:
|
||||
|
||||
README
|
||||
api/*
|
||||
|
||||
Indices and tables
|
||||
==================
|
||||
|
||||
* :ref:`genindex`
|
||||
* :ref:`modindex`
|
||||
* :ref:`search`
|
||||
@@ -0,0 +1,36 @@
|
||||
@ECHO OFF
|
||||
|
||||
pushd %~dp0
|
||||
|
||||
REM Command file for Sphinx documentation
|
||||
|
||||
if "%SPHINXBUILD%" == "" (
|
||||
set SPHINXBUILD=sphinx-build
|
||||
)
|
||||
set SOURCEDIR=.
|
||||
set BUILDDIR=_build
|
||||
set SPHINXPROJ=blinkpy
|
||||
|
||||
if "%1" == "" goto help
|
||||
|
||||
%SPHINXBUILD% >NUL 2>NUL
|
||||
if errorlevel 9009 (
|
||||
echo.
|
||||
echo.The 'sphinx-build' command was not found. Make sure you have Sphinx
|
||||
echo.installed, then set the SPHINXBUILD environment variable to point
|
||||
echo.to the full path of the 'sphinx-build' executable. Alternatively you
|
||||
echo.may add the Sphinx directory to PATH.
|
||||
echo.
|
||||
echo.If you don't have Sphinx installed, grab it from
|
||||
echo.http://sphinx-doc.org/
|
||||
exit /b 1
|
||||
)
|
||||
|
||||
%SPHINXBUILD% -M %1 %SOURCEDIR% %BUILDDIR% %SPHINXOPTS%
|
||||
goto end
|
||||
|
||||
:help
|
||||
%SPHINXBUILD% -M help %SOURCEDIR% %BUILDDIR% %SPHINXOPTS%
|
||||
|
||||
:end
|
||||
popd
|
||||
@@ -1,3 +0,0 @@
|
||||
"""Init file for blinkpy helper functions."""
|
||||
from helpers import constants
|
||||
from helpers import errors
|
||||
@@ -1,6 +0,0 @@
|
||||
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")
|
||||
AUTH_TOKEN = (3, "Authentication header incorrect. Are you sure you logged in and received your token?")
|
||||
REQUEST = (4, "Cannot perform request (get/post type incorrect)")
|
||||
|
||||
@@ -7,6 +7,7 @@ reports=no
|
||||
# unused-argument - generic callbacks and setup methods create a lot of warnings
|
||||
# too-many-* - are not enforced for the sake of readability
|
||||
# too-few-* - same as too-many-*
|
||||
# no-else-return - I don't see any reason to enforce this. both forms are readable
|
||||
|
||||
disable=
|
||||
locally-disabled,
|
||||
@@ -20,4 +21,5 @@ disable=
|
||||
too-many-return-statements,
|
||||
too-many-statements,
|
||||
too-many-lines,
|
||||
too-few-public-methods,
|
||||
too-few-public-methods,
|
||||
no-else-return,
|
||||
|
||||
+1
-4
@@ -1,5 +1,2 @@
|
||||
pytest>=2.9.2
|
||||
pytest-cov>=2.3.1
|
||||
pytest-timeout>=1.0.0
|
||||
requests>=2.12.4
|
||||
testtools
|
||||
testtools
|
||||
|
||||
+10
-5
@@ -1,5 +1,10 @@
|
||||
flake8==3.3
|
||||
pylint==1.6.5
|
||||
pydocstyle==1.1.1
|
||||
pytest>=2.9.2
|
||||
pytest-sugar>=0.8.0
|
||||
flake8==3.5.0
|
||||
flake8-docstrings==1.3.0
|
||||
pylint==2.1.1
|
||||
pydocstyle==2.1.1
|
||||
pytest==3.7.1
|
||||
pytest-cov>=2.3.1
|
||||
pytest-sugar>=0.9.0
|
||||
pytest-timeout>=1.0.0
|
||||
restructuredtext-lint>=1.0.1
|
||||
pygments>=2.2.0
|
||||
|
||||
@@ -1,24 +1,48 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from os.path import abspath, dirname, join
|
||||
from setuptools import setup, find_packages
|
||||
from helpers.constants import (__version__, PROJECT_PACKAGE_NAME,
|
||||
PROJECT_LICENSE, PROJECT_URL,
|
||||
PROJECT_EMAIL, PROJECT_DESCRIPTION,
|
||||
PROJECT_CLASSIFIERS, PROJECT_AUTHOR,
|
||||
PROJECT_LONG_DESCRIPTION)
|
||||
# Fixes import error due to breaking change in pip 10
|
||||
# https://stackoverflow.com/a/49867265
|
||||
# pip > 10
|
||||
try:
|
||||
from pip._internal.req import parse_requirements
|
||||
# pip < 10
|
||||
except ImportError:
|
||||
from pip.req import parse_requirements
|
||||
from blinkpy.helpers.constants import (
|
||||
__version__, PROJECT_PACKAGE_NAME, PROJECT_LICENSE, PROJECT_URL,
|
||||
PROJECT_EMAIL, PROJECT_DESCRIPTION, PROJECT_CLASSIFIERS, PROJECT_AUTHOR,
|
||||
)
|
||||
|
||||
this_dir = abspath(dirname(__file__))
|
||||
|
||||
install_reqs = parse_requirements('{}/requirements.txt'.format(this_dir), session=False)
|
||||
test_reqs = parse_requirements('{}/requirements_test.txt'.format(this_dir), session=False)
|
||||
REQUIRES = [str(ir.req) for ir in install_reqs]
|
||||
TEST_REQUIRES = [str(ir.req) for ir in test_reqs]
|
||||
|
||||
PACKAGES = find_packages(exclude=['tests*', 'docs'])
|
||||
|
||||
with open('{}/README.rst'.format(this_dir), encoding='utf-8') as readme_file:
|
||||
long_description = readme_file.read()
|
||||
|
||||
name = PROJECT_PACKAGE_NAME
|
||||
version = __version__
|
||||
|
||||
setup(
|
||||
name = PROJECT_PACKAGE_NAME,
|
||||
version = __version__,
|
||||
description = PROJECT_DESCRIPTION,
|
||||
long_description = PROJECT_LONG_DESCRIPTION,
|
||||
long_description = long_description,
|
||||
author = PROJECT_AUTHOR,
|
||||
author_email = PROJECT_EMAIL,
|
||||
license = PROJECT_LICENSE,
|
||||
url = PROJECT_URL,
|
||||
platforms = 'any',
|
||||
py_modules = ['blinkpy'],
|
||||
packages=find_packages(),
|
||||
install_requires = ['requests>=2,<3'],
|
||||
packages=PACKAGES,
|
||||
include_package_data = True,
|
||||
install_requires = REQUIRES,
|
||||
test_suite = 'tests',
|
||||
classifiers = PROJECT_CLASSIFIERS
|
||||
)
|
||||
)
|
||||
|
||||
+56
-291
@@ -1,306 +1,71 @@
|
||||
"""
|
||||
Mock responses that mimic actual responses from Blink servers.
|
||||
"""Simple mock responses definitions."""
|
||||
|
||||
This file should be updated any time the Blink server responses
|
||||
change so we can make sure blinkpy can still communicate.
|
||||
"""
|
||||
from blinkpy.helpers.util import BlinkURLHandler
|
||||
import blinkpy.helpers.constants as const
|
||||
|
||||
import helpers.constants as const
|
||||
|
||||
NETWORKS_RESPONSE = {}
|
||||
NETWORKS_RESPONSE['summary'] = {'onboarded': True, 'name': 'Nilfgaard'}
|
||||
NETWORKS_RESPONSE['networks'] = [{
|
||||
'network_key': None,
|
||||
'name': NETWORKS_RESPONSE['summary']['name'],
|
||||
'account_id': 1989,
|
||||
'id': 9898,
|
||||
'encryption_key': None,
|
||||
'armed': True,
|
||||
'ping_interval': 60,
|
||||
'video_destination': 'server',
|
||||
'arm_string': 'Armed',
|
||||
'feature_plan_id': None
|
||||
}]
|
||||
|
||||
NEW_THUMBNAIL = '/NEW/THUMBNAIL/YAY'
|
||||
|
||||
FIRST_CAMERA = {'device_type': 'camera',
|
||||
'notifications': 1,
|
||||
'battery': 2,
|
||||
'active': 'enabled',
|
||||
'enabled': True,
|
||||
'temp': 70,
|
||||
'updated_at': '2017-01-01T01:23:45+00:00',
|
||||
'lfr_strength': 3,
|
||||
'armed': True,
|
||||
'device_id': 2112,
|
||||
'wifi_strength': 5,
|
||||
'thumbnail': '/this/is/url1/thumb',
|
||||
'name': 'First Camera',
|
||||
'status': 'done'}
|
||||
|
||||
SECOND_CAMERA = {'device_type': 'camera',
|
||||
'notifications': 0,
|
||||
'battery': 3,
|
||||
'active': 'disabled',
|
||||
'enabled': False,
|
||||
'temp': 82,
|
||||
'updated_at': '2017-12-21T01:21:12+00:00',
|
||||
'lfr_strength': 1,
|
||||
'armed': False,
|
||||
'device_id': 1221,
|
||||
'wifi_strength': 1,
|
||||
'thumbnail': '/this/is/url2/thumb',
|
||||
'name': 'Second Camera',
|
||||
'status': 'done'}
|
||||
|
||||
SYNC_MODULE = {'updated_at': '1970-01-01T01:00:00+00:00',
|
||||
'device_type': 'sync_module',
|
||||
'notifications': 2,
|
||||
'device_id': 7990,
|
||||
'status': 'online'}
|
||||
|
||||
FIRST_EVENT = {'camera_name': FIRST_CAMERA['name'],
|
||||
'updated_at': '2017-10-10T03:37:37+00:00',
|
||||
'sync_module_id': None,
|
||||
'camera': FIRST_CAMERA['device_id'],
|
||||
'type': 'motion',
|
||||
'duration': None,
|
||||
'status': None,
|
||||
'created_at': '2017-01-28T19:51:52+00:00',
|
||||
'camera_id': FIRST_CAMERA['device_id'],
|
||||
'id': 666777888,
|
||||
'siren_id': None,
|
||||
'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'],
|
||||
'notified': True,
|
||||
'siren': None,
|
||||
'syncmodule': None,
|
||||
'video_url': FIRST_CAMERA['thumbnail'] + '.mp4',
|
||||
'command_id': None,
|
||||
'network_id': None,
|
||||
'account': NETWORKS_RESPONSE['networks'][0]['account_id'],
|
||||
'video_id': 123000321}
|
||||
|
||||
SECOND_EVENT = {'updated_at': '2017-01-28T19:51:29+00:00',
|
||||
'sync_module_id': SYNC_MODULE['device_id'],
|
||||
'camera': None,
|
||||
'type': 'armed',
|
||||
'duration': None,
|
||||
'status': None,
|
||||
'created_at': '2017-01-28T19:51:29+00:00',
|
||||
'camera_id': None,
|
||||
'id': 333777555,
|
||||
'siren_id': None,
|
||||
'account_id': NETWORKS_RESPONSE['networks'][0]['account_id'],
|
||||
'notified': False,
|
||||
'siren': None,
|
||||
'syncmodule': SYNC_MODULE['device_id'],
|
||||
'command_id': None,
|
||||
'network_id': None,
|
||||
'account': NETWORKS_RESPONSE['networks'][0]['account_id']}
|
||||
|
||||
"""Fake response content."""
|
||||
LOGIN_RESPONSE = {}
|
||||
LOGIN_RESPONSE['region'] = {'ciri': 'Cintra'}
|
||||
LOGIN_RESPONSE['networks'] = {
|
||||
NETWORKS_RESPONSE['networks'][0]['id']: NETWORKS_RESPONSE['summary']
|
||||
LOGIN_RESPONSE = {
|
||||
'region': {'mock': 'Test'},
|
||||
'networks': {
|
||||
'1234': {'name': 'test', 'onboarded': True}
|
||||
},
|
||||
'authtoken': {'authtoken': 'foobar123', 'message': 'auth'}
|
||||
}
|
||||
LOGIN_RESPONSE['authtoken'] = {'authtoken': 'foobar7117', 'message': 'auth'}
|
||||
|
||||
RESPONSE = {}
|
||||
RESPONSE['account'] = {'notifications': SYNC_MODULE['notifications']}
|
||||
RESPONSE['devices'] = [FIRST_CAMERA, SECOND_CAMERA, SYNC_MODULE]
|
||||
RESPONSE['network'] = {'armed': NETWORKS_RESPONSE['networks'][0]['armed'],
|
||||
'wifi_strength': 4,
|
||||
'warning': 0,
|
||||
'name': NETWORKS_RESPONSE['summary']['name'],
|
||||
'notifications': SYNC_MODULE['notifications']}
|
||||
RESPONSE['event'] = [FIRST_EVENT, SECOND_EVENT]
|
||||
RESPONSE['syncmodule'] = {'name': 'Vengerberg', 'status': 'online'}
|
||||
|
||||
BAD_RESPONSE = {}
|
||||
BAD_RESPONSE['account'] = {'nothing': 'here'}
|
||||
BAD_RESPONSE['devices'] = [{'foo': 'bar', 'device_type': 'camera'},
|
||||
{'device_type': 'camera', 'device_id': 2112}]
|
||||
BAD_RESPONSE['network'] = {'bar': 'foo'}
|
||||
|
||||
MOCK_BYTES = '\x00\x10JFIF\x00\x01'
|
||||
|
||||
IMAGE_TO_WRITE_URL = list()
|
||||
IMAGE_TO_WRITE_URL.append('https://ciri.' + const.BLINK_URL +
|
||||
FIRST_CAMERA['thumbnail'] + '.jpg')
|
||||
IMAGE_TO_WRITE_URL.append('https://ciri.' + const.BLINK_URL +
|
||||
SECOND_CAMERA['thumbnail'] + '.jpg')
|
||||
|
||||
FAKE_FILES = list()
|
||||
|
||||
|
||||
def mocked_requests_post(*args, **kwargs):
|
||||
"""Mock post request."""
|
||||
class MockPostResponse:
|
||||
"""Class for mock post response."""
|
||||
class MockResponse:
|
||||
"""Class for mock request response."""
|
||||
|
||||
def __init__(self, json_data, status_code):
|
||||
"""Initialze mock post response."""
|
||||
self.json_data = json_data
|
||||
self.status_code = status_code
|
||||
def __init__(self, json_data, status_code, raw_data=None):
|
||||
"""Initialize mock get response."""
|
||||
self.json_data = json_data
|
||||
self.status_code = status_code
|
||||
self.raw_data = raw_data
|
||||
|
||||
def json(self):
|
||||
"""Return json data from post request."""
|
||||
return self.json_data
|
||||
def json(self):
|
||||
"""Return json data from get_request."""
|
||||
return self.json_data
|
||||
|
||||
# pylint: disable=global-variable-not-assigned
|
||||
global RESPONSE
|
||||
# pylint: disable=global-variable-not-assigned
|
||||
global NETWORKS_RESPONSE
|
||||
|
||||
if args[0] is not None:
|
||||
url_tail = args[0].split("/")[-1]
|
||||
else:
|
||||
return MockPostResponse({'message': 'ERROR', 'code': 404}, 404)
|
||||
|
||||
if args[0] == const.LOGIN_URL:
|
||||
# Request to login
|
||||
return MockPostResponse(LOGIN_RESPONSE, 200)
|
||||
elif args[0] == const.LOGIN_BACKUP_URL:
|
||||
return MockPostResponse(LOGIN_RESPONSE, 200)
|
||||
elif url_tail == 'arm' or url_tail == 'disarm':
|
||||
# Request to arm/disarm system
|
||||
NETWORKS_RESPONSE['networks'][0]['armed'] = url_tail == 'arm'
|
||||
RESPONSE['network']['armed'] = url_tail == 'arm'
|
||||
return MockPostResponse({}, 200)
|
||||
elif url_tail == 'enable' or url_tail == 'disable':
|
||||
# Request to enable/disable motion detection per camera
|
||||
received_id = args[0].split("/")[-2]
|
||||
all_devices = list()
|
||||
for element in RESPONSE['devices']:
|
||||
all_devices.append(element)
|
||||
current_id = element['device_id']
|
||||
if str(current_id) == received_id:
|
||||
element['armed'] = url_tail == 'enable'
|
||||
element['enabled'] = url_tail == 'enable'
|
||||
RESPONSE['devices'] = all_devices
|
||||
return MockPostResponse({}, 200)
|
||||
elif url_tail == 'thumbnail':
|
||||
# Requesting a new image
|
||||
received_id = args[0].split("/")[-2]
|
||||
all_devices = list()
|
||||
for element in RESPONSE['devices']:
|
||||
all_devices.append(element)
|
||||
if str(element['device_id']) == received_id:
|
||||
element['thumbnail'] = NEW_THUMBNAIL
|
||||
RESPONSE['devices'] = all_devices
|
||||
return MockPostResponse({}, 200)
|
||||
|
||||
return MockPostResponse({'message': 'ERROR', 'code': 404}, 404)
|
||||
@property
|
||||
def raw(self):
|
||||
"""Return raw data from get request."""
|
||||
return self.raw_data
|
||||
|
||||
|
||||
def mocked_requests_get(*args, **kwargs):
|
||||
"""Mock get request."""
|
||||
class MockGetResponse:
|
||||
"""Class for mock get response."""
|
||||
def mocked_session_send(*args, **kwargs):
|
||||
"""Mock session."""
|
||||
prepped = args[0]
|
||||
url = prepped.url
|
||||
header = prepped.headers
|
||||
method = prepped.method
|
||||
if method == 'GET':
|
||||
expected_token = LOGIN_RESPONSE['authtoken']['authtoken']
|
||||
if header['TOKEN_AUTH'] != expected_token:
|
||||
response = {'message': 'Not Authorized', 'code': 400}
|
||||
status = 400
|
||||
elif url == 'use_bad_response':
|
||||
response = {'foo': 'bar'}
|
||||
status = 200
|
||||
elif url == 'reauth':
|
||||
response = {'message': 'REAUTH', 'code': 777}
|
||||
status = 777
|
||||
else:
|
||||
response = {'test': 'foo'}
|
||||
status = 200
|
||||
elif method == 'POST':
|
||||
if url in (const.LOGIN_URL, const.LOGIN_BACKUP_URL):
|
||||
response = LOGIN_RESPONSE
|
||||
status = 200
|
||||
elif url == 'http://wrong.url/' or url is None:
|
||||
response = {'message': 'Error', 'code': 404}
|
||||
status = 404
|
||||
else:
|
||||
response = {'message': 'foo', 'code': 200}
|
||||
status = 200
|
||||
|
||||
def __init__(self, json_data, status_code, raw_data=None):
|
||||
"""Initialize mock get response."""
|
||||
self.json_data = json_data
|
||||
self.status_code = status_code
|
||||
self.raw_data = raw_data
|
||||
|
||||
def json(self):
|
||||
"""Return json data from get request."""
|
||||
return self.json_data
|
||||
|
||||
@property
|
||||
def raw(self):
|
||||
"""Return raw data from get request."""
|
||||
return self.raw_data
|
||||
rx_header = kwargs.pop('headers')
|
||||
expected_token = LOGIN_RESPONSE['authtoken']['authtoken']
|
||||
# pylint: disable=unused-variable
|
||||
if ('Content-Type' not in rx_header
|
||||
and rx_header['TOKEN_AUTH'] != expected_token):
|
||||
return MockGetResponse({'message': 'Not Authorized', 'code': 400}, 400)
|
||||
|
||||
(region_id, region), = LOGIN_RESPONSE['region'].items()
|
||||
if (args[0] != 'use_bad_response'
|
||||
and args[0] != 'reauth' and args[0] is not None):
|
||||
set_region_id = args[0].split('/')[2].split('.')[0]
|
||||
else:
|
||||
set_region_id = 'ciri'
|
||||
|
||||
if set_region_id == 'rest':
|
||||
set_region_id = (set_region_id + '.' +
|
||||
args[0].split('/')[2].split('.')[1])
|
||||
region_id = 'rest.piri'
|
||||
neturl = 'https://' + set_region_id + '.' + const.BLINK_URL + '/networks'
|
||||
if args[0] == neturl:
|
||||
return MockGetResponse(NETWORKS_RESPONSE, 200)
|
||||
elif set_region_id != region_id:
|
||||
raise ConnectionError('Received region id ' + region_id +
|
||||
' Expected ' + set_region_id)
|
||||
elif args[0] in IMAGE_TO_WRITE_URL:
|
||||
return MockGetResponse({}, 200, raw_data=MOCK_BYTES)
|
||||
elif args[0] == 'use_bad_response':
|
||||
return MockGetResponse(BAD_RESPONSE, 200)
|
||||
elif args[0] == 'reauth':
|
||||
return MockGetResponse({'message': 'REAUTH', 'code': 777}, 777)
|
||||
else:
|
||||
return MockGetResponse(RESPONSE, 200)
|
||||
|
||||
return MockGetResponse({'message': 'ERROR', 'code': 404}, 404)
|
||||
return MockResponse(response, status)
|
||||
|
||||
|
||||
def mocked_copyfileobj(*args, **kwargs):
|
||||
"""Mock shutil.copyfileobj."""
|
||||
class MockCopyFileObj:
|
||||
"""Class for mock copy file."""
|
||||
|
||||
def __init__(self, src, dst):
|
||||
"""Initialize copyfile mock."""
|
||||
self.src = src
|
||||
self.dst = dst
|
||||
# pylint: disable=global-variable-not-assigned
|
||||
global FAKE_FILES
|
||||
mockobj = MockCopyFileObj(args[0], args[1])
|
||||
FAKE_FILES.append(mockobj.src)
|
||||
return
|
||||
|
||||
|
||||
def get_test_cameras(base_url):
|
||||
"""Helper function to return cameras named in this file."""
|
||||
test_cameras = dict()
|
||||
for element in RESPONSE['devices']:
|
||||
if ('device_type' in element and
|
||||
element['device_type'] == 'camera'):
|
||||
test_cameras[element['name']] = {
|
||||
'device_id': str(element['device_id']),
|
||||
'armed': element['armed'],
|
||||
'thumbnail': (base_url +
|
||||
element['thumbnail'] + '.jpg'),
|
||||
'temperature': element['temp'],
|
||||
'battery': element['battery'],
|
||||
'notifications': element['notifications']
|
||||
}
|
||||
return test_cameras
|
||||
|
||||
|
||||
def get_test_id_table():
|
||||
"""Helper function to return mock id table."""
|
||||
test_id_table = dict()
|
||||
for element in RESPONSE['devices']:
|
||||
if ('device_type' in element and
|
||||
element['device_type'] == 'camera'):
|
||||
test_id_table[str(element['device_id'])] = element['name']
|
||||
return test_id_table
|
||||
|
||||
|
||||
class MockURLHandler(object):
|
||||
class MockURLHandler(BlinkURLHandler):
|
||||
"""Mocks URL Handler in blinkpy module."""
|
||||
|
||||
def __init__(self, region_id):
|
||||
"""Initialize the urls."""
|
||||
self.base_url = 'https://' + region_id + '.' + const.BLINK_URL
|
||||
self.home_url = 'reauth'
|
||||
self.event_url = self.base_url + '/events/network/'
|
||||
self.network_url = self.base_url + '/network/'
|
||||
self.networks_url = self.base_url + '/networks'
|
||||
pass
|
||||
|
||||
@@ -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])
|
||||
@@ -1,107 +0,0 @@
|
||||
"""
|
||||
Tests the camera initialization and attributes of
|
||||
individual BlinkCamera instantiations.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import blinkpy
|
||||
import tests.mock_responses as mresp
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
|
||||
|
||||
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)
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
self.blink = None
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_camera_properties(self, mock_get, mock_post):
|
||||
"""Tests all property set/recall."""
|
||||
test_value = 'foobar'
|
||||
test_region_id = list(mresp.LOGIN_RESPONSE['region'].keys())[0]
|
||||
self.blink.setup_system()
|
||||
for name in self.blink.cameras:
|
||||
camera = self.blink.cameras[name]
|
||||
camera.name = test_value
|
||||
camera.clip = test_value + '.mp4'
|
||||
camera.thumbnail = test_value + '.jpg'
|
||||
camera.temperature = 10
|
||||
camera.battery = 0
|
||||
camera.notifications = 100
|
||||
camera.image_link = test_value + '/image.jpg'
|
||||
camera.arm_link = test_value + '/arm'
|
||||
camera.header = {'foo': 'bar'}
|
||||
camera.motion = {'bar': 'foo'}
|
||||
self.assertEqual(camera.clip, test_value + '.mp4')
|
||||
self.assertEqual(camera.name, test_value)
|
||||
self.assertEqual(camera.thumbnail, test_value + '.jpg')
|
||||
self.assertEqual(camera.temperature, 10)
|
||||
self.assertEqual(camera.battery, 0)
|
||||
self.assertEqual(camera.notifications, 100)
|
||||
self.assertEqual(camera.image_link, test_value + '/image.jpg')
|
||||
self.assertEqual(camera.arm_link, test_value + '/arm')
|
||||
self.assertEqual(camera.header, {'foo': 'bar'})
|
||||
self.assertEqual(camera.motion, {'bar': 'foo'})
|
||||
self.assertEqual(camera.region_id, test_region_id)
|
||||
|
||||
self.assertEqual(camera.battery_string, "Low")
|
||||
camera.battery = 3
|
||||
self.assertEqual(camera.battery_string, "OK")
|
||||
camera.battery = -10
|
||||
self.assertEqual(camera.battery_string, "Unknown")
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_camera_values_from_setup(self, mock_get, mock_post):
|
||||
"""Tests all property values after camera setup."""
|
||||
self.blink.setup_system()
|
||||
|
||||
# Get expected test values
|
||||
test_network_id = str(mresp.NETWORKS_RESPONSE['networks'][0]['id'])
|
||||
# pylint: disable=unused-variable
|
||||
(region_id, region), = mresp.LOGIN_RESPONSE['region'].items()
|
||||
# pylint: disable=protected-access
|
||||
expected_header = self.blink._auth_header
|
||||
test_urls = blinkpy.BlinkURLHandler(region_id)
|
||||
|
||||
test_cameras = mresp.get_test_cameras(test_urls.base_url)
|
||||
test_net_id_url = test_urls.network_url + test_network_id
|
||||
for name in self.blink.cameras:
|
||||
camera = self.blink.cameras[name]
|
||||
self.assertEqual(name, camera.name)
|
||||
if name in test_cameras:
|
||||
self.assertEqual(camera.id,
|
||||
test_cameras[name]['device_id'])
|
||||
self.assertEqual(camera.armed,
|
||||
test_cameras[name]['armed'])
|
||||
self.assertEqual(camera.thumbnail,
|
||||
test_cameras[name]['thumbnail'])
|
||||
self.assertEqual(camera.temperature,
|
||||
test_cameras[name]['temperature'])
|
||||
self.assertEqual(camera.battery,
|
||||
test_cameras[name]['battery'])
|
||||
self.assertEqual(camera.notifications,
|
||||
test_cameras[name]['notifications'])
|
||||
else:
|
||||
self.fail("Camera wasn't initialized: " + name)
|
||||
|
||||
expected_arm_link = test_net_id_url + '/camera/' + camera.id + '/'
|
||||
expected_image_link = expected_arm_link + 'thumbnail'
|
||||
self.assertEqual(camera.image_link, expected_image_link)
|
||||
self.assertEqual(camera.arm_link, expected_arm_link)
|
||||
self.assertEqual(camera.header, expected_header)
|
||||
+51
-154
@@ -1,14 +1,42 @@
|
||||
"""Tests camera and system functions."""
|
||||
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import blinkpy
|
||||
|
||||
from requests import Request
|
||||
|
||||
from blinkpy import blinkpy
|
||||
from blinkpy.sync_module import BlinkSyncModule
|
||||
from blinkpy.helpers.util import create_session
|
||||
import tests.mock_responses as mresp
|
||||
|
||||
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.helpers.util.Session.send',
|
||||
side_effect=mresp.mocked_session_send)
|
||||
class TestBlinkFunctions(unittest.TestCase):
|
||||
"""Test Blink and BlinkCamera functions in blinkpy."""
|
||||
|
||||
@@ -16,160 +44,29 @@ class TestBlinkFunctions(unittest.TestCase):
|
||||
"""Set up Blink module."""
|
||||
self.blink = blinkpy.Blink(username=USERNAME,
|
||||
password=PASSWORD)
|
||||
(self.region_id, self.region), = mresp.LOGIN_RESPONSE['region'].items()
|
||||
self.test_urls = blinkpy.BlinkURLHandler(self.region_id)
|
||||
self.urls = self.test_urls
|
||||
# pylint: disable=protected-access
|
||||
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
|
||||
self.region = None
|
||||
self.region_id = None
|
||||
self.test_urls = None
|
||||
self.urls = None
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_set_motion_detect(self, mock_get, mock_post):
|
||||
"""Checks if we can set motion detection."""
|
||||
self.blink.setup_system()
|
||||
self.test_urls = blinkpy.BlinkURLHandler(self.region_id)
|
||||
test_cameras = mresp.get_test_cameras(self.test_urls.base_url)
|
||||
for camera_name in test_cameras:
|
||||
self.blink.cameras[camera_name].set_motion_detect(True)
|
||||
self.blink.refresh()
|
||||
self.assertEqual(self.blink.cameras[camera_name].armed, True)
|
||||
self.blink.cameras[camera_name].set_motion_detect(False)
|
||||
self.blink.refresh()
|
||||
self.assertEqual(self.blink.cameras[camera_name].armed, False)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_last_motion(self, mock_get, mock_post):
|
||||
"""Checks that we can get the last motion info."""
|
||||
self.test_urls = blinkpy.BlinkURLHandler(self.region_id)
|
||||
test_events = mresp.RESPONSE['event']
|
||||
test_video = dict()
|
||||
test_image = dict()
|
||||
test_time = dict()
|
||||
for event in test_events:
|
||||
if event['type'] == 'motion':
|
||||
url = self.test_urls.base_url + event['video_url']
|
||||
test_video[event['camera_name']] = url
|
||||
test_image[event['camera_name']] = url[:-3] + 'jpg'
|
||||
test_time[event['camera_name']] = event['created_at']
|
||||
|
||||
self.blink.setup_system()
|
||||
for name in self.blink.cameras:
|
||||
camera = self.blink.cameras[name]
|
||||
self.blink.last_motion()
|
||||
if name in test_video:
|
||||
self.assertEqual(camera.motion['video'], test_video[name])
|
||||
else:
|
||||
self.assertEqual(camera.motion, {})
|
||||
if name in test_image:
|
||||
self.assertEqual(camera.motion['image'], test_image[name])
|
||||
else:
|
||||
self.assertEqual(camera.motion, {})
|
||||
if name in test_video:
|
||||
self.assertEqual(camera.motion['time'], test_time[name])
|
||||
else:
|
||||
self.assertEqual(camera.motion, {})
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_take_new_picture(self, mock_get, mock_post):
|
||||
"""Checks if we can take a new picture and retrieve the thumbnail."""
|
||||
self.blink.setup_system()
|
||||
test_cameras = mresp.get_test_cameras(self.test_urls.base_url)
|
||||
test_thumbnail = self.test_urls.base_url + mresp.NEW_THUMBNAIL + '.jpg'
|
||||
# Snap picture for each camera and check new thumb
|
||||
for camera_name in test_cameras:
|
||||
camera = self.blink.cameras[camera_name]
|
||||
camera.snap_picture()
|
||||
camera.image_refresh()
|
||||
self.assertEqual(camera.thumbnail, test_thumbnail)
|
||||
|
||||
# Manually set thumbnail, and then globally refresh and check
|
||||
for camera_name in test_cameras:
|
||||
camera = self.blink.cameras[camera_name]
|
||||
camera.thumbnail = 'Testing'
|
||||
self.assertEqual(camera.thumbnail, 'Testing')
|
||||
|
||||
self.blink.refresh()
|
||||
for camera_name in test_cameras:
|
||||
camera = self.blink.cameras[camera_name]
|
||||
self.assertEqual(camera.thumbnail, test_thumbnail)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_image_with_bad_data(self, mock_get, mock_post):
|
||||
"""Checks for handling of bad keys."""
|
||||
self.blink.setup_system()
|
||||
for camera_name in self.blink.cameras:
|
||||
camera = self.blink.cameras[camera_name]
|
||||
camera.snap_picture()
|
||||
camera.urls.home_url = "use_bad_response"
|
||||
self.assertEqual(camera.image_refresh(), None)
|
||||
|
||||
def test_camera_update(self):
|
||||
"""Checks that the update function is doing the right thing."""
|
||||
self.test_urls = blinkpy.BlinkURLHandler('test')
|
||||
test_config = mresp.FIRST_CAMERA
|
||||
self.urls = self.test_urls
|
||||
test_camera = blinkpy.blinkpy.BlinkCamera(test_config, self)
|
||||
test_update = mresp.SECOND_CAMERA
|
||||
test_camera.update(test_update)
|
||||
test_image_url = self.test_urls.base_url + test_update['thumbnail']
|
||||
test_thumbnail = test_image_url + '.jpg'
|
||||
test_clip = test_image_url + '.mp4'
|
||||
self.assertEqual(test_camera.name, test_update['name'])
|
||||
self.assertEqual(test_camera.armed, test_update['armed'])
|
||||
self.assertEqual(test_camera.thumbnail, test_thumbnail)
|
||||
self.assertEqual(test_camera.clip, test_clip)
|
||||
self.assertEqual(test_camera.temperature, test_update['temp'])
|
||||
self.assertEqual(test_camera.battery, test_update['battery'])
|
||||
self.assertEqual(test_camera.notifications,
|
||||
test_update['notifications'])
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_camera_thumbs(self, mock_get, mock_post):
|
||||
"""Checks to see if we can retrieve camera thumbs."""
|
||||
self.test_urls = blinkpy.BlinkURLHandler(self.region_id)
|
||||
test_cameras = mresp.get_test_cameras(self.test_urls.base_url)
|
||||
self.blink.setup_system()
|
||||
for name in self.blink.cameras:
|
||||
thumb = self.blink.camera_thumbs[name]
|
||||
self.assertEqual(test_cameras[name]['thumbnail'], thumb)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.copyfileobj',
|
||||
side_effect=mresp.mocked_copyfileobj)
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_image_to_file(self, mock_get, mock_post, mock_copyfileobj):
|
||||
"""Checks that we can write an image to file."""
|
||||
self.blink.setup_system()
|
||||
cameras = self.blink.cameras
|
||||
filename = '/tmp/test.jpg'
|
||||
test_files = list()
|
||||
for camera_name in cameras:
|
||||
camera = cameras[camera_name]
|
||||
test_files.append(mresp.MOCK_BYTES)
|
||||
mock_fh = mock.mock_open()
|
||||
with mock.patch('builtins.open', mock_fh, create=True):
|
||||
camera.image_to_file(camera_name + filename)
|
||||
mock_fh.assert_called_once_with(camera_name + filename, 'wb')
|
||||
self.assertEqual(test_files, mresp.FAKE_FILES)
|
||||
@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'},
|
||||
'networks': {'1234': {'name': 'foobar', 'onboarded': True}}}
|
||||
]
|
||||
self.blink.get_auth_token()
|
||||
self.assertEqual(self.blink.region_id, 'piri')
|
||||
self.assertEqual(self.blink.region, 'UNKNOWN')
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._token, 'foobar123')
|
||||
|
||||
@@ -1,201 +0,0 @@
|
||||
"""
|
||||
Tests the system initialization and attributes of
|
||||
the main Blink system. Tests if we properly catch
|
||||
any communication related errors at startup.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import blinkpy
|
||||
import tests.mock_responses as mresp
|
||||
import helpers.constants as const
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
|
||||
|
||||
class TestBlinkSetup(unittest.TestCase):
|
||||
"""Test the Blink class in blinkpy."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up Blink module."""
|
||||
self.blink_no_cred = blinkpy.Blink()
|
||||
self.blink = blinkpy.Blink(username=USERNAME,
|
||||
password=PASSWORD)
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
self.blink = None
|
||||
self.blink_no_cred = None
|
||||
|
||||
def test_initialization(self):
|
||||
"""Verify we can initialize blink."""
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._username, USERNAME)
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._password, PASSWORD)
|
||||
|
||||
def test_no_credentials(self):
|
||||
"""Check that we throw an exception when no username/password."""
|
||||
with self.assertRaises(blinkpy.BlinkAuthenticationException):
|
||||
self.blink_no_cred.get_auth_token()
|
||||
with self.assertRaises(blinkpy.BlinkAuthenticationException):
|
||||
self.blink_no_cred.setup_system()
|
||||
# pylint: disable=protected-access
|
||||
self.blink_no_cred._username = USERNAME
|
||||
with self.assertRaises(blinkpy.BlinkAuthenticationException):
|
||||
self.blink_no_cred.get_auth_token()
|
||||
|
||||
def test_no_auth_header(self):
|
||||
"""Check that we throw an exception when no auth header given."""
|
||||
# pylint: disable=unused-variable
|
||||
(region_id, region), = mresp.LOGIN_RESPONSE['region'].items()
|
||||
self.blink.urls = blinkpy.BlinkURLHandler(region_id)
|
||||
with self.assertRaises(blinkpy.BlinkException):
|
||||
self.blink.get_ids()
|
||||
with self.assertRaises(blinkpy.BlinkException):
|
||||
self.blink.get_summary()
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.getpass.getpass')
|
||||
def test_manual_login(self, getpwd):
|
||||
"""Check that we can manually use the login() function."""
|
||||
getpwd.return_value = PASSWORD
|
||||
with mock.patch('builtins.input', return_value=USERNAME):
|
||||
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.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_bad_request(self, mock_get, mock_post):
|
||||
"""Check that we raise an Exception with a bad request."""
|
||||
with self.assertRaises(blinkpy.BlinkException):
|
||||
# pylint: disable=protected-access
|
||||
blinkpy.blinkpy._request(None, reqtype='bad')
|
||||
|
||||
with self.assertRaises(blinkpy.BlinkAuthenticationException):
|
||||
# pylint: disable=protected-access
|
||||
blinkpy.blinkpy._request(None, reqtype='post', is_retry=True)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_system_bad_refresh(self, mock_get, mock_post):
|
||||
"""Checks for handling of bad keys."""
|
||||
self.blink.setup_system()
|
||||
self.blink.urls.home_url = 'use_bad_response'
|
||||
self.assertEqual(self.blink.refresh(), None)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_full_setup(self, mock_get, mock_post):
|
||||
"""Check that we can set Blink up properly."""
|
||||
self.blink.setup_system()
|
||||
|
||||
# Get all test values
|
||||
authtoken = mresp.LOGIN_RESPONSE['authtoken']['authtoken']
|
||||
(region_id, region), = mresp.LOGIN_RESPONSE['region'].items()
|
||||
host = region_id + '.' + const.BLINK_URL
|
||||
network_id = mresp.NETWORKS_RESPONSE['networks'][0]['id']
|
||||
account_id = mresp.NETWORKS_RESPONSE['networks'][0]['account_id']
|
||||
test_urls = blinkpy.BlinkURLHandler(region_id)
|
||||
test_cameras = mresp.get_test_cameras(test_urls.base_url)
|
||||
test_camera_id = mresp.get_test_id_table()
|
||||
|
||||
# Check that all links have been set properly
|
||||
self.assertEqual(self.blink.region_id, region_id)
|
||||
self.assertEqual(self.blink.urls.base_url, test_urls.base_url)
|
||||
self.assertEqual(self.blink.urls.home_url, test_urls.home_url)
|
||||
self.assertEqual(self.blink.urls.event_url, test_urls.event_url)
|
||||
self.assertEqual(self.blink.urls.network_url, test_urls.network_url)
|
||||
self.assertEqual(self.blink.urls.networks_url, test_urls.networks_url)
|
||||
|
||||
# Check that all properties have been set after startup
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._token, authtoken)
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._host, host)
|
||||
self.assertEqual(self.blink.network_id, str(network_id))
|
||||
self.assertEqual(self.blink.account_id, str(account_id))
|
||||
self.assertEqual(self.blink.region, region)
|
||||
|
||||
# Verify we have initialized all the cameras
|
||||
self.assertEqual(self.blink.id_table, test_camera_id)
|
||||
for camera in test_cameras:
|
||||
self.assertTrue(camera in self.blink.cameras)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_arm_disarm_system(self, mock_get, mock_post):
|
||||
"""Check that we can arm/disarm the system"""
|
||||
self.blink.setup_system()
|
||||
self.blink.arm = False
|
||||
self.assertIs(self.blink.arm, False)
|
||||
self.blink.arm = True
|
||||
self.assertIs(self.blink.arm, True)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_check_online_status(self, mock_get, mock_post):
|
||||
"""Check that we can get our online status."""
|
||||
self.blink.setup_system()
|
||||
expected_status = const.ONLINE[mresp.RESPONSE['syncmodule']['status']]
|
||||
self.assertIs(self.blink.online, expected_status)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_setup_backup_subdomain(self, mock_get, mock_post):
|
||||
"""Check that we can use the 'rest.piri' subdomain."""
|
||||
test_urls = blinkpy.BlinkURLHandler('rest.piri')
|
||||
with mock.patch('helpers.constants.LOGIN_URL',
|
||||
return_value=const.LOGIN_URL + 'NO'):
|
||||
self.blink.setup_system()
|
||||
self.assertEqual(self.blink.region_id, 'rest.piri')
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._host, 'rest.piri.' + const.BLINK_URL)
|
||||
self.assertEqual(self.blink.urls.base_url, test_urls.base_url)
|
||||
self.assertEqual(self.blink.urls.home_url, test_urls.home_url)
|
||||
self.assertEqual(self.blink.urls.event_url, test_urls.event_url)
|
||||
self.assertEqual(self.blink.urls.network_url, test_urls.network_url)
|
||||
self.assertEqual(self.blink.urls.networks_url, test_urls.networks_url)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
@mock.patch('blinkpy.blinkpy.BlinkURLHandler',
|
||||
side_effect=mresp.MockURLHandler)
|
||||
def test_continuous_bad_auth(self, mock_url, mock_get, mock_post):
|
||||
"""Check that we don't get stuck if we can't reauthorize."""
|
||||
with self.assertRaises(blinkpy.BlinkAuthenticationException):
|
||||
self.blink.setup_system()
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.requests.post',
|
||||
side_effect=mresp.mocked_requests_post)
|
||||
@mock.patch('blinkpy.blinkpy.requests.get',
|
||||
side_effect=mresp.mocked_requests_get)
|
||||
def test_reauthorization_attempt(self, mock_get, mock_post):
|
||||
"""Check that we can reauthorize after first unsuccessful attempt."""
|
||||
self.blink.setup_system()
|
||||
# pylint: disable=protected-access
|
||||
original_header = self.blink._auth_header
|
||||
bad_header = {'Host': self.blink._host, 'TOKEN_AUTH': 'BADTOKEN'}
|
||||
self.blink._auth_header = bad_header
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._auth_header, bad_header)
|
||||
self.blink.get_summary()
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._auth_header, original_header)
|
||||
@@ -0,0 +1,127 @@
|
||||
"""
|
||||
Test full system.
|
||||
|
||||
Tests the system initialization and attributes of
|
||||
the main Blink system. Tests if we properly catch
|
||||
any communication related errors at startup.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest import mock
|
||||
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)
|
||||
import tests.mock_responses as mresp
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
|
||||
|
||||
@mock.patch('blinkpy.helpers.util.Session.send',
|
||||
side_effect=mresp.mocked_session_send)
|
||||
class TestBlinkSetup(unittest.TestCase):
|
||||
"""Test the Blink class in blinkpy."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up Blink module."""
|
||||
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."""
|
||||
self.blink = None
|
||||
self.blink_no_cred = None
|
||||
|
||||
def test_initialization(self, mock_sess):
|
||||
"""Verify we can initialize blink."""
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._username, USERNAME)
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._password, PASSWORD)
|
||||
|
||||
def test_no_credentials(self, mock_sess):
|
||||
"""Check that we throw an exception when no username/password."""
|
||||
with self.assertRaises(BlinkAuthenticationException):
|
||||
self.blink_no_cred.get_auth_token()
|
||||
# pylint: disable=protected-access
|
||||
self.blink_no_cred._username = USERNAME
|
||||
with self.assertRaises(BlinkAuthenticationException):
|
||||
self.blink_no_cred.get_auth_token()
|
||||
|
||||
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()
|
||||
self.blink.urls = BlinkURLHandler(region_id)
|
||||
with self.assertRaises(BlinkException):
|
||||
self.blink.get_ids()
|
||||
|
||||
@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):
|
||||
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)
|
||||
|
||||
def test_bad_request(self, mock_sess):
|
||||
"""Check that we raise an Exception with a bad request."""
|
||||
self.blink.session = create_session()
|
||||
with self.assertRaises(BlinkException):
|
||||
http_req(self.blink, reqtype='bad')
|
||||
|
||||
with self.assertRaises(BlinkAuthenticationException):
|
||||
http_req(self.blink, reqtype='post', is_retry=True)
|
||||
|
||||
def test_authentication(self, mock_sess):
|
||||
"""Check that we can authenticate Blink up properly."""
|
||||
authtoken = self.blink.get_auth_token()['TOKEN_AUTH']
|
||||
expected = mresp.LOGIN_RESPONSE['authtoken']['authtoken']
|
||||
self.assertEqual(authtoken, expected)
|
||||
|
||||
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'}
|
||||
# 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')
|
||||
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)
|
||||
@@ -0,0 +1,201 @@
|
||||
"""
|
||||
Test all camera attributes.
|
||||
|
||||
Tests the camera initialization and attributes of
|
||||
individual BlinkCamera instantiations once the
|
||||
Blink system is set up.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
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, MAX_CLIPS
|
||||
import tests.mock_responses as mresp
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
|
||||
CAMERA_CFG = {
|
||||
'camera': [
|
||||
{
|
||||
'battery_voltage': 90,
|
||||
'motion_alert': True,
|
||||
'wifi_strength': -30,
|
||||
'temperature': 68
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@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)
|
||||
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.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
|
||||
|
||||
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)
|
||||
|
||||
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')
|
||||
|
||||
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': '',
|
||||
}
|
||||
self.camera.sync.homescreen = {
|
||||
'devices': [
|
||||
{'foo': 'bar'},
|
||||
{'device_type': 'foobar'},
|
||||
{'device_type': 'camera',
|
||||
'name': 'new',
|
||||
'thumbnail': '/new/thumb'}
|
||||
]
|
||||
}
|
||||
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"]
|
||||
)
|
||||
|
||||
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)
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Tests camera and system functions."""
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from blinkpy import blinkpy
|
||||
from blinkpy.sync_module import BlinkSyncModule
|
||||
from blinkpy.camera import BlinkCamera
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
|
||||
|
||||
@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)
|
||||
# pylint: disable=protected-access
|
||||
self.blink._auth_header = {
|
||||
'Host': 'test.url.tld',
|
||||
'TOKEN_AUTH': 'foobar123'
|
||||
}
|
||||
self.blink.urls = blinkpy.BlinkURLHandler('test')
|
||||
self.blink.sync = BlinkSyncModule(self.blink)
|
||||
self.camera = BlinkCamera(self.blink.sync)
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
self.blink = None
|
||||
self.camera = None
|
||||
|
||||
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."""
|
||||
mock_resp.return_value = [
|
||||
{
|
||||
'camera_name': 'foobar',
|
||||
'address': '/test/clip_1900_01_01_12_00_00AM.mp4',
|
||||
'thumbnail': '/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)
|
||||
|
||||
def test_get_videos_multi_page(self, mock_resp):
|
||||
"""Test video access with multiple pages."""
|
||||
mock_resp.return_value = [
|
||||
{
|
||||
'camera_name': 'test',
|
||||
'address': '/foo/bar_1900_01_01_12_00_00AM.mp4',
|
||||
'thumbnail': '/foobar'
|
||||
}
|
||||
]
|
||||
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_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')
|
||||
@@ -1,6 +1,7 @@
|
||||
[tox]
|
||||
envlist = py34, py35, py36, lint
|
||||
envlist = build, py35, py36, py37, py38, lint
|
||||
skip_missing_interpreters = True
|
||||
skipsdist = True
|
||||
|
||||
[testenv]
|
||||
setenv =
|
||||
@@ -19,6 +20,22 @@ deps =
|
||||
basepython = python3
|
||||
ignore_errors = True
|
||||
commands =
|
||||
pylint --rcfile={toxinidir}/pylintrc blinkpy.py tests
|
||||
flake8 blinkpy.py tests
|
||||
pydocstyle blinkpy.py tests
|
||||
pylint --rcfile={toxinidir}/pylintrc blinkpy tests
|
||||
flake8 blinkpy tests
|
||||
pydocstyle blinkpy tests
|
||||
rst-lint README.rst
|
||||
rst-lint CHANGES.rst
|
||||
|
||||
[testenv:build]
|
||||
recreate = True
|
||||
skip_install = True
|
||||
whitelist_externals =
|
||||
/bin/sh
|
||||
/bin/rm
|
||||
deps =
|
||||
-r{toxinidir}/requirements_test.txt
|
||||
commands =
|
||||
/bin/rm -rf build dist
|
||||
python setup.py bdist_wheel
|
||||
/bin/sh -c "pip install --upgrade dist/*.whl"
|
||||
py.test tests
|
||||
|
||||
Reference in New Issue
Block a user