From c529f474fe1c5fe9d2fc038765f6f4f39cda5b82 Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Sun, 12 Mar 2017 19:45:33 -0400 Subject: [PATCH] Added support for rest.piri subdomain --- blinkpy.py | 21 +++++++++++++++++---- helpers/constants.py | 1 + tests/mock_responses.py | 9 ++++++++- tests/test_blink_system.py | 19 +++++++++++++++++++ 4 files changed, 45 insertions(+), 5 deletions(-) diff --git a/blinkpy.py b/blinkpy.py index 296f4d3..6391c56 100644 --- a/blinkpy.py +++ b/blinkpy.py @@ -18,15 +18,19 @@ 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 _request(url, data=None, headers=None, reqtype='get', stream=False, json_resp=True): """Wrapper function for request.""" - if reqtype is 'post': + if reqtype is 'post' and json_resp: response = requests.post(url, headers=headers, data=data).json() + elif reqtype is 'post' and not json_resp: + response = requests.post(url, headers=headers, + data=data) elif reqtype is 'get' and json_resp: response = requests.get(url, headers=headers, stream=stream).json() @@ -441,10 +445,19 @@ class Blink(object): "client_specifier": "iPhone 9.2 | 2.2 | 222" }) response = _request(LOGIN_URL, headers=headers, - data=data, reqtype='post') - self._token = response['authtoken']['authtoken'] - (self._region_id, self._region), = response['region'].items() + 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(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} diff --git a/helpers/constants.py b/helpers/constants.py index 08c7861..bf56aff 100644 --- a/helpers/constants.py +++ b/helpers/constants.py @@ -45,6 +45,7 @@ URLS ''' BLINK_URL = 'immedia-semi.com' LOGIN_URL = 'https://prod.' + BLINK_URL + '/login' +LOGIN_BACKUP_URL = 'https://rest.piri/' + BLINK_URL + '/login' BASE_URL = 'https://prod.' + BLINK_URL DEFAULT_URL = 'prod.' + BLINK_URL diff --git a/tests/mock_responses.py b/tests/mock_responses.py index c57e72e..3161b43 100644 --- a/tests/mock_responses.py +++ b/tests/mock_responses.py @@ -156,6 +156,8 @@ def mocked_requests_post(*args, **kwargs): 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' @@ -210,11 +212,16 @@ def mocked_requests_get(*args, **kwargs): # pylint: disable=unused-variable (region_id, region), = LOGIN_RESPONSE['region'].items() set_region_id = args[0].split('/')[2].split('.')[0] + 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 url ' + args[0]) + 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) else: diff --git a/tests/test_blink_system.py b/tests/test_blink_system.py index 8e9ab31..37b2be3 100644 --- a/tests/test_blink_system.py +++ b/tests/test_blink_system.py @@ -142,3 +142,22 @@ class TestBlinkSetup(unittest.TestCase): 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)