Merge pull request #319 from fronzbot/login-fix

Changed the way the new login method switch is handled

app/app.py

@@ -1,11 +1,12 @@
 """Script to run blinkpy as an app."""
 from os import environ
 from datetime import datetime, timedelta
-from blinkpy import blinkpy
+from blinkpy.blinkpy import Blink
+from blinkpy.auth import Auth
+from blinkpy.helpers.util import json_load
 
 
-USERNAME = environ.get("USERNAME")
-PASSWORD = environ.get("PASSWORD")
+CREDFILE = environ.get("CREDFILE")
 TIMEDELTA = timedelta(environ.get("TIMEDELTA", 1))
 
 
@@ -21,11 +22,18 @@ def download_videos(blink, save_dir="/media"):
 
 def start():
     """Startup blink app."""
-    blink = blinkpy.Blink(username=USERNAME, password=PASSWORD)
+    blink = Blink()
+    blink.auth = Auth(json_load(CREDFILE))
     blink.start()
     return blink
 
 
+def main():
+    """Run the app."""
+    blink = start()
+    download_videos(blink)
+    blink.save(CREDFILE)
+
+
 if __name__ == "__main__":
-    BLINK = start()
-    download_videos(BLINK)
+    main()

blinkpy/api.py

@@ -32,7 +32,7 @@ def request_login(
             "client_name": "Computer",
             "client_type": "android",
             "os_version": "5.1.1",
-            "reauth": "true",
+            "reauth": "false",
         }
     )
     return auth.query(

blinkpy/auth.py

@@ -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_method = login_method
         self.login_response = None
         self.is_errored = False
         self.no_prompt = no_prompt
@@ -53,6 +55,13 @@ class Auth:
             return None
         return {"Host": self.host, "TOKEN_AUTH": self.token}
 
+    @property
+    def login_url(self):
+        """Return login url."""
+        if self.login_method not in LOGIN_ENDPOINT:
+            return LOGIN_ENDPOINT["v4"]
+        return LOGIN_ENDPOINT[self.login_method]
+
     def create_session(self):
         """Create a session for blink communication."""
         sess = Session()
@@ -73,11 +82,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()

blinkpy/blinkpy.py

@@ -100,6 +100,7 @@ class Blink:
             self.auth.startup()
             self.setup_login_ids()
             self.setup_urls()
+            self.get_homescreen()
         except (LoginError, TokenRefreshFailed, BlinkSetupError):
             _LOGGER.error("Cannot setup Blink platform.")
             self.available = False
@@ -144,17 +145,20 @@ class Blink:
         self.sync[name] = BlinkSyncModule(self, name, network_id, cameras)
         self.sync[name].start()
 
-    def setup_owls(self):
-        """Check for mini cameras."""
+    def get_homescreen(self):
+        """Get homecreen information."""
         if self.no_owls:
             _LOGGER.debug("Skipping owl extraction.")
-            return []
-        response = api.request_homescreen(self)
-        self.homescreen = response
+            self.homescreen = {}
+            return
+        self.homescreen = api.request_homescreen(self)
+
+    def setup_owls(self):
+        """Check for mini cameras."""
         network_list = []
         camera_list = []
         try:
-            for owl in response["owls"]:
+            for owl in self.homescreen["owls"]:
                 name = owl["name"]
                 network_id = str(owl["network_id"])
                 if network_id in self.network_ids:

blinkpy/helpers/constants.py

@@ -4,7 +4,7 @@ import os
 
 MAJOR_VERSION = 0
 MINOR_VERSION = 16
-PATCH_VERSION = "0-rc5"
+PATCH_VERSION = "0-rc10"
 
 __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

blinkpy/sync_module.py

@@ -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
 

requirements.txt

@@ -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

requirements_test.txt

@@ -6,8 +6,8 @@ pre-commit==2.5.1
 pylint==2.5.3
 pydocstyle==5.0.2
 pytest==5.4.3
-pytest-cov==2.9.0
+pytest-cov==2.10.0
 pytest-sugar==0.9.3
-pytest-timeout==1.3.4
+pytest-timeout==1.4.1
 restructuredtext-lint==1.3.1
 pygments==2.6.1

tests/test_auth.py

@@ -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")
@@ -239,6 +254,15 @@ class TestAuth(unittest.TestCase):
         mock_validate.side_effect = [UnauthorizedError, TokenRefreshFailed]
         self.assertEqual(self.auth.query(url="http://example.com"), None)
 
+    def test_login_methods(self):
+        """Test correct login url returned."""
+        auth = Auth()
+        self.assertEqual(auth.login_url, const.LOGIN_ENDPOINT["v4"])
+        auth.login_method = "v3"
+        self.assertEqual(auth.login_url, const.LOGIN_ENDPOINT["v3"])
+        auth.login_method = "foobar"
+        self.assertEqual(auth.login_url, const.LOGIN_ENDPOINT["v4"])
+
 
 class MockSession:
     """Object to mock a session."""

tests/test_blinkpy.py

@@ -197,12 +197,11 @@ class TestBlinkSetup(unittest.TestCase):
         self.assertEqual(combined["fizz"], "buzz")
         self.assertEqual(combined["bar"], "foo")
 
-    @mock.patch("blinkpy.api.request_homescreen")
     @mock.patch("blinkpy.blinkpy.BlinkOwl.start")
-    def test_initialize_blink_minis(self, mock_start, mock_home):
+    def test_initialize_blink_minis(self, mock_start):
         """Test blink mini initialization."""
         mock_start.return_value = True
-        mock_home.return_value = {
+        self.blink.homescreen = {
             "owls": [
                 {
                     "enabled": False,
@@ -235,11 +234,10 @@ class TestBlinkSetup(unittest.TestCase):
         self.assertEqual(self.blink.sync["foo"].name, "foo")
         self.assertEqual(self.blink.sync["bar"].name, "bar")
 
-    @mock.patch("blinkpy.api.request_homescreen")
-    def test_blink_mini_cameras_returned(self, mock_home):
+    def test_blink_mini_cameras_returned(self):
         """Test that blink mini cameras are found if attached to sync module."""
         self.blink.network_ids = ["1234"]
-        mock_home.return_value = {
+        self.blink.homescreen = {
             "owls": [
                 {
                     "id": 1,
@@ -261,16 +259,16 @@ class TestBlinkSetup(unittest.TestCase):
 
         self.blink.no_owls = True
         self.blink.network_ids = []
+        self.blink.get_homescreen()
         result = self.blink.setup_owls()
         self.assertEqual(self.blink.network_ids, [])
         self.assertEqual(result, [])
 
-    @mock.patch("blinkpy.api.request_homescreen")
     @mock.patch("blinkpy.api.request_camera_usage")
-    def test_blink_mini_attached_to_sync(self, mock_usage, mock_home):
+    def test_blink_mini_attached_to_sync(self, mock_usage):
         """Test that blink mini cameras are properly attached to sync module."""
         self.blink.network_ids = ["1234"]
-        mock_home.return_value = {
+        self.blink.homescreen = {
             "owls": [
                 {
                     "id": 1,

tests/test_sync_module.py

@@ -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."""