Compare commits

..
155 Commits
Author SHA1 Message Date
Kevin FronczakandGitHub 4e6957a772 Merge pull request #100 from fronzbot/dev
Release 0.10.0
2018-10-16 17:43:21 -04:00
Kevin FronczakandGitHub e747fb1010 Merge branch 'master' into dev 2018-10-16 17:28:34 -04:00
Kevin FronczakandGitHub 9d110cdd72 Merge pull request #99 from fronzbot/0.10.0
Version bump
2018-10-16 16:53:51 -04:00
Kevin Fronczak 6fa0eacff3 Fix trailing whitespace 2018-10-16 16:26:47 -04:00
Kevin Fronczak 7ce924f52e Version bump 2018-10-16 16:24:39 -04:00
Kevin FronczakandGitHub 91b3c2a64b Merge pull request #98 from fronzbot/update-travis
Added py38 to tox and travis
2018-10-16 16:19:11 -04:00
Kevin FronczakandGitHub 97579400b1 Merge pull request #97 from fronzbot/multi-network-support
Adds multi-network handling (Thanks @dgomes and @dib0 )
2018-10-16 16:17:48 -04:00
Kevin Fronczak 63b01dde14 Removed condition on py38-dev 2018-10-16 15:43:02 -04:00
Kevin Fronczak d9b996ab02 Added py38 2018-10-16 15:41:12 -04:00
Kevin Fronczak 3af105edf5 Check for 'None' clip before saving to cache 2018-10-16 15:26:15 -04:00
Kevin Fronczak 5057592b0d Fix blink_functions test 2018-10-16 13:52:09 -04:00
Kevin Fronczak 13bb819e07 Added network checking 2018-10-16 13:28:44 -04:00
Kevin Fronczak eefa9252e5 [WIP] Adding in network checking 2018-10-16 12:27:57 -04:00
Kevin FronczakandGitHub c5b84ac357 Merge pull request #96 from fronzbot/update-changes-rst
Add changes from PR #94
2018-10-16 11:42:12 -04:00
Kevin Fronczak 787256ce66 Fixed bad linkwq 2018-10-16 11:28:18 -04:00
Kevin Fronczak 2f1877267f Add changes from PR #94 2018-10-16 11:23:31 -04:00
Kevin FronczakandGitHub 9c34fc9281 Merge pull request #94 from fronzbot/breakout-api-calls
Improve API calls
2018-10-16 11:18:52 -04:00
Kevin Fronczak 1b9bba7c3d Modified tests, added homescreen thumbnail backup Fixes #93 2018-10-15 20:52:05 -04:00
Kevin FronczakandGitHub 123867ecd8 Merge pull request #95 from fronzbot/update-changes
Updated CHANGES to reflect recent releases
2018-10-14 08:54:01 -04:00
Kevin Fronczak 26d900897b Updated CHANGES to reflect recent releases 2018-10-14 00:27:59 -04:00
Kevin Fronczak f7a5b5cc03 Tests now working
- Simplified tests a bit
2018-10-14 00:12:44 -04:00
Kevin Fronczak 7fd822caf2 Removed old api calls, using new api module
- Tests currently broken
- wifi_strength is now a number of bars rather than dBm level
- camera no longer has armed or status attributes
- added serial numbers for sync and cameras
2018-10-13 20:38:01 -04:00
Kevin Fronczak 01c6f0b12b Add all requests to single api file 2018-10-13 16:49:31 -04:00
Kevin FronczakandGitHub 62c89a925e Dev version bump 2018-09-28 15:00:51 -04:00
Kevin FronczakandGitHub 07e1759819 Merge pull request #85 from fronzbot/v0.9.0
V0.9.0
2018-09-27 21:46:04 -04:00
Kevin FronczakandGitHub 8fac6c5bc0 Merge branch 'master' into v0.9.0 2018-09-27 21:42:05 -04:00
Kevin Fronczak bd4307af58 Version bump 2018-09-27 21:33:53 -04:00
Kevin FronczakandGitHub bd6e3bb4e3 Merge pull request #84 from fronzbot/better-docstrings
Added docstrings, removed docs/_build
2018-09-27 16:17:07 -04:00
Kevin Fronczak c2630a43e1 Added docstrings, removed docs/_build 2018-09-27 16:13:19 -04:00
Kevin FronczakandGitHub 2e6fb97cc5 Merge pull request #83 from fronzbot/use-sessions
Use sessions for requests
2018-09-27 15:19:11 -04:00
Kevin Fronczak 674dae4440 Fix lint error 2018-09-27 15:15:25 -04:00
Kevin Fronczak a1082812bb Modified tests to pass with session
- Added session mock
- Removed post and get mocks
- Modified the throttling for events and summary
    - Now they're throttled, but it's overridden during a system refresh
