diff --git a/src/elmo/api/client.py b/src/elmo/api/client.py index cd80354..cb3f649 100644 --- a/src/elmo/api/client.py +++ b/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 diff --git a/tests/test_client.py b/tests/test_client.py index 2ff989e..e6901ce 100644 --- a/tests/test_client.py +++ b/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,7 +413,7 @@ 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") @@ -381,20 +421,6 @@ def test_client_auth_econnect_web_login_metronet(server): 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 = """