From f80a93f8e1b25e16b325e5e3920aa241a26af47a Mon Sep 17 00:00:00 2001 From: Kevin Fronczak Date: Tue, 5 May 2020 12:46:48 -0400 Subject: [PATCH] Refactor login logic in preperation for 2FA --- .gitignore | 2 +- blinkpy/api.py | 21 ++++++- blinkpy/blinkpy.py | 107 ++++++---------------------------- blinkpy/helpers/constants.py | 10 ++-- blinkpy/helpers/util.py | 7 +++ blinkpy/login_handler.py | 106 +++++++++++++++++++++++++++++++++ tests/mock_responses.py | 4 +- tests/test_api.py | 46 --------------- tests/test_blink_functions.py | 16 +++-- tests/test_blinkpy.py | 99 ++----------------------------- tests/test_login_handler.py | 70 ++++++++++++++++++++++ 11 files changed, 243 insertions(+), 245 deletions(-) create mode 100644 blinkpy/login_handler.py delete mode 100644 tests/test_api.py create mode 100644 tests/test_login_handler.py diff --git a/.gitignore b/.gitignore index 54a40bf..b35c2f0 100644 --- a/.gitignore +++ b/.gitignore @@ -3,7 +3,7 @@ .tox/* __pycache__/* htmlcov/* -.coverage +.coverage.* coverage.xml *.pyc *.egg*/* diff --git a/blinkpy/api.py b/blinkpy/api.py index 7b1412f..2e7b7e7 100644 --- a/blinkpy/api.py +++ b/blinkpy/api.py @@ -11,7 +11,9 @@ _LOGGER = logging.getLogger(__name__) MIN_THROTTLE_TIME = 2 -def request_login(blink, url, username, password, is_retry=False): +def request_login( + blink, url, username, password, notification_key, uid, is_retry=False +): """ Login request. @@ -26,7 +28,14 @@ def request_login(blink, url, username, password, is_retry=False): { "email": username, "password": password, - "client_specifier": "iPhone 9.2 | 2.2 | 222", + "notification_key": notification_key, + "unique_id": uid, + "app_version": "6.0.7 (520300) #afb0be72a", + "client_name": "Computer", + "client_type": "android", + "device_identifier": "Blinkpy", + "os_version": "5.1.1", + "reauth": "true", } ) return http_req( @@ -40,6 +49,14 @@ def request_login(blink, url, username, password, is_retry=False): ) +def request_verify(blink, verify_key): + """Send verification key to blink servers.""" + url = "api/v4/account/{}/client/{}/pin/verify".format( + blink.account_id, blink.client_id + ) + return http_post(blink, url) + + def request_networks(blink): """Request all networks information.""" url = "{}/networks".format(blink.urls.base_url) diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index 347abc1..a9eca1d 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -15,9 +15,7 @@ blinkpy is in no way affiliated with Blink, nor Immedia Inc. import os.path import time -import getpass import logging -import json from shutil import copyfileobj from requests.structures import CaseInsensitiveDict @@ -26,25 +24,22 @@ from slugify import slugify from blinkpy import api from blinkpy.sync_module import BlinkSyncModule -from blinkpy.helpers import errors as ERROR from blinkpy.helpers.util import ( create_session, merge_dicts, get_time, BlinkURLHandler, - BlinkAuthenticationException, Throttle, ) from blinkpy.helpers.constants import ( BLINK_URL, - LOGIN_URL, - OLD_LOGIN_URL, - LOGIN_BACKUP_URL, DEFAULT_MOTION_INTERVAL, DEFAULT_REFRESH, MIN_THROTTLE_TIME, + LOGIN_URLS, ) from blinkpy.helpers.constants import __version__ +from blinkpy.login_handler import LoginHandler _LOGGER = logging.getLogger(__name__) @@ -81,13 +76,14 @@ class Blink: endpoints (only use if you are having api issues). """ - self._username = username - self._password = password - self._cred_file = cred_file + self.login_handler = LoginHandler( + username=username, password=password, cred_file=cred_file + ) self._token = None self._auth_header = None self._host = None self.account_id = None + self.client_id = None self.network_ids = [] self.urls = None self.sync = CaseInsensitiveDict({}) @@ -99,11 +95,12 @@ class Blink: self.networks = [] self.cameras = CaseInsensitiveDict({}) self.video_list = CaseInsensitiveDict({}) - self._login_url = LOGIN_URL + self.login_url = LOGIN_URLS[0] self.login_urls = [] self.motion_interval = motion_interval self.version = __version__ self.legacy = legacy_subdomain + self.available = False @property def auth_header(self): @@ -117,12 +114,7 @@ class Blink: 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: - if not self.login(): - return - elif not self.get_auth_token(): - return - + self.get_auth_token() camera_list = self.get_cameras() networks = self.get_ids() for network_name, network_id in networks.items(): @@ -137,88 +129,33 @@ class Blink: self.cameras = self.merge_cameras() def login(self): - """Prompt user for username and password.""" - if self._cred_file is not None and os.path.isfile(self._cred_file): - try: - with open(self._cred_file, "r") as json_file: - creds = json.load(json_file) - self._username = creds["username"] - self._password = creds["password"] - except ValueError: - _LOGGER.error( - "Improperly formated json file %s.", self._cred_file, exc_info=True - ) - return False - except KeyError: - _LOGGER.error("JSON file information incomplete %s.", exc_info=True) - return False - else: - self._username = input("Username:") - self._password = getpass.getpass("Password:") - - if self.get_auth_token(): - _LOGGER.debug("Login successful!") - return True - _LOGGER.warning("Unable to login with %s.", self._username) - return False + """Login method. DEPRECATED.""" + _LOGGER.warning( + "Method is deprecated and will be removed in a future version. Please use the LoginHandler.login() method instead." + ) + return self.login_handler.login(self) def get_auth_token(self, is_retry=False): """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) - - self.login_urls = [LOGIN_URL, OLD_LOGIN_URL, LOGIN_BACKUP_URL] - - response = self.login_request(is_retry=is_retry) + response = self.login_handler.login(self) if not response: return False + self.login_url = self.login_handler.login_url + ((self.region_id, self.region),) = response["region"].items() self._host = "{}.{}".format(self.region_id, BLINK_URL) self._token = response["authtoken"]["authtoken"] self.networks = response["networks"] - + self.client_id = response["client"]["id"] + self.account_id = response["account"]["id"] self._auth_header = {"Host": self._host, "TOKEN_AUTH": self._token} self.urls = BlinkURLHandler(self.region_id, legacy=self.legacy) return self._auth_header - def login_request(self, is_retry=False): - """Make a login request.""" - try: - login_url = self.login_urls.pop(0) - except IndexError: - _LOGGER.error("Could not login to blink servers.") - return False - - _LOGGER.info("Attempting login with %s", login_url) - - response = api.request_login( - self, login_url, self._username, self._password, is_retry=is_retry - ) - try: - if response.status_code != 200: - response = self.login_request(is_retry=True) - response = response.json() - ((self.region_id, self.region),) = response["region"].items() - - except AttributeError: - _LOGGER.error("Login API endpoint failed with response %s", response) - return False - - except KeyError: - _LOGGER.warning("Could not extract region info.") - self.region_id = "piri" - self.region = "UNKNOWN" - - self._login_url = login_url - return response - def get_ids(self): """Set the network ID and Account ID.""" - response = api.request_networks(self) all_networks = [] network_dict = {} for network, status in self.networks.items(): @@ -226,12 +163,6 @@ class Blink: all_networks.append("{}".format(network)) network_dict[status["name"]] = network - # For the first onboarded network we find, grab the account id - for resp in response["networks"]: - if str(resp["id"]) in all_networks: - self.account_id = resp["account_id"] - break - self.network_ids = all_networks return network_dict diff --git a/blinkpy/helpers/constants.py b/blinkpy/helpers/constants.py index 6cdbaba..6099f02 100644 --- a/blinkpy/helpers/constants.py +++ b/blinkpy/helpers/constants.py @@ -48,9 +48,10 @@ URLS BLINK_URL = "immedia-semi.com" DEFAULT_URL = "{}.{}".format("rest-prod", BLINK_URL) BASE_URL = "https://{}".format(DEFAULT_URL) -LOGIN_URL = "{}/api/v2/login".format(BASE_URL) -OLD_LOGIN_URL = "{}/login".format(BASE_URL) -LOGIN_BACKUP_URL = "https://{}.{}/login".format("rest-piri", BLINK_URL) +LOGIN_URLS = [ + "{}/api/v4/login".format(BASE_URL), + "{}/api/v2/login".format(BASE_URL), +] """ Dictionaries @@ -61,7 +62,8 @@ ONLINE = {"online": True, "offline": False} OTHER """ TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S%z" - DEFAULT_MOTION_INTERVAL = 1 DEFAULT_REFRESH = 30 MIN_THROTTLE_TIME = 2 +SIZE_NOTIFICATION_KEY = 152 +SIZE_UID = 16 diff --git a/blinkpy/helpers/util.py b/blinkpy/helpers/util.py index 3d7637d..9079e3d 100644 --- a/blinkpy/helpers/util.py +++ b/blinkpy/helpers/util.py @@ -2,6 +2,7 @@ import logging import time +import secrets from calendar import timegm from functools import partial, wraps from requests import Request, Session, exceptions @@ -13,6 +14,12 @@ import blinkpy.helpers.errors as ERROR _LOGGER = logging.getLogger(__name__) +def gen_uid(size): + """Create a random sring.""" + full_token = secrets.token_hex(size) + return full_token[0:size] + + def time_to_seconds(timestamp): """Convert TIMESTAMP_FORMAT time to seconds.""" try: diff --git a/blinkpy/login_handler.py b/blinkpy/login_handler.py new file mode 100644 index 0000000..b2e22a0 --- /dev/null +++ b/blinkpy/login_handler.py @@ -0,0 +1,106 @@ +"""Login handler for blink.""" +import json +import logging +from os.path import isfile +from getpass import getpass +from blinkpy import api +from blinkpy.helpers import util +from blinkpy.helpers import constants as const + +_LOGGER = logging.getLogger(__name__) + + +class LoginHandler: + """Class to handle login communication.""" + + def __init__( + self, username=None, password=None, cred_file=None, + ): + """ + Initialize login handler. + + :param username: Blink username + :param password: Blink password + :param cred_file: JSON formatted credential file. + """ + self.login_url = None + self.login_urls = const.LOGIN_URLS + self.cred_file = cred_file + self.data = { + "username": username, + "password": password, + "uid": util.gen_uid(const.SIZE_UID), + "notification_key": util.gen_uid(const.SIZE_NOTIFICATION_KEY), + } + + def check_cred_file(self): + """Check if credential file supplied and use if so.""" + if isfile(self.cred_file): + try: + with open(self.cred_file, "r") as json_file: + creds = json.load(json_file) + self.data["username"] = creds["username"] + self.data["password"] = creds["password"] + + except ValueError: + _LOGGER.error( + "Improperly formatted json file %s.", self.cred_file, exc_info=True + ) + return False + + except KeyError: + _LOGGER.error("JSON file information incomplete %s.", exc_info=True) + return False + return True + return False + + def check_login(self): + """Check login information and prompt if not available.""" + if self.data["username"] is None: + self.data["username"] = input("Username:") + if self.data["password"] is None: + self.data["password"] = getpass("Password:") + + if self.data["username"] and self.data["password"]: + return True + return False + + def validate_response(self, url, response): + """Validate response from login endpoint.""" + try: + if response.status_code != 200: + return False + except AttributeError: + _LOGGER.error( + "Response for %s did not return a status code. Deprecated endpoint?", + url, + ) + return False + return True + + def login(self, blink): + """Attempt login to blink servers.""" + if self.cred_file is not None: + self.check_cred_file() + if not self.check_login(): + _LOGGER.error("Cannot login with username %s", self.data["username"]) + return False + + for url in self.login_urls: + _LOGGER.info("Attempting login with %s", url) + response = api.request_login( + blink, + url, + self.data["username"], + self.data["password"], + self.data["notification_key"], + self.data["uid"], + is_retry=False, + ) + + if self.validate_response(url, response): + self.login_url = url + return response.json() + + _LOGGER.error("Failed to login to Blink servers. Last response: %s", response) + return False diff --git a/tests/mock_responses.py b/tests/mock_responses.py index 14ec206..ea83776 100644 --- a/tests/mock_responses.py +++ b/tests/mock_responses.py @@ -7,6 +7,8 @@ LOGIN_RESPONSE = { "region": {"mock": "Test"}, "networks": {"1234": {"name": "test", "onboarded": True}}, "authtoken": {"authtoken": "foobar123", "message": "auth"}, + "client": {"id": "5678"}, + "account": {"id": "1337"}, } @@ -50,7 +52,7 @@ def mocked_session_send(*args, **kwargs): response = {"test": "foo"} status = 200 elif method == "POST": - if url in (const.LOGIN_URL, const.LOGIN_BACKUP_URL): + if url in const.LOGIN_URLS: response = LOGIN_RESPONSE status = 200 elif url == "http://wrong.url/" or url is None: diff --git a/tests/test_api.py b/tests/test_api.py deleted file mode 100644 index f22daaa..0000000 --- a/tests/test_api.py +++ /dev/null @@ -1,46 +0,0 @@ -"""Test various api functions.""" - -import unittest -from unittest import mock -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 - - @mock.patch("blinkpy.blinkpy.Blink.get_auth_token") - def test_http_req_connect_error(self, mock_auth): - """Test http_get error condition.""" - mock_auth.return_value = {"foo": "bar"} - firstlog = ( - "INFO:blinkpy.helpers.util:" "Cannot connect to server with url {}." - ).format("http://notreal.fake") - nextlog = ( - "INFO:blinkpy.helpers.util:" - "Auth token expired, attempting reauthorization." - ) - lastlog = ( - "ERROR:blinkpy.helpers.util:" - "Endpoint {} failed. Possible issue with " - "Blink servers." - ).format("http://notreal.fake") - expected = [firstlog, nextlog, firstlog, lastlog] - 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) diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index 775e1b5..cd1c056 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -50,7 +50,7 @@ class TestBlinkFunctions(unittest.TestCase): """Clean up after test.""" self.blink = None - @mock.patch("blinkpy.blinkpy.api.request_login") + @mock.patch("blinkpy.login_handler.api.request_login") def test_backup_url(self, req, mock_sess): """Test backup login method.""" json_resp = { @@ -60,16 +60,14 @@ class TestBlinkFunctions(unittest.TestCase): bad_req = mresp.MockResponse({}, 404) new_req = mresp.MockResponse(json_resp, 200) req.side_effect = [bad_req, bad_req, new_req] - self.blink.login_urls = ["test1", "test2", "test3"] - self.blink.login_request() - # pylint: disable=protected-access - self.assertEqual(self.blink._login_url, "test3") + self.blink.login_handler.login_urls = ["test1", "test2", "test3"] + self.blink.login_handler.login(self.blink) + self.assertEqual(self.blink.login_handler.login_url, "test3") req.side_effect = [bad_req, new_req, bad_req] - self.blink.login_urls = ["test1", "test2", "test3"] - self.blink.login_request() - # pylint: disable=protected-access - self.assertEqual(self.blink._login_url, "test2") + self.blink.login_handler.login_urls = ["test1", "test2", "test3"] + self.blink.login_handler.login(self.blink) + self.assertEqual(self.blink.login_handler.login_url, "test2") def test_merge_cameras(self, mock_sess): """Test merge camera functionality.""" diff --git a/tests/test_blinkpy.py b/tests/test_blinkpy.py index c76d43e..5a13482 100644 --- a/tests/test_blinkpy.py +++ b/tests/test_blinkpy.py @@ -14,7 +14,6 @@ from blinkpy.sync_module import BlinkSyncModule from blinkpy.helpers.util import ( http_req, create_session, - BlinkAuthenticationException, BlinkException, BlinkURLHandler, ) @@ -31,7 +30,6 @@ class TestBlinkSetup(unittest.TestCase): def setUp(self): """Set up Blink module.""" - self.blink_no_cred = Blink() self.blink = Blink(username=USERNAME, password=PASSWORD) self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", "1234", []) self.blink.urls = BlinkURLHandler("test") @@ -40,84 +38,12 @@ class TestBlinkSetup(unittest.TestCase): 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.""" self.assertEqual(self.blink.version, __version__) - # 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) - - @mock.patch("blinkpy.blinkpy.getpass.getpass") - @mock.patch("blinkpy.blinkpy.Blink.get_auth_token") - def test_no_cred_file(self, getpwd, getauth, mock_sess): - """Check that normal login occurs when cred file doesn't exist.""" - # pylint: disable=protected-access - self.blink._cred_file = "/tmp/fake.file" - getpwd.return_value = PASSWORD - getauth.return_value = True - with mock.patch("builtins.input", return_value=USERNAME): - self.assertTrue(self.blink.login()) - - def test_exit_on_missing_json(self, mock_sess): - """Test that we fail on missing json data.""" - # pylint: disable=protected-access - self.blink._cred_file = "/tmp/fake.file" - with mock.patch("os.path.isfile", return_value=True): - with mock.patch("builtins.open", mock.mock_open(read_data="{}")): - self.assertFalse(self.blink.login()) - - def test_exit_on_bad_json(self, mock_sess): - """Test that we fail on bad json format.""" - # pylint: disable=protected-access - self.blink._cred_file = "/tmp/fake.file" - with mock.patch("os.path.isfile", return_value=True): - with mock.patch("builtins.open", mock.mock_open(read_data="{]")): - self.assertFalse(self.blink.login()) - - @mock.patch("blinkpy.blinkpy.json.load") - def test_cred_file(self, mockjson, mock_sess): - """Test that loading credential file works.""" - # pylint: disable=protected-access - self.blink_no_cred._cred_file = "/tmp/fake.file" - mockjson.return_value = {"username": "foo", "password": "bar"} - with mock.patch("os.path.isfile", return_value=True): - with mock.patch("builtins.open", mock.mock_open(read_data="")): - self.assertTrue(self.blink_no_cred.login()) - # pylint: disable=protected-access - self.assertEqual(self.blink_no_cred._username, "foo") - # pylint: disable=protected-access - self.assertEqual(self.blink_no_cred._password, "bar") + self.assertEqual(self.blink.login_handler.data["username"], USERNAME) + self.assertEqual(self.blink.login_handler.data["password"], PASSWORD) def test_bad_request(self, mock_sess): """Check that we raise an Exception with a bad request.""" @@ -147,15 +73,8 @@ class TestBlinkSetup(unittest.TestCase): 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): + def test_multiple_networks(self, 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, "name": "foo"}, "5678": {"onboarded": True, "name": "bar"}, @@ -163,26 +82,18 @@ class TestBlinkSetup(unittest.TestCase): } self.blink.get_ids() self.assertTrue("5678" in self.blink.network_ids) - self.assertEqual(self.blink.account_id, 2222) - @mock.patch("blinkpy.api.request_networks") - def test_multiple_onboarded_networks(self, mock_net, mock_sess): + def test_multiple_onboarded_networks(self, mock_sess): """Check that we handle multiple networks appropriately.""" - mock_net.return_value = { - "networks": [ - {"id": 0000, "account_id": 2222}, - {"id": 5678, "account_id": 1111}, - ] - } self.blink.networks = { "0000": {"onboarded": False, "name": "foo"}, "5678": {"onboarded": True, "name": "bar"}, "1234": {"onboarded": True, "name": "test"}, } self.blink.get_ids() + self.assertTrue("0000" not in self.blink.network_ids) self.assertTrue("5678" in self.blink.network_ids) self.assertTrue("1234" in self.blink.network_ids) - self.assertEqual(self.blink.account_id, 1111) @mock.patch("blinkpy.blinkpy.time.time") def test_throttle(self, mock_time, mock_sess): diff --git a/tests/test_login_handler.py b/tests/test_login_handler.py new file mode 100644 index 0000000..79fb355 --- /dev/null +++ b/tests/test_login_handler.py @@ -0,0 +1,70 @@ +"""Test login handler.""" + +import unittest +from unittest import mock +from blinkpy.login_handler import LoginHandler +import tests.mock_responses as mresp + +USERNAME = "foobar" +PASSWORD = "deadbeef" + + +@mock.patch("blinkpy.helpers.util.Session.send", side_effect=mresp.mocked_session_send) +class TestLoginHandler(unittest.TestCase): + """Test the LoginHandler class in blinkpy.""" + + def setUp(self): + """Set up Login Handler.""" + self.login_handler = LoginHandler() + + def tearDown(self): + """Clean up after test.""" + self.login_handler = None + + @mock.patch("blinkpy.login_handler.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.login_handler.check_login()) + self.assertEqual(self.login_handler.data["username"], USERNAME) + self.assertEqual(self.login_handler.data["password"], PASSWORD) + + def test_no_cred_file(self, mock_sess): + """Check that we return false when cred file doesn't exist.""" + self.login_handler.cred_file = "/tmp/fake.file" + self.assertFalse(self.login_handler.check_cred_file()) + + @mock.patch("blinkpy.login_handler.isfile") + def test_exit_on_missing_json(self, mockisfile, mock_sess): + """Test that we fail on missing json data.""" + self.login_handler.cred_file = "/tmp/fake.file" + mockisfile.return_value = True + with mock.patch("builtins.open", mock.mock_open(read_data="{}")): + self.assertFalse(self.login_handler.check_cred_file()) + + @mock.patch("blinkpy.login_handler.json.load") + @mock.patch("blinkpy.login_handler.isfile") + def test_cred_file(self, mockisfile, mockjson, mock_sess): + """Test that loading credential file works.""" + self.login_handler.cred_file = "/tmp/fake.file" + mockjson.return_value = {"username": "foo", "password": "bar"} + mockisfile.return_value = True + with mock.patch("builtins.open", mock.mock_open(read_data="")): + self.assertTrue(self.login_handler.check_cred_file()) + self.assertEqual(self.login_handler.data["username"], "foo") + self.assertEqual(self.login_handler.data["password"], "bar") + + def test_bad_response(self, mock_sess): + """Check bad response from server.""" + self.assertFalse(self.login_handler.validate_response(None, None)) + + def test_bad_response_code(self, mock_sess): + """Check bad response code from server.""" + fake_resp = mresp.MockResponse(None, 404) + self.assertFalse(self.login_handler.validate_response(None, fake_resp)) + + def test_good_response_code(self, mock_sess): + """Check good response code from server.""" + fake_resp = mresp.MockResponse(None, 200) + self.assertTrue(self.login_handler.validate_response(None, fake_resp))