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
40 changes: 19 additions & 21 deletions src/elmo/api/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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"]:
Expand All @@ -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

Expand Down
62 changes: 44 additions & 18 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = """
Expand Down Expand Up @@ -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",
Expand All @@ -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 = """
Expand Down