fix(api): use weblogin access token with API base_url redirect

Author: palazzem

src/elmo/api/client.py

@@ -53,12 +53,14 @@ def __init__(self, base_url=None, domain=None, session_id=None):
         self._session = Session()
         self._session_id = session_id
         self._panel = None
+        self._web_login = base_url == ELMO_E_CONNECT
         self._lock = Lock()
 
         # Debug
         _LOGGER.debug(f"Client | Library version: {__version__}")
         _LOGGER.debug(f"Client | Router: {self._router._base_url}")
         _LOGGER.debug(f"Client | Domain: {self._domain}")
+        _LOGGER.debug(f"Client | Web login: {self._web_login}")
 
     def auth(self, username, password):
         """Authenticate the client and retrieves the access token. This method uses
@@ -75,23 +77,6 @@ def auth(self, username, password):
             the `ElmoClient` instance.
         """
         try:
-            if self._router._base_url == ELMO_E_CONNECT:
-                # Web login is required for Elmo E-Connect because, at the moment, the
-                # e-Connect Cloud API login does not register the client session in the backend.
-                # This prevents the client from attaching to server events (e.g. long polling updates).
-                web_login_url = f"https://webservice.elmospa.com/{self._domain}"
-                payload = {
-                    "IsDisableAccountCreation": "True",
-                    "IsAllowThemeChange": "True",
-                    "UserName": username,
-                    "Password": password,
-                    "RememberMe": "false",
-                }
-                _LOGGER.debug("Client | e-Connect Web Login detected")
-                web_response = self._session.post(web_login_url, data=payload)
-                web_response.raise_for_status()
-
-            # API login
             payload = {"username": username, "password": password}
             if self._domain is not None:
                 payload["domain"] = self._domain
@@ -107,11 +92,8 @@ def auth(self, username, password):
 
         # Store the session_id and the panel details (if available)
         data = response.json()
+        self._session_id = data["SessionId"]
         self._panel = {_camel_to_snake_case(k): v for k, v in data.get("Panel", {}).items()}
-        if self._router._base_url == ELMO_E_CONNECT:
-            self._session_id = extract_session_id_from_html(web_response.text)
-        else:
-            self._session_id = data["SessionId"]
 
         # Register the redirect URL and try the authentication again
         if data["Redirect"]:
@@ -122,6 +104,22 @@ def auth(self, username, password):
             data = redirect.json()
             self._session_id = data["SessionId"]
 
+        if self._web_login:
+            # Web login is required for Elmo E-Connect because, at the moment, the
+            # e-Connect Cloud API login does not register the client session in the backend.
+            # This prevents the client from attaching to server events (e.g. long polling updates).
+            web_login_url = f"https://webservice.elmospa.com/{self._domain}"
+            payload = {
+                "IsDisableAccountCreation": "True",
+                "IsAllowThemeChange": "True",
+                "UserName": username,
+                "Password": password,
+                "RememberMe": "false",
+            }
+            web_response = self._session.post(web_login_url, data=payload)
+            web_response.raise_for_status()
+            self._session_id = extract_session_id_from_html(web_response.text)
+
         _LOGGER.debug(f"Client | Authentication successful: {_sanitize_session_id(self._session_id)}")
         return self._session_id
 

tests/test_client.py

@@ -239,6 +239,46 @@ def test_client_auth_redirect(server):
     assert len(server.calls) == 2
 
 
+def test_client_auth_redirect_web_login(server):
+    """Ensure web login session token is used when redirect is required.
+    Regression test: https://github.com/palazzem/econnect-python/issues/158
+    """
+    redirect = """
+        {
+            "SessionId": "00000000-0000-0000-0000-000000000000",
+            "Domain": "domain",
+            "Redirect": true,
+            "RedirectTo": "https://redirect.example.com"
+        }
+    """
+    login = """
+        {
+            "SessionId": "99999999-9999-9999-9999-999999999999",
+            "Username": "test",
+            "Domain": "domain",
+            "Language": "en",
+            "IsActivated": true,
+            "IsConnected": true,
+            "IsLoggedIn": false,
+            "IsLoginInProgress": false,
+            "CanElevate": true,
+            "AccountId": 100,
+            "IsManaged": false,
+            "Redirect": false,
+            "IsElevation": false
+        }
+    """
+    server.add(responses.GET, "https://connect.elmospa.com/api/login", body=redirect, status=200)
+    server.add(responses.GET, "https://redirect.example.com/api/login", body=login, status=200)
+    server.add(responses.POST, "https://webservice.elmospa.com/domain", body=r.STATUS_PAGE, status=200)
+    client = ElmoClient(base_url=ELMO_E_CONNECT, domain="domain")
+    # Test
+    assert client.auth("test", "test")
+    assert len(server.calls) == 3
+    assert client._router._base_url == "https://redirect.example.com"
+    assert client._session_id == "f8h23b4e-7a9f-4d3f-9b08-2769263ee33c"
+
+
 def test_client_auth_infinite_redirect(server):
     """Should prevent infinite redirects in the auth() call."""
     redirect = """
@@ -349,16 +389,16 @@ def test_client_poll(server):
 
 
 def test_client_auth_econnect_web_login(server):
-    """Web login should be used when accessing with e-Connect.
+    """Ensure API and Web login are executed when using e-Connect cloud API.
     Regression test: https://github.com/palazzem/econnect-python/issues/158
     """
+    server.add(responses.GET, "https://connect.elmospa.com/api/login", body=r.LOGIN, status=200)
     server.add(responses.POST, "https://webservice.elmospa.com/domain", body=r.STATUS_PAGE, status=200)
-    server.add(responses.GET, f"{ELMO_E_CONNECT}/api/login", body=r.LOGIN, status=200)
     client = ElmoClient(base_url=ELMO_E_CONNECT, domain="domain")
     # Test
     client.auth("test", "test")
-    request_body = dict(item.split("=") for item in server.calls[0].request.body.split("&"))
     assert len(server.calls) == 2
+    request_body = dict(item.split("=") for item in server.calls[1].request.body.split("&"))
     assert client._session_id == "f8h23b4e-7a9f-4d3f-9b08-2769263ee33c"
     assert request_body == {
         "IsDisableAccountCreation": "True",
@@ -373,28 +413,14 @@ def test_client_auth_econnect_web_login_metronet(server):
     """Web login should NOT be used when accessing with Metronet.
     Regression test: https://github.com/palazzem/econnect-python/issues/158
     """
-    server.add(responses.GET, f"{IESS_METRONET}/api/login", body=r.LOGIN, status=200)
+    server.add(responses.GET, "https://metronet.iessonline.com/api/login", body=r.LOGIN, status=200)
     client = ElmoClient(base_url=IESS_METRONET, domain="domain")
     # Test
     client.auth("test", "test")
     assert client._session_id == "00000000-0000-0000-0000-000000000000"
     assert len(server.calls) == 1
 
 
-def test_client_auth_econnect_web_login_forbidden(server):
-    """Should raise an exception if credentials are not valid in the web login form."""
-    server.add(
-        responses.POST, "https://webservice.elmospa.com/domain", body="Username or Password is invalid", status=403
-    )
-    client = ElmoClient(base_url=ELMO_E_CONNECT, domain="domain")
-    # Test
-    with pytest.raises(CredentialError):
-        client.auth("test", "test")
-    assert client._session_id is None
-    assert client._panel is None
-    assert len(server.calls) == 1
-
-
 def test_client_poll_with_changes(server):
     """Should return a dict with updated states."""
     html = """