From af0cdb006a2187864704adfbe6a8506dd95ce3d6 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Sun, 12 Mar 2017 18:21:28 -0400 Subject: [PATCH] Completed test suite. Fixed some minor issues with file writing and regions --- API.md | 287 ++++++++++++++++++++++++++++++++++ CONTRIBUTING.md | 83 ++++++++++ LICENSE => LICENSE.md | 0 MANIFEST.in | 3 +- blinkpy.py | 6 +- pylintrc | 2 + tests/mock_responses.py | 110 ++++++++++++- tests/test_blink_cameras.py | 16 +- tests/test_blink_functions.py | 160 ++++++++++++++++++- tests/test_blink_hass.py | 1 - tests/test_blink_system.py | 29 +++- tox.ini | 4 +- 12 files changed, 665 insertions(+), 36 deletions(-) create mode 100644 API.md create mode 100644 CONTRIBUTING.md rename LICENSE => LICENSE.md (100%) delete mode 100644 tests/test_blink_hass.py diff --git a/API.md b/API.md new file mode 100644 index 0000000..90549bc --- /dev/null +++ b/API.md @@ -0,0 +1,287 @@ +# BlinkMonitorProtocol +Unofficial documentation for the Client API of the Blink Wire-Free HD Home Monitoring and Alert System. + +Copied from https://github.com/MattTW/BlinkMonitorProtocol + +I am not affiliated with the company in any way - this documentation is strictly **"AS-IS"**. My goal was to uncover enough to arm and disarm the system programatically so that I can issue those commands in sync with my home alarm system arm/disarm. Just some raw notes at this point but should be enough for creating programmatic APIs. Lots more to be discovered and documented - feel free to contribute! + +The Client API is a straightforward REST API using JSON and HTTPS. + +## This Document +The purpose here is to describe what is going on behind the scenes, and what commands we know are available to communicate with Blink servers. This is NOT a description of the blinkpy module, but a description of the commands blinkpy relies on to effectively communicate. + +##Login + +Client login to the Blink Servers. + +**Request:** +>curl -H "Host: prod.immedia-semi.com" -H "Content-Type: application/json" --data-binary '{ +> "password" : "*your blink password*", +> "client_specifier" : "iPhone 9.2 | 2.2 | 222", +> "email" : "*your blink login/email*" +>}' --compressed https://prod.immedia-semi.com/login + +**Response:** +>{"authtoken":{"authtoken":"*an auth token*","message":"auth"}} + +**Notes:** +The authtoken value is passed in a header in future calls. + +##Networks + +Obtain information about the Blink networks defined for the logged in user. + +**Request:** +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/networks + +**Response:** +JSON response containing information including Network ID and Account ID. + +**Notes:** +Network ID is needed to issue arm/disarm calls + + +##Sync Modules + +Obtain information about the Blink Sync Modules on the given network. + +**Request:** +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/network/*network_id_from_networks_call*/syncmodules + +**Response:** +JSON response containing information about the known state of the Sync module, most notably if it is online + +**Notes:** +Probably not strictly needed but checking result can verify that the sync module is online and will respond to requests to arm/disarm, etc. + + +##Arm + +Arm the given network (start recording/reporting motion events) + +**Request:** +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --data-binary --compressed https://prod.immedia-semi.com/network/*network_id_from_networks_call*/arm + +**Response:** +JSON response containing information about the arm command request, including the command/request ID + +**Notes:** +When this call returns, it does not mean the arm request is complete, the client must gather the request ID from the response and poll for the status of the command. + +##Disarm + +Disarm the given network (stop recording/reporting motion events) + +**Request:** +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --data-binary --compressed https://prod.immedia-semi.com/network/*network_id_from_networks_call*/disarm + +**Response:** +JSON response containing information about the disarm command request, including the command/request ID + +**Notes:** +When this call returns, it does not mean the disarm request is complete, the client must gather the request ID from the response and poll for the status of the command. + + +##Command Status + +Get status info on the given command + +**Request:** +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/network/*network_id*/command/*command_id* + +**Response:** +JSON response containing state information of the given command, most notably whether it has completed and was successful. + +**Notes:** +After an arm/disarm command, the client appears to poll this URL every second or so until the response indicates the command is complete. + +**Known Commands:** +lv_relay, arm, disarm, thumbnail, clip + +##Home Screen + +Return information displayed on the home screen of the mobile client + +**Request:** +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/homescreen + +**Response:** +JSON response containing information that the mobile client displays on the home page, including: status, armed state, links to thumbnails for each camera, etc. + +**Notes:** +Not necessary to as part of issuing arm/disarm commands, but contains good summary info. + +##Events, thumbnails & video captures + +**Request** +Get events for a given network (sync module) -- Need network ID from home + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/events/network/*network__id* + +**Response** +A json list of evets incluing URL's. Replace the "mp4" with "jpg" extension to get the thumbnail of each clip + + +**Request** +Get a video clip from the events list + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed **video url from events list.mp4** > video.mp4 + +**Response** +The mp4 video + +**Request** +Get a thumbnail from the events list + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed **video url from events list.jpg** > video_thumb.jpg + +**Response** +The jpg bytes. + +**Notes** +Note that you replace the 'mp4' with a 'jpg' to get the thumbnail + +**Request** +Captures a new thumbnail for a camera + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --data-binary --compressed https://prod.immedia-semi.com/network/*network_id*/camera/*camera_id*/thumbnail + +**Response** +Command information. + +**Request** +Captures a new video for a camera + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --data-binary --compressed https://prod.immedia-semi.com/network/*network_id*/camera/*camera_id*/clip + +**Response** +Command information. + +##Video Information + +**Request** +Get the total number of videos in the system + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/api/v2/videos/count + +**Response** +JSON response containing the total video count. + +**Request** +Gets a paginated set of video information + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/api/v2/videos/page/0 + +**Response** +JSON response containing a set of video information, including: camera name, creation time, thumbnail URI, size, length + +**Request** +Gets information for a specific video by ID + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/api/v2/video/*video_id* + +**Response** +JSON response containing video information + +**Request** +Gets a list of unwatched videos + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/api/v2/videos/unwatched + +**Response** +JSON response containing unwatched video information + +**Request** +Deletes a video + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --data-binary --compressed https://prod.immedia-semi.com/api/v2/video/*video_id*/delete + +**Response** +Unknown - not tested + +**Request** +Deletes all videos + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --data-binary --compressed https://prod.immedia-semi.com/api/v2/videos/deleteall + +**Response** +Unknown - not tested + +##Cameras + +**Request** +Gets a list of cameras + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/network/*network_id*/cameras + +**Response** +JSON response containing camera information + +**Request** +Gets information for one camera + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/network/*network_id*/camera/*camera_id* + +**Response** +JSON response containing camera information + +**Request** +Gets camera sensor information + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/network/*network_id*/camera/*camera_id*/signals + +**Response** +JSON response containing camera sensor information, such as wifi strength, temperature, and battery level + +**Request** +Enables motion detection for one camera + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: $auth_token" --data-binary --compressed https://prod.immedia-semi.com/network/*network_id*/camera/*camera_id*/enable + +**Response** +JSON response containing camera information + +**Request** +Disables motion detection for one camera + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: $auth_token" --data-binary --compressed https://prod.immedia-semi.com/network/*network_id*/camera/*camera_id*/disable + +**Response** +JSON response containing camera information + +*Note*: enabling or disabling motion detection is independent of arming or disarming the system. No motion detection or video recording will take place unless the system is armed. + + +##Miscellaneous + +**Request** +Gets information about devices that have connected to the blink service + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/account/clients + +**Response** +JSON response containing client information, including: type, name, connection time, user ID + +**Request** +Gets information about supported regions + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/regions + +**Response** +JSON response containing region information + +**Request** +Gets information about system health + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/health + +**Response** +"all ports tested are open" + +**Request** +Gets information about programs + +>curl -H "Host: prod.immedia-semi.com" -H "TOKEN_AUTH: *authtoken from login*" --compressed https://prod.immedia-semi.com/api/v1/networks/*network_id*/programs + +**Response** +Unknown. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..af0aa2e --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,83 @@ +# Contributing to blinkpy + +Everyone is welcome to contribute to blinkpy! The process to get started is described below. + +## Fork the Repository + +You can do this right in gituhb: just click the 'fork' button at the top right. + +## Setup Local Repository + +```shell +$ git clone https://github.com//blinkpy.git +$ cd blinkpy +$ git remote add upstream https://github.com/fronzbot/blinkpy.git +``` + +## Create a Local Branch + +First, you need to make sure you're on the 'dev' branch: +``git checkout dev`` +Next, you will want to create a new branch to hold your changes: +``git checkout -b `` + +## Make changes + +Now you can make changes to your code. It is worthwhile to test your code as you progress (see the **Testing** section) + +## Commit Your Changes + +To commit changes to your branch, simply add the files you want and the commit them to the branch. After that, you can push to your fork on GitHub: + +```shell +$ git add . +$ git commit -m "Put your commit text here. Please be concise, but descriptive." +$ git push origin HEAD +``` + +## Testing + +It is important to test the code to make sure your changes don't break anything major and that they pass PEP8 style conventions. +FIrst, you need to locally install ``tox`` + +```shell +$ pip3 install tox +``` + +You can then run all of the tests with the following command: + +```shell +$ tox +``` + +### Tips + +If you only want to see if you can pass the local tests, you can run ``tox -e py34``. If you just want to check for style violations, you can run ``tox -e lint``. Regardless, when you submit a pull request, your code MUST pass both the unit tests, and the linters. + +If you need to change anything in ``requirements.txt`` for any reason, you'll want to regenerate the virtual envrionments used by ``tox`` by running with the ``-r`` flag: ``tox -r`` + +Please do not locally disable any linter warnings within the ``blinkpy.py`` module itself (it's ok to do this in any of the ``test_*.py`` files) + +# Catching Up With Reality + +If your code is taking a while to develop, you may be behind the ``dev`` branch, in which case you need to catch up before creating your pull-request. To do this you can run ``git rebase`` as follows (running this on your local branch): + +```shell +$ git fetch upstream/dev +$ git rebase upstrea/dev +``` + +If rebase detects conflicts, repeat the following process until all changes have been resolved: + +1. ``git status`` shows you the filw with a conflict. You will need to edit that file and resolve the lines between ``<<<< | >>>>`. +2. Add the modified file: ``git add `` or ``git add .``. +3. Continue rebase: ``git rebase --continue``. +4. Repeat until all conflicts resolved. + +# Creating a Pull Request + +Please follow these steps to create a pull request against the ``dev`` branch: [Creating a Pull Request](https://help.github.com/articles/creating-a-pull-request/) + +# Monitor Build Status + +Once you create your PR, you can monitor the status of your build [here](https://travis-ci.org/fronzbot/blinkpy), Your code will be tested to ensure it passes and won't cause any problems after merging. \ No newline at end of file diff --git a/LICENSE b/LICENSE.md similarity index 100% rename from LICENSE rename to LICENSE.md diff --git a/MANIFEST.in b/MANIFEST.in index 831efda..3eb1ac3 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,4 +1,5 @@ include README.rst -include LICENSE +include LICENSE.md +include API.md include tests/*.py include helpers/*.py diff --git a/blinkpy.py b/blinkpy.py index 1a773b5..296f4d3 100644 --- a/blinkpy.py +++ b/blinkpy.py @@ -14,6 +14,7 @@ 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, @@ -247,11 +248,10 @@ class BlinkCamera(object): """Write image to file.""" thumb = self.image_refresh() response = _request(thumb, headers=self._header, - stream=True, json_resp=False) + reqtype='get', stream=True, json_resp=False) if response.status_code == 200: with open(path, 'wb') as imgfile: - for chunk in response: - imgfile.write(chunk) + copyfileobj(response.raw, imgfile) class Blink(object): diff --git a/pylintrc b/pylintrc index c458c3e..4c1c361 100644 --- a/pylintrc +++ b/pylintrc @@ -3,6 +3,7 @@ reports=no # Reasons disabled: # locally-disabled - it spams too much +# duplicate-code - it's annoying # 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-* @@ -10,6 +11,7 @@ reports=no disable= locally-disabled, unused-argument, + duplicate-code, too-many-arguments, too-many-branches, too-many-instance-attributes, diff --git a/tests/mock_responses.py b/tests/mock_responses.py index 5a21cf6..c57e72e 100644 --- a/tests/mock_responses.py +++ b/tests/mock_responses.py @@ -22,6 +22,8 @@ NETWORKS_RESPONSE['networks'] = [{ 'feature_plan_id': None }] +NEW_THUMBNAIL = '/NEW/THUMBNAIL/YAY' + FIRST_CAMERA = {'device_type': 'camera', 'notifications': 1, 'battery': 2, @@ -116,6 +118,16 @@ RESPONSE['network'] = {'armed': NETWORKS_RESPONSE['networks'][0]['armed'], RESPONSE['event'] = [FIRST_EVENT, SECOND_EVENT] RESPONSE['syncmodule'] = {'name': 'Vengerberg', 'status': 'online'} +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.""" @@ -131,17 +143,46 @@ def mocked_requests_post(*args, **kwargs): """Return json data from post request.""" return self.json_data - url_tail = args[0].split("/")[-1] + # 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 url_tail == 'arm' or url_tail == 'disarm': - # pylint: disable=global-variable-not-assigned - global NETWORKS_RESPONSE - # pylint: disable=global-variable-not-assigned - global RESPONSE + # 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) @@ -151,14 +192,21 @@ def mocked_requests_get(*args, **kwargs): class MockGetResponse: """Class for mock get response.""" - def __init__(self, json_data, status_code): - """Initialze mock get response.""" + 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 json data from get request.""" return self.json_data + + @property + def raw(self): + """Return raw data from get request.""" + return self.raw_data + # pylint: disable=unused-variable (region_id, region), = LOGIN_RESPONSE['region'].items() set_region_id = args[0].split('/')[2].split('.')[0] @@ -167,7 +215,53 @@ def mocked_requests_get(*args, **kwargs): return MockGetResponse(NETWORKS_RESPONSE, 200) elif set_region_id != region_id: raise ConnectionError('Received url ' + args[0]) + elif args[0] in IMAGE_TO_WRITE_URL: + return MockGetResponse({}, 200, raw_data=MOCK_BYTES) else: return MockGetResponse(RESPONSE, 200) return MockGetResponse({'message': 'ERROR', 'code': 404}, 404) + + +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 diff --git a/tests/test_blink_cameras.py b/tests/test_blink_cameras.py index 0f80693..dd5c62b 100644 --- a/tests/test_blink_cameras.py +++ b/tests/test_blink_cameras.py @@ -31,6 +31,7 @@ class TestBlinkCameraSetup(unittest.TestCase): 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] @@ -54,6 +55,7 @@ class TestBlinkCameraSetup(unittest.TestCase): 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) @mock.patch('blinkpy.blinkpy.requests.post', side_effect=mresp.mocked_requests_post) @@ -71,19 +73,7 @@ class TestBlinkCameraSetup(unittest.TestCase): expected_header = self.blink._auth_header test_urls = blinkpy.BlinkURLHandler(region_id) - test_cameras = dict() - for element in mresp.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': (test_urls.base_url + - element['thumbnail'] + '.jpg'), - 'temperature': element['temp'], - 'battery': element['battery'], - 'notifications': element['notifications'] - } + 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] diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index f98cfbd..ad3b6f9 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -1 +1,159 @@ -# Test blink camera functions +"""Tests camera and system functions.""" + +import unittest +from unittest import mock +import blinkpy +import tests.mock_responses as mresp + +USERNAME = 'foobar' +PASSWORD = 'deadbeef' + + +class TestBlinkFunctions(unittest.TestCase): + """Test Blink and BlinkCamera functions in blinkpy.""" + + def setUp(self): + """Set up Blink module.""" + self.blink = blinkpy.Blink(username=USERNAME, + password=PASSWORD) + (self.region_id, self.region), = mresp.LOGIN_RESPONSE['region'].items() + self.test_urls = blinkpy.BlinkURLHandler(self.region_id) + + def tearDown(self): + """Clean up after test.""" + self.blink = None + self.region = None + self.region_id = None + self.test_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) + + 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 + test_camera = blinkpy.blinkpy.BlinkCamera(test_config, self.test_urls) + 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) diff --git a/tests/test_blink_hass.py b/tests/test_blink_hass.py deleted file mode 100644 index 79ae84e..0000000 --- a/tests/test_blink_hass.py +++ /dev/null @@ -1 +0,0 @@ -# Test functionality used for blink component in home assistant diff --git a/tests/test_blink_system.py b/tests/test_blink_system.py index 360abd1..8e9ab31 100644 --- a/tests/test_blink_system.py +++ b/tests/test_blink_system.py @@ -41,6 +41,10 @@ class TestBlinkSetup(unittest.TestCase): 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 excpetion when no auth header given.""" @@ -49,6 +53,8 @@ class TestBlinkSetup(unittest.TestCase): 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): @@ -61,6 +67,20 @@ class TestBlinkSetup(unittest.TestCase): # 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') + @mock.patch('blinkpy.blinkpy.requests.post', side_effect=mresp.mocked_requests_post) @mock.patch('blinkpy.blinkpy.requests.get', @@ -76,13 +96,8 @@ class TestBlinkSetup(unittest.TestCase): 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 = list() - test_camera_id = dict() - for element in mresp.RESPONSE['devices']: - if ('device_type' in element and - element['device_type'] == 'camera'): - test_cameras.append(element['name']) - test_camera_id[str(element['device_id'])] = element['name'] + 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) diff --git a/tox.ini b/tox.ini index 5b91d33..9a9d595 100644 --- a/tox.ini +++ b/tox.ini @@ -7,7 +7,7 @@ setenv = LANG=en_US.UTF-8 PYTHONPATH = {toxinidir} commands = - py.test -v --timeout=30 --duration=10 --cov=blinkpy --cov-report term {posargs} + py.test -v --timeout=30 --duration=10 --cov=blinkpy --cov-report term-missing {posargs} deps = -r{toxinidir}/requirements.txt -r{toxinidir}/requirements_test.txt @@ -19,6 +19,6 @@ deps = basepython = python3 ignore_errors = True commands = - pylint --rcfile={toxinidir}/pylintrc --load-plugins=pylint.extensions.mccabe blinkpy.py tests + pylint --rcfile={toxinidir}/pylintrc blinkpy.py tests flake8 blinkpy.py tests pydocstyle blinkpy.py tests