diff --git a/src/elmo/api/client.py b/src/elmo/api/client.py index 79b2b38..cd80354 100644 --- a/src/elmo/api/client.py +++ b/src/elmo/api/client.py @@ -9,7 +9,12 @@ from .. import query as q from ..__about__ import __version__ -from ..utils import _camel_to_snake_case, _sanitize_session_id +from ..systems import ELMO_E_CONNECT +from ..utils import ( + _camel_to_snake_case, + _sanitize_session_id, + extract_session_id_from_html, +) from .decorators import require_lock, require_session from .exceptions import ( CodeError, @@ -49,6 +54,7 @@ def __init__(self, base_url=None, domain=None, session_id=None): self._session_id = session_id self._panel = None self._lock = Lock() + # Debug _LOGGER.debug(f"Client | Library version: {__version__}") _LOGGER.debug(f"Client | Router: {self._router._base_url}") @@ -56,7 +62,7 @@ def __init__(self, base_url=None, domain=None, session_id=None): def auth(self, username, password): """Authenticate the client and retrieves the access token. This method uses - the Authentication API. + the Authentication API, or the web login form if the base_url is Elmo E-Connect. Args: username: the Username used for the authentication. @@ -69,11 +75,28 @@ 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 - _LOGGER.debug("Client | Client Authentication") + _LOGGER.debug("Client | API Authentication") response = self._session.get(self._router.auth, params=payload) response.raise_for_status() except HTTPError as err: @@ -84,8 +107,11 @@ 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"]: diff --git a/src/elmo/utils.py b/src/elmo/utils.py index b5cb1f9..6caf7ec 100644 --- a/src/elmo/utils.py +++ b/src/elmo/utils.py @@ -2,6 +2,8 @@ import re from functools import lru_cache +from .api.exceptions import ParseError + _LOGGER = logging.getLogger(__name__) @@ -58,3 +60,29 @@ def _camel_to_snake_case(name): name = name.lower() return name + + +def extract_session_id_from_html(html_content: str) -> str: + """Extract the session ID from the HTML source containing the specific JavaScript block. + + This function uses a raw string (r"") for the regex pattern to avoid escaping issues. + The regex pattern is designed to find "var sessionId = '...'" and capture the ID within the quotes. + It captures any character except the closing single quote. + + Args: + html_content (str): The HTML source code as a string. + + Returns: + str: The extracted session ID string. + + Raises: + ParseError: If the session ID is not found in the HTML content. + """ + pattern = r"var\s+sessionId\s*=\s*'([^']+)'" + match = re.search(pattern, html_content) + if match: + return match.group(1) + else: + _LOGGER.error("Client | Session ID not found in e-Connect status page.") + _LOGGER.debug("Client | HTML content: %s", html_content) + raise ParseError("Session ID not found in e-Connect status page.") diff --git a/tests/fixtures/responses.py b/tests/fixtures/responses.py index 295a58e..792f502 100644 --- a/tests/fixtures/responses.py +++ b/tests/fixtures/responses.py @@ -96,6 +96,7 @@ def test_client_get_sectors_status(server): "PrivacyLink": "/PrivacyAndTerms/v1/Informativa_privacy_econnect_2020_09.pdf", "TermsLink": "/PrivacyAndTerms/v1/CONTRATTO_UTILIZZATORE_FINALE_2020_02_07.pdf" }""" + UPDATES = """ { "ConnectionStatus": false, @@ -118,6 +119,7 @@ def test_client_get_sectors_status(server): "HasChanges": true } """ + SYNC_LOGIN = """[ { "Poller": {"Poller": 1, "Panel": 1}, @@ -125,6 +127,7 @@ def test_client_get_sectors_status(server): "Successful": true } ]""" + SYNC_LOGOUT = """[ { "Poller": {"Poller": 1, "Panel": 1}, @@ -132,6 +135,7 @@ def test_client_get_sectors_status(server): "Successful": true } ]""" + SYNC_SEND_COMMAND = """[ { "Poller": {"Poller": 1, "Panel": 1}, @@ -139,6 +143,7 @@ def test_client_get_sectors_status(server): "Successful": true } ]""" + STRINGS = """[ { "AccountId": 1, @@ -229,6 +234,7 @@ def test_client_get_sectors_status(server): "Version": "AAAAAAAAgRw=" } ]""" + AREAS = """[ { "Active": true, @@ -283,6 +289,7 @@ def test_client_get_sectors_status(server): "InProgress": false } ]""" + INPUTS = """[ { "Alarm": true, @@ -333,6 +340,7 @@ def test_client_get_sectors_status(server): "InProgress": false } ]""" + OUTPUTS = """[ { "Active": true, @@ -379,3 +387,341 @@ def test_client_get_sectors_status(server): "InProgress": false } ]""" + +STATUS_PAGE = """ + +""" diff --git a/tests/scripts/debug_polling.py b/tests/scripts/debug_polling.py new file mode 100644 index 0000000..1550122 --- /dev/null +++ b/tests/scripts/debug_polling.py @@ -0,0 +1,65 @@ +""" +Debug script for polling e-Connect/IESS alarm systems. + +This script connects to an e-Connect/IESS alarm system via the E-Connect or Metronet +cloud platform, authenticates using provided credentials, and continuously polls +for updates on sectors, inputs, outputs, alerts, and panel status. + +It prints the 'last_id' received for each category in every poll cycle. + +Usage: + python tests/scripts/debug_polling.py + +Arguments: + domain: The domain identifier for the alarm system (e.g., vendor). + system: The cloud platform to connect to ('econnect' or 'iess'). + username: The username for authentication. + password: The password for authentication. +""" + +import argparse + +from elmo import query as q +from elmo import systems +from elmo.api.client import ElmoClient + +if __name__ == "__main__": + # Argument Parsing + parser = argparse.ArgumentParser(description="Poll e-Connect/IESS alarm systems.") + parser.add_argument("system", choices=["econnect", "iess"], help="Cloud platform ('econnect' or 'iess').") + parser.add_argument("domain", help="Domain identifier for the alarm system.") + parser.add_argument("username", help="Username for authentication.") + parser.add_argument("password", help="Password for authentication.") + args = parser.parse_args() + + # Determine system URL + system_url = systems.ELMO_E_CONNECT if args.system == "econnect" else systems.IESS_METRONET + + # Initialize Client + client = ElmoClient(base_url=system_url, domain=args.domain) + client.auth(args.username, args.password) + print("Authenticated") + last_ids = {} + + # Poll the system + while True: + sectors = client.query(q.SECTORS) + inputs = client.query(q.INPUTS) + outputs = client.query(q.OUTPUTS) + alerts = client.query(q.ALERTS) + panel = client.query(q.PANEL) + + last_ids[q.SECTORS] = sectors.get("last_id", 0) + last_ids[q.INPUTS] = inputs.get("last_id", 0) + last_ids[q.OUTPUTS] = outputs.get("last_id", 0) + last_ids[q.ALERTS] = alerts.get("last_id", 0) + last_ids[q.PANEL] = panel.get("last_id", 0) + + print("-" * 100) + print(f"Sectors: {sectors['last_id']}") + print(f"Inputs: {inputs['last_id']}") + print(f"Outputs: {outputs['last_id']}") + print(f"Alerts: {alerts['last_id']}") + print(f"Panel: {panel['last_id']}") + print("-" * 100) + client.poll(last_ids) diff --git a/tests/test_client.py b/tests/test_client.py index 028e9ec..2ff989e 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -348,6 +348,53 @@ def test_client_poll(server): assert "ConnectionStatus=1" in body +def test_client_auth_econnect_web_login(server): + """Web login should be used when accessing with e-Connect. + Regression test: https://github.com/palazzem/econnect-python/issues/158 + """ + 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 + 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 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) + 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 = """ diff --git a/tests/test_utils.py b/tests/test_utils.py index ef520a5..12fb7c1 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,4 +1,13 @@ -from elmo.utils import _camel_to_snake_case, _sanitize_session_id +import pytest + +from elmo.api.exceptions import ParseError +from elmo.utils import ( + _camel_to_snake_case, + _sanitize_session_id, + extract_session_id_from_html, +) + +from .fixtures.responses import STATUS_PAGE def test_sanitize_identifier(): @@ -64,3 +73,16 @@ def test_camel_to_snake_case_cache(mocker): _camel_to_snake_case("camelCase") _camel_to_snake_case("camelCase") assert mocked_sub.call_count == 4 + + +def test_extract_session_id_from_html(): + """Ensure the session ID is extracted from e-Connect status page.""" + html_content = STATUS_PAGE + assert extract_session_id_from_html(html_content) == "f8h23b4e-7a9f-4d3f-9b08-2769263ee33c" + + +def test_extract_session_id_from_html_no_session_id(): + """Ensure an error is raised when the session ID is not found in the HTML content.""" + html_content = STATUS_PAGE.replace("var sessionId = 'f8h23b4e-7a9f-4d3f-9b08-2769263ee33c';", "") + with pytest.raises(ParseError): + extract_session_id_from_html(html_content)