Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f8db8eb3b6 | ||
|
|
a34b6dec67 | ||
|
|
bded3eb156 | ||
|
|
9ca73934ce | ||
|
|
e02b429523 | ||
|
|
22e4b7dabe |
+6
-4
@@ -12,7 +12,7 @@ _LOGGER = logging.getLogger(__name__)
|
||||
class Auth:
|
||||
"""Class to handle login communication."""
|
||||
|
||||
def __init__(self, login_data=None, no_prompt=False):
|
||||
def __init__(self, login_data=None, no_prompt=False, login_method="v4"):
|
||||
"""
|
||||
Initialize auth handler.
|
||||
|
||||
@@ -22,6 +22,7 @@ class Auth:
|
||||
- password
|
||||
:param no_prompt: Should any user input prompts
|
||||
be supressed? True/FALSE
|
||||
:param login_method: Choose the login endpoint to use. Default: v4. v3 uses email verification rather than a 2FA code.
|
||||
"""
|
||||
if login_data is None:
|
||||
login_data = {}
|
||||
@@ -31,6 +32,7 @@ class Auth:
|
||||
self.region_id = login_data.get("region_id", None)
|
||||
self.client_id = login_data.get("client_id", None)
|
||||
self.account_id = login_data.get("account_id", None)
|
||||
self.login_url = LOGIN_ENDPOINT[login_method]
|
||||
self.login_response = None
|
||||
self.is_errored = False
|
||||
self.no_prompt = no_prompt
|
||||
@@ -73,11 +75,11 @@ class Auth:
|
||||
|
||||
self.data = util.validate_login_data(self.data)
|
||||
|
||||
def login(self, login_url=LOGIN_ENDPOINT):
|
||||
def login(self):
|
||||
"""Attempt login to blink servers."""
|
||||
self.validate_login()
|
||||
_LOGGER.info("Attempting login with %s", login_url)
|
||||
response = api.request_login(self, login_url, self.data, is_retry=False,)
|
||||
_LOGGER.info("Attempting login with %s", self.login_url)
|
||||
response = api.request_login(self, self.login_url, self.data, is_retry=False,)
|
||||
try:
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
|
||||
@@ -4,7 +4,7 @@ import os
|
||||
|
||||
MAJOR_VERSION = 0
|
||||
MINOR_VERSION = 16
|
||||
PATCH_VERSION = "0-rc7"
|
||||
PATCH_VERSION = "0-rc9"
|
||||
|
||||
__version__ = f"{MAJOR_VERSION}.{MINOR_VERSION}.{PATCH_VERSION}"
|
||||
|
||||
@@ -48,7 +48,10 @@ URLS
|
||||
BLINK_URL = "immedia-semi.com"
|
||||
DEFAULT_URL = f"rest-prod.{BLINK_URL}"
|
||||
BASE_URL = f"https://{DEFAULT_URL}"
|
||||
LOGIN_ENDPOINT = f"{BASE_URL}/api/v4/account/login"
|
||||
LOGIN_ENDPOINT = {
|
||||
"v4": f"{BASE_URL}/api/v4/account/login",
|
||||
"v3": f"{BASE_URL}/api/v3/login",
|
||||
}
|
||||
|
||||
"""
|
||||
Dictionaries
|
||||
|
||||
@@ -171,7 +171,7 @@ class BlinkSyncModule:
|
||||
return response["camera"][0]
|
||||
except (TypeError, KeyError):
|
||||
_LOGGER.error("Could not extract camera info: %s", response, exc_info=True)
|
||||
return []
|
||||
return {}
|
||||
|
||||
def get_network_info(self):
|
||||
"""Retrieve network status."""
|
||||
@@ -191,7 +191,9 @@ class BlinkSyncModule:
|
||||
self.check_new_videos()
|
||||
for camera_name in self.cameras.keys():
|
||||
camera_id = self.cameras[camera_name].camera_id
|
||||
camera_info = self.get_camera_info(camera_id)
|
||||
camera_info = self.get_camera_info(
|
||||
camera_id, owl_info=self.get_owl_info(camera_name)
|
||||
)
|
||||
self.cameras[camera_name].update(camera_info, force_cache=force_cache)
|
||||
self.available = True
|
||||
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
python-dateutil~=2.8.1
|
||||
requests~=2.23.0
|
||||
requests~=2.24.0
|
||||
python-slugify~=4.0.0
|
||||
testtools==2.4.0
|
||||
|
||||
@@ -144,6 +144,21 @@ class TestAuth(unittest.TestCase):
|
||||
fake_resp = mresp.MockResponse({"foo": "bar"}, 200)
|
||||
mock_req.return_value = fake_resp
|
||||
self.assertEqual(self.auth.login(), {"foo": "bar"})
|
||||
mock_req.assert_called_with(
|
||||
self.auth, const.LOGIN_ENDPOINT["v4"], {}, is_retry=False
|
||||
)
|
||||
|
||||
@mock.patch("blinkpy.auth.Auth.validate_login", return_value=None)
|
||||
@mock.patch("blinkpy.auth.api.request_login")
|
||||
def test_login_v3(self, mock_req, mock_validate):
|
||||
"""Test login handling."""
|
||||
auth_v3 = Auth(login_method="v3")
|
||||
fake_resp = mresp.MockResponse({"foo": "bar"}, 200)
|
||||
mock_req.return_value = fake_resp
|
||||
self.assertEqual(auth_v3.login(), {"foo": "bar"})
|
||||
mock_req.assert_called_with(
|
||||
auth_v3, const.LOGIN_ENDPOINT["v3"], {}, is_retry=False
|
||||
)
|
||||
|
||||
@mock.patch("blinkpy.auth.Auth.validate_login", return_value=None)
|
||||
@mock.patch("blinkpy.auth.api.request_login")
|
||||
|
||||
@@ -80,11 +80,11 @@ class TestBlinkSyncModule(unittest.TestCase):
|
||||
def test_get_camera_info_fail(self, mock_resp):
|
||||
"""Test handling of failed get camera info function."""
|
||||
mock_resp.return_value = None
|
||||
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), [])
|
||||
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), {})
|
||||
mock_resp.return_value = {}
|
||||
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), [])
|
||||
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), {})
|
||||
mock_resp.return_value = {"camera": None}
|
||||
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), [])
|
||||
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), {})
|
||||
|
||||
def test_get_network_info(self, mock_resp):
|
||||
"""Test network retrieval."""
|
||||
|
||||
Reference in New Issue
Block a user