- Store the login url in a protected variable
2018-09-27 15:10:48 -04:00
Kevin Fronczak 9d838fe4eb Added session for requests
- fixed refresh issue throttling summary and events
2018-09-27 12:35:15 -04:00
Kevin FronczakandGitHub fe0aef69e4 Merge pull request #82 from fronzbot/refactor
Refactor blinkpy code
2018-09-27 11:15:17 -04:00
Kevin Fronczak 1b1235210e Split up tests, added cache test 2018-09-27 11:07:31 -04:00
Kevin Fronczak 451d725946 Update functionality in README 2018-09-26 13:02:48 -04:00
Kevin Fronczak 9d5ae9552a Check if first init, if so force cache update 2018-09-26 12:45:29 -04:00
Kevin Fronczak c3c558e934 Fix broken refresh throttle 2018-09-26 12:24:45 -04:00
Kevin Fronczak a570413e66 Fixed broken tests...probably should re-write them 2018-09-26 12:11:41 -04:00
Kevin Fronczak 3e23374b32 Modified requests so cameras only talk to sync 2018-09-26 11:09:51 -04:00
Kevin Fronczak 1c6a5eb6c6 Fixed tests. Need to improve mocking in functions test. 2018-09-25 16:02:59 -04:00
Kevin Fronczak 40e9e4bd6c Fixed camera tests 2018-09-25 14:31:23 -04:00
Kevin Fronczak e1ff851068 First cut of refactoring 2018-09-25 14:16:14 -04:00
Kevin FronczakandGitHub 8a4e507f22 Merge pull request #81 from fronzbot/add-refresh-rate
Add refresh rate and image caching
2018-09-25 12:23:41 -04:00
Kevin Fronczak cd7b5bfabc Fixing travis errors with new python versions 2018-09-25 12:15:19 -04:00
Kevin Fronczak 346882f1ff Remove Python 3.4 from travis, add py38-dev 2018-09-25 12:07:40 -04:00
Kevin Fronczak 5bc2466da2 Add py37 to tox.ini 2018-09-25 12:04:22 -04:00
Kevin Fronczak 672f59ab4e Added image and video caching 2018-09-25 12:03:32 -04:00
Kevin Fronczak ae26a9e371 Added refresh rate throttling 2018-09-25 11:07:40 -04:00
Kevin FronczakandGitHub 37fc047174 Merge pull request #80 from fronzbot/v0.8.3
V0.8.3
2018-09-24 22:31:12 -04:00
Kevin FronczakandGitHub 6343965bbc Merge branch 'master' into v0.8.3 2018-09-24 22:30:56 -04:00
Kevin Fronczak 411da60bb2 version_bump 2018-09-24 22:30:17 -04:00
Kevin Fronczak 72c36d8381 revert tests 2018-09-24 22:27:42 -04:00
Kevin Fronczak b0974c1ab2 reverting to 0.8.1 style thumbnail retreival 2018-09-24 22:25:43 -04:00
Kevin Fronczak a76eac085a Fix thumbnail since we aren't using homescreen anymore 2018-09-24 22:12:07 -04:00
Kevin FronczakandGitHub 5319714a37 Merge pull request #79 from fronzbot/v0.8.2
V0.8.2
2018-09-24 21:36:44 -04:00
Kevin FronczakandGitHub 5d9105655f Merge branch 'master' into v0.8.2 2018-09-24 21:32:02 -04:00
Kevin Fronczak b98c03b96b version bump 2018-09-24 21:30:23 -04:00
Kevin FronczakandGitHub ec703ae4b0 Merge pull request #78 from fronzbot/sort-motion-clips
Changed motion check to look at sorted date list
2018-09-24 21:29:01 -04:00
Kevin Fronczak 9ce09b4761 Fix thumbnails grabbing wrong date 2018-09-24 21:24:21 -04:00
Kevin Fronczak 6d002fc8b0 record_dates needs to default to dict, not list 2018-09-24 20:59:16 -04:00
Kevin Fronczak 7228340677 Changed motion check to look at sorted date list
- Previous method using clips resulted in an unsorted array, so the
newest clip wasn't always added during check
2018-09-24 20:26:36 -04:00
Kevin FronczakandGitHub 9335a37feb Merge pull request #77 from fronzbot/v0.8.1
v0.8.1
2018-09-24 13:35:48 -04:00
Kevin Fronczak 371348533f Fix trailing whitespace 2018-09-24 13:30:56 -04:00
Kevin Fronczak 26f6defe60 Version bump 2018-09-24 13:27:10 -04:00
Kevin FronczakandGitHub 52bc4af1f0 Merge pull request #76 from fronzbot/better-motion
Fix motion detection
2018-09-24 13:25:14 -04:00
Kevin Fronczak 8c94305804 Fix motion detection
- Use an array of recent video clips to determine if motion has been
detected.
- Reset the value every system refresh
2018-09-24 13:17:15 -04:00
Kevin FronczakandGitHub 0eb57698bd Merge pull request #72 from fronzbot/lint-upgrade
Update linter versions
2018-08-09 09:43:51 -04:00
Kevin FronczakandGitHub 7b0db80edd Disable no-else-return 2018-08-09 09:37:57 -04:00
Kevin FronczakandGitHub 0e63e97a57 Remove whitespace 2018-08-09 09:27:36 -04:00
Kevin FronczakandGitHub e4a21964d2 Fix useless-import-alias 2018-08-09 09:26:22 -04:00
Kevin FronczakandGitHub 39ad37f23b Fix pylint errors
- change if comparison to fix (consider-using-in)
- Disabled no else-if-return check
2018-08-09 09:25:42 -04:00
Kevin FronczakandGitHub 34bc61e1c3 Fix pylint warnings
- Remove object from class declarations
- Remove useless returns from functions
2018-08-09 09:21:49 -04:00
Kevin FronczakandGitHub e6a7c367b3 Merge branch 'dev' into lint-upgrade 2018-08-09 07:35:26 -04:00
Kevin FronczakandGitHub 66afd33b79 Merge pull request #71 from fronzbot/upgrade-pytest
Pytest 3.7.1
2018-08-09 07:28:51 -04:00
Kevin FronczakandGitHub cf617d720e Update linter versions 2018-08-09 07:27:12 -04:00
Kevin FronczakandGitHub 4f2eca08e0 Update requirements_test.txt 2018-08-09 07:21:05 -04:00
Kevin FronczakandGitHub 2972780c18 Merge pull request #69 from fronzbot/dev-version-bump
Dev version bump
2018-05-21 11:47:52 -04:00
Kevin Fronczak 9c3f3f2f98 Version bump 2018-05-21 11:44:05 -04:00
Kevin FronczakandGitHub 0d83a7fcb1 Merge pull request #68 from fronzbot/v0.8.0
V0.8.0
2018-05-21 11:40:14 -04:00
Kevin FronczakandGitHub 80f4ca7c4f Merge branch 'master' into v0.8.0 2018-05-21 11:36:34 -04:00
Kevin Fronczak 96e7fce304 Version bump 2018-05-21 11:34:09 -04:00
Kevin FronczakandGitHub 13a358c07c Merge pull request #67 from fronzbot/celcius-support
Add celcius property for cameras
2018-05-21 11:26:43 -04:00
Kevin Fronczak 389745533c Add celcius property for cameras 2018-05-21 11:12:32 -04:00
Kevin FronczakandGitHub 65dd547a32 Merge pull request #65 from fronzbot/battery-percent
Added new request url for camera config
2018-05-21 11:03:25 -04:00
Kevin Fronczak 91a877da79 Fix tests, updated changes in changelog 2018-05-21 10:54:32 -04:00
Kevin Fronczak d357215e0e Added new request url for camera config
- Adds support for battery voltage level
- Added new camera properties (wifi_strength, motion_detected)
- Store full camera config from request into dict
2018-05-21 00:16:24 -04:00
Kevin FronczakandGitHub ff47dc4731 Merge pull request #63 from fronzbot/release-0.7.1
v0.7.1
2018-05-09 11:15:56 -04:00
Kevin Fronczak fedea74256 Version bump 2018-05-09 11:08:35 -04:00
Kevin FronczakandGitHub 66a6701a66 Merge pull request #62 from fronzbot/gitlab
Add gitlab support, py3.7 checks.  Also removes py34 tests
2018-05-09 10:46:13 -04:00
Kevin Fronczak 5bdfd6da52 Add gitlab support, py3.7 checks. Also removes py34 tests 2018-05-09 10:29:33 -04:00
Kevin FronczakandGitHub a8f90400b2 Merge pull request #61 from fronzbot/pip-fix
Fixed import error for pip version 10
2018-05-09 10:21:43 -04:00
Kevin Fronczak 3348386a4c Style fixes 2018-05-09 10:05:03 -04:00
Kevin Fronczak 1b53c4a0dd Hound should ignore setup.py 2018-05-09 10:02:58 -04:00
Kevin Fronczak 24158638ef Fix hound 2018-05-09 10:00:13 -04:00
Kevin Fronczak 901cbfc9d1 Fixed import error for pip version 10 2018-05-09 09:49:41 -04:00
Kevin FronczakandGitHub 9f85068a0d Merge pull request #59 from fronzbot/dev-version-bump
dev version bump
2018-02-08 17:02:38 -05:00
Kevin FronczakandGitHub 90b83544f4 Update constants.py 2018-02-08 16:53:03 -05:00
Kevin FronczakandGitHub 1d8e8e53d9 Update CHANGES.rst 2018-02-08 16:52:17 -05:00
Kevin FronczakandGitHub 46a0a433ec Merge pull request #58 from fronzbot/master
Rebase with master
2018-02-08 16:51:05 -05:00
Kevin FronczakandGitHub d38d30471a Update CHANGES.rst 2018-02-08 16:46:21 -05:00
Kevin FronczakandGitHub 5aad8ea641 Merge pull request #57 from fronzbot/dev
v0.7.0
2018-02-08 16:44:37 -05:00
Kevin FronczakandGitHub 74f4bd7763 Version bump 2018-02-08 16:35:37 -05:00
Kevin FronczakandGitHub 51dd69a51f Merge pull request #56 from fronzbot/camera-attributes
Camera attributes
2018-02-08 16:22:14 -05:00
Kevin FronczakandGitHub 950e7d4971 Merge pull request #55 from fronzbot/auth-at-login
Added authentication to login function and deprecated setup_system
2018-02-08 16:18:19 -05:00
Kevin Fronczak 623b694c78 Added attributes property to camera object 2018-02-08 16:16:23 -05:00
Kevin Fronczak 7162ea24cc Added authentication to login function and deprecated setup_system 2018-02-08 15:31:26 -05:00
Kevin FronczakandGitHub 968b77b8bd Merge pull request #53 from fronzbot/logging-support
Added logging
2018-02-08 14:59:46 -05:00
Kevin Fronczak 6e00b126d1 Added logging 2018-02-08 14:49:10 -05:00
Kevin FronczakandGitHub 003ceb0944 Merge pull request #50 from b10m/active_vs_armed
Handle the armed status better
2018-02-03 09:53:43 -05:00
Kevin Fronczak ce6dec50cd Fixed requirement issue causing tox failure
- Updated camera test to check for new status change
- Added updated to CHANGES.rst
2018-02-03 09:50:08 -05:00
B10m 87643ddcab Reflect changes in tests 2018-02-03 01:11:27 +01:00
B10m 017d8aa6c0 Handle the armed status better
Blink allows users to arm/disarm individual cameras. This is pretty useful for
inside and outside cameras, for instance.

The 'armed' information in the JSON however, merely indicates whether the
*system* is armed, not whether the camera itself is armed. This value, which is
the same as blink.arm, does not indicate whether the camera has motion
detection enabled.

In the JSON, the true status information lies in the 'active' key. The value
can be 'armed', 'disarmed' or 'disabled'. This commit reflects that value
in the armed property.
2018-02-03 00:32:20 +01:00
Kevin FronczakandGitHub 87b83465b3 Merge pull request #48 from fronzbot/update-readme
Add doc badge
2018-01-20 23:26:57 -05:00
Kevin Fronczak a1900f0130 Add doc badge 2018-01-20 23:26:03 -05:00
Kevin FronczakandGitHub cf9ecff529 Merge pull request #47 from fronzbot/docs
Updated theme
2018-01-20 23:18:14 -05:00
Kevin Fronczak db3dee81e9 Updated theme 2018-01-20 23:17:06 -05:00
Kevin FronczakandGitHub 798032be08 Merge pull request #46 from fronzbot/docs
Add documentation
2018-01-20 23:01:44 -05:00
Kevin Fronczak d6834dca06 Ignore docs for Hound 2018-01-20 22:57:38 -05:00
Kevin Fronczak 933cd7229a Add documentation 2018-01-20 22:53:02 -05:00
Kevin FronczakandGitHub 8e51a86971 Merge pull request #45 from fronzbot/video-clip-support
Added ability to download recent video clip
2018-01-20 20:22:36 -05:00
Kevin Fronczak 8ab44041eb Added ability to download recent video clip 2018-01-20 20:13:33 -05:00
Kevin Fronczak 79191d757a Changed coveralls badge to dev branch 2018-01-20 20:03:28 -05:00
Kevin FronczakandGitHub 7d01f54d73 Merge pull request #44 from fronzbot/update-api
Update Blink API Calls
2018-01-20 20:00:55 -05:00
Kevin Fronczak 63c0e11cc9 Fix linting error 2018-01-20 19:57:06 -05:00
Kevin Fronczak 7347d03f93 Added case-sensitive test, logged all changes 2018-01-20 19:53:27 -05:00
Kevin Fronczak fffb10d7e6 Refactored unit tests for special functions 2018-01-20 19:41:15 -05:00
Kevin Fronczak 91b3de1444 Refactored requests, expanded video retreival
- Removed blinkpy.last_motion()
- blinkpy.get_summary() changed to blinkpy.summary property
2018-01-20 16:53:38 -05:00
Kevin Fronczak 8340f2f78a WIP Simplified unit tests 2018-01-20 15:38:37 -05:00
Kevin Fronczak e65840d269 Fix lint 2018-01-20 11:43:46 -05:00
Kevin Fronczak c245bc24a6 Added video retrieval 2018-01-20 11:35:01 -05:00
Kevin Fronczak 71ebc3dc8a Updated links to retrieve images 2018-01-20 10:42:14 -05:00
Kevin FronczakandGitHub 7ade321dd3 Merge pull request #43 from fronzbot/update-travis
Update travis to use 3.6 for linting and build
2018-01-20 09:19:45 -05:00
Kevin Fronczak 19c0114eea Update travis to use 3.6 for linting and build 2018-01-20 09:15:09 -05:00
Kevin Fronczak 968c0cf8fc Fixed typos 2018-01-19 14:52:40 -05:00
Kevin Fronczak f8d4874f2f Updated README badges to dev branch, improved setup script 2018-01-19 14:47:12 -05:00
Kevin Fronczak 261d6293dd Fix lint, upgrade pylint to 1.8.1 2018-01-19 14:38:16 -05:00
Kevin Fronczak 988265fd9f Removed API warning 2018-01-19 11:45:08 -05:00
Kevin Fronczak ba03894e94 Refactored code in prep for API overhaul 2018-01-19 11:41:26 -05:00
Kevin FronczakandGitHub 556c28d4b9 Update README.rst 2017-12-29 20:09:34 -05:00
Kevin FronczakandGitHub 9d2a8c6b69 Update README.rst 2017-07-05 13:33:05 -04:00
Kevin FronczakandGitHub 89b517f47e Merge pull request #39 from fronzbot/pylint-upgrade
Upgrade to pylint==1.7.2
2017-07-05 13:31:45 -04:00
Kevin Fronczak 150016f543 Fixed underline too short 2017-07-05 13:24:22 -04:00
Kevin Fronczak 5df599be15 Upgrade to pylint==1.7.2
Also added PyPi badge to readme
2017-07-05 13:16:12 -04:00
Kevin FronczakandGitHub 0fd4b4b274 Merge pull request #37 from fronzbot/case_insensitive_cameras
Added case handling on camera dict
2017-06-18 18:47:41 -04:00
Kevin FronczakandGitHub dd744053f4 Fixed issue link 2017-06-18 18:41:21 -04:00
Kevin Fronczak be68c0d714 Added case handling on camera dict 2017-06-18 18:37:19 -04:00
Kevin FronczakandGitHub 2a4b0d3f4b Merge pull request #33 from fronzbot/fronzbot-patch-1
Update README.rst
2017-05-15 18:24:17 -04:00
Kevin FronczakandGitHub 46dd002e5d Update README.rst 2017-05-15 18:18:59 -04:00
Kevin FronczakandGitHub 04bc0ace3f Update README.rst
Fixed incorrect example code
2017-05-15 18:17:46 -04:00
Kevin FronczakandGitHub fa6cc49343 Merge pull request #32 from fronzbot/rst-linter
Added rst checking
2017-05-13 20:26:12 -04:00
Kevin Fronczak 231d191556 Added rst checking
Added reStructuredText checking so that the README is actually readable when uploaded to PyPi
2017-05-13 20:14:32 -04:00
Kevin FronczakandGitHub b6c8141042 Merge pull request #31 from fronzbot/upgrade-tests
Upgrade tests
2017-05-13 18:34:10 -04:00
Kevin Fronczak ae209a37f7 Pass lint version bumps 2017-05-13 18:27:01 -04:00
Kevin Fronczak 0d07ae1b4c Upgrade tests
-  pylint to 1.7.1
-  pydocstyle to 2.0.0
-  Added bdist_wheel test to ensure packaging working properly
2017-05-13 12:44:36 -04:00
38 changed files with 1910 additions and 1259 deletions
+2
View File
@@ -1,3 +1,4 @@
.pytest_cache/*
.cache/*
.tox/*
__pycache__/*
@@ -8,3 +9,4 @@ htmlcov/*
dist/*
.sh
build/*
docs/_build
+25
View File
@@ -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
+5
View File
@@ -1,2 +1,7 @@
python:
enabled: true
ignore_file: setup.py
ignored_directories:
- docs
fail_on_violations: true
+11 -6
View File
@@ -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
View File
@@ -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
View File
@@ -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
-5
View File
@@ -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
View File
@@ -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'])
+1
View File
@@ -0,0 +1 @@
"""Init file for blinkpy."""
+159
View File
@@ -0,0 +1,159 @@
"""Implements known blink API calls."""
import logging
from json import dumps
import blinkpy.helpers.errors as ERROR
from blinkpy.helpers.util import http_req, BlinkException
from blinkpy.helpers.constants import DEFAULT_URL
_LOGGER = logging.getLogger(__name__)
def request_login(blink, url, username, password):
"""Login request."""
headers = {
'Host': DEFAULT_URL,
'Content-Type': 'application/json'
}
data = dumps({
'email': username,
'password': password,
'client_specifier': 'iPhone 9.2 | 2.2 | 222'
})
return http_req(blink, url=url, headers=headers, data=data,
json_resp=False, reqtype='post')
def request_networks(blink):
"""Request network information."""
url = "{}/networks".format(blink.urls.base_url)
return http_get(blink, url)
def request_syncmodule(blink, network):
"""Request sync module info."""
url = "{}/network/{}/syncmodules".format(blink.urls.base_url, network)
return http_get(blink, url)
def request_system_arm(blink, network):
"""Arm system."""
url = "{}/network/{}/arm".format(blink.urls.base_url, network)
return http_post(blink, url)
def request_system_disarm(blink, network):
"""Disarm system."""
url = "{}/network/{}/disarm".format(blink.urls.base_url, network)
return http_post(blink, url)
def request_command_status(blink, network, command_id):
"""Request command status."""
url = "{}/network/{}/command_id/{}".format(blink.urls.base_url,
network,
command_id)
return http_get(blink, url)
def request_homescreen(blink):
"""Request homescreen info."""
url = "{}/homescreen".format(blink.urls.base_url)
return http_get(blink, url)
def request_sync_events(blink, network):
"""Request events from sync module."""
url = "{}/events/network/{}".format(blink.urls.base_url, network)
return http_get(blink, url)
def request_new_image(blink, network, camera_id):
"""Request to capture new thumbnail for camera."""
url = "{}/network/{}/camera/{}/thumbnail".format(blink.urls.base_url,
network,
camera_id)
return http_post(blink, url)
def request_new_video(blink, network, camera_id):
"""Request to capture new video clip."""
url = "{}/network/{}/camera/{}/clip".format(blink.urls.base_url,
network,
camera_id)
return http_post(blink, url)
def request_video_count(blink, headers):
"""Request total video count."""
url = "{}/api/v2/videos/count".format(blink.urls.base_url)
return http_get(blink, url)
def request_videos(blink, page=0):
"""Perform a request for videos."""
url = "{}/api/v2/videos/page/{}".format(blink.urls.base_url, page)
return http_get(blink, url)
def request_cameras(blink, network):
"""Request all camera information."""
url = "{}/network/{}/cameras".format(blink.urls.base_url, network)
return http_get(blink, url)
def request_camera_info(blink, network, camera_id):
"""Request camera info for one camera."""
url = "{}/network/{}/camera/{}".format(blink.urls.base_url,
network,
camera_id)
return http_get(blink, url)
def request_camera_sensors(blink, network, camera_id):
"""Request camera sensor info for one camera."""
url = "{}/network/{}/camera/{}/signals".format(blink.urls.base_url,
network,
camera_id)
return http_get(blink, url)
def request_motion_detection_enable(blink, network, camera_id):
"""Enable motion detection for a camera."""
url = "{}/network/{}/camera/{}/enable".format(blink.urls.base_url,
network,
camera_id)
return http_post(blink, url)
def request_motion_detection_disable(blink, network, camera_id):
"""Disable motion detection for a camera."""
url = "{}/network/{}/camera/{}/disable".format(blink.urls.base_url,
network,
camera_id)
return http_post(blink, url)
def http_get(blink, url, stream=False, json=True):
"""
Perform an http get request.
:param url: URL to perform get request.
:param stream: Stream response? True/FALSE
:param json: Return json response? TRUE/False
"""
if blink.auth_header is None:
raise BlinkException(ERROR.AUTH_TOKEN)
return http_req(blink, url=url, headers=blink.auth_header,
reqtype='get', stream=stream, json_resp=json)
def http_post(blink, url):
"""
Perform an http post request.
:param url: URL to perfom post request.
"""
if blink.auth_header is None:
raise BlinkException(ERROR.AUTH_TOKEN)
return http_req(blink, url=url, headers=blink.auth_header, reqtype='post')
+171
View File
@@ -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
+205
View File
@@ -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
+1
View File
@@ -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'
]
+13
View File
@@ -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)")
+99
View File
@@ -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)
+172
View File
@@ -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
+20
View File
@@ -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)
+1
View File
@@ -0,0 +1 @@
../README.rst
+30
View File
@@ -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
View File
@@ -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'),
]
+22
View File
@@ -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`
+36
View File
@@ -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
-3
View File
@@ -1,3 +0,0 @@
"""Init file for blinkpy helper functions."""
from helpers import constants
from helpers import errors
-6
View File
@@ -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)")
+3 -1
View File
@@ -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
View File
@@ -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
View File
@@ -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
+33 -9
View File
@@ -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
View File
@@ -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
+32
View File
@@ -0,0 +1,32 @@
"""Test various api functions."""
import unittest
from blinkpy import api
from blinkpy.blinkpy import Blink
from blinkpy.helpers.util import create_session
class TestBlinkAPI(unittest.TestCase):
"""Test the blink api module."""
def setUp(self):
"""Initialize the blink module."""
self.blink = Blink()
self.blink.session = create_session()
# pylint: disable=protected-access
self.blink._auth_header = {}
def tearDown(self):
"""Tear down blink module."""
self.blink = None
def test_http_req_connect_error(self):
"""Test http_get error condition."""
expected = ("ERROR:blinkpy.helpers.util:"
"Cannot connect to server. Possible outage.")
with self.assertLogs() as getlog:
api.http_get(self.blink, 'http://notreal.fake')
with self.assertLogs() as postlog:
api.http_post(self.blink, 'http://notreal.fake')
self.assertEqual(getlog.output, [expected])
self.assertEqual(postlog.output, [expected])
-107
View File
@@ -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
View File
@@ -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')
-201
View File
@@ -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)
+127
View File
@@ -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)
+201
View File
@@ -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)
+100
View File
@@ -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')
+21 -4
View File
@@ -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