Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 21 additions & 8 deletions src/elmo/api/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,12 @@

from .. import query as q
from ..__about__ import __version__
from ..systems import ELMO_E_CONNECT
from ..systems import (
ELMO_E_CONNECT,
ELMO_E_CONNECT_WEB_LOGIN,
IESS_METRONET,
IESS_METRONET_WEB_LOGIN,
)
from ..utils import (
_camel_to_snake_case,
_sanitize_session_id,
Expand Down Expand Up @@ -53,14 +58,25 @@ 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()

# Web login is required for E-Connect and Metronet because, at the moment, the
# Cloud API login does not register the client session in their backend, causing
# updates to be not received by the client during the long-polling `has_updates`.
domain_url = "" if self._domain == "default" else self._domain
self._web_login = base_url in [ELMO_E_CONNECT, IESS_METRONET]
if base_url == ELMO_E_CONNECT:
self._web_login_url = f"{ELMO_E_CONNECT_WEB_LOGIN}/{domain_url}"
elif base_url == IESS_METRONET:
self._web_login_url = f"{IESS_METRONET_WEB_LOGIN}/{domain_url}"
else:
self._web_login_url = None

# 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}")
_LOGGER.debug(f"Client | Web login URL: {self._web_login_url}")

def auth(self, username, password):
"""Authenticate the client and retrieves the access token. This method uses
Expand Down Expand Up @@ -104,19 +120,16 @@ def auth(self, username, password):
data = redirect.json()
self._session_id = data["SessionId"]

# Retrieve the session_id using the web login form
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 = self._session.post(self._web_login_url, data=payload)
web_response.raise_for_status()
self._session_id = extract_session_id_from_html(web_response.text)

Expand Down
2 changes: 2 additions & 0 deletions src/elmo/systems.py
Original file line number Diff line number Diff line change
@@ -1,2 +1,4 @@
ELMO_E_CONNECT = "https://connect.elmospa.com"
ELMO_E_CONNECT_WEB_LOGIN = "https://webservice.elmospa.com"
IESS_METRONET = "https://metronet.iessonline.com"
IESS_METRONET_WEB_LOGIN = "https://metronet.iessonline.com"
101 changes: 96 additions & 5 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ def test_client_auth_redirect(server):
assert len(server.calls) == 2


def test_client_auth_redirect_web_login(server):
def test_client_auth_redirect_web_login_econnect(server):
"""Ensure web login session token is used when redirect is required.
Regression test: https://github.com/palazzem/econnect-python/issues/158
"""
Expand Down Expand Up @@ -279,6 +279,46 @@ def test_client_auth_redirect_web_login(server):
assert client._session_id == "f8h23b4e-7a9f-4d3f-9b08-2769263ee33c"


def test_client_auth_redirect_web_login_metronet(server):
"""Ensure web login session token is used when redirect is required.
Regression test: https://github.com/palazzem/econnect-python/issues/168
"""
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://metronet.iessonline.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://metronet.iessonline.com/domain", body=r.STATUS_PAGE, status=200)
client = ElmoClient(base_url=IESS_METRONET, 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 = """
Expand Down Expand Up @@ -409,16 +449,67 @@ def test_client_auth_econnect_web_login(server):
}


def test_client_auth_econnect_web_login_metronet(server):
"""Web login should NOT be used when accessing with Metronet.
def test_client_auth_econnect_web_login_with_default_domain(server):
"""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/", body=r.STATUS_PAGE, status=200)
client = ElmoClient(base_url=ELMO_E_CONNECT, domain="default")
# Test
client.auth("test", "test")
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",
"IsAllowThemeChange": "True",
"UserName": "test",
"Password": "test",
"RememberMe": "false",
}


def test_client_auth_econnect_web_login_metronet(server):
"""Web login must be used when accessing with Metronet.
Regression test: https://github.com/palazzem/econnect-python/issues/186
"""
server.add(responses.GET, "https://metronet.iessonline.com/api/login", body=r.LOGIN, status=200)
server.add(responses.POST, "https://metronet.iessonline.com/domain", body=r.STATUS_PAGE, 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
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",
"IsAllowThemeChange": "True",
"UserName": "test",
"Password": "test",
"RememberMe": "false",
}


def test_client_auth_metronet_web_login_with_default_domain(server):
"""Web login must be used when accessing with Metronet.
Regression test: https://github.com/palazzem/econnect-python/issues/186
"""
server.add(responses.GET, "https://metronet.iessonline.com/api/login", body=r.LOGIN, status=200)
server.add(responses.POST, "https://metronet.iessonline.com/", body=r.STATUS_PAGE, status=200)
client = ElmoClient(base_url=IESS_METRONET, domain="default")
# Test
client.auth("test", "test")
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",
"IsAllowThemeChange": "True",
"UserName": "test",
"Password": "test",
"RememberMe": "false",
}


def test_client_poll_with_changes(server):
Expand Down