refactor!: add query.ALERTS to query the system (#118)

#118

src/elmo/api/client.py

@@ -488,46 +488,6 @@ def _get_descriptions(self):
         _LOGGER.debug(f"Client | Descriptions retrieved (in-cache): {descriptions}")
         return descriptions
 
-    @require_session
-    def get_status(self):
-        """Retrieve the status and convert its keys to snake_case format.
-
-        This method sends a POST request to retrieve the status and then converts
-        the keys of the 'PanelAnomalies' section of the response to snake_case format
-        for easier usage in other modules.
-
-        Args:
-            None
-
-        Returns:
-            dict: A dictionary containing the status with keys in snake_case format.
-
-        Raises:
-            requests.HTTPError: If the POST request returns a bad response.
-            ParseError: If the response doesn't have the expected format or missing 'PanelAnomalies'.
-        """
-        payload = {"sessionId": self._session_id}
-        response = self._session.post(self._router.status, data=payload)
-        _LOGGER.debug(f"Client | Status response: {response}")
-        response.raise_for_status()
-
-        try:
-            # Check if the response has the expected format
-            msg = response.json()
-            status = msg["PanelLeds"]
-            anomalies = msg["PanelAnomalies"]
-        except (KeyError, ValueError):
-            raise ParseError("Unexpected response format from the server.")
-
-        # Merge the 'status' and 'anomalies' dictionaries
-        merged_dict = {**status, **anomalies}
-
-        # Convert the dict to a snake_case one to simplify the usage in other modules
-        snake_case_dict = {_camel_to_snake_case(k): v for k, v in merged_dict.items()}
-        _LOGGER.debug(f"Client | Status retrieved: {snake_case_dict}")
-
-        return snake_case_dict
-
     @require_session
     def query(self, query):
         """Query an Elmo System to retrieve registered entries. It's possible to query
@@ -580,48 +540,70 @@ def query(self, query):
             key_group = "inputs"
             endpoint = self._router.inputs
             _LOGGER.debug("Client | Querying inputs")
+        elif query == q.ALERTS:
+            endpoint = self._router.status
+            _LOGGER.debug("Client | Querying alerts")
         else:
             # Bail-out if the query is not recognized
             raise QueryNotValid()
 
         response = self._session.post(endpoint, data={"sessionId": self._session_id})
         response.raise_for_status()
 
-        # Retrieve description or use the cache
-        descriptions = self._get_descriptions()
-
-        # Filter only entries that are used
-        # `excluded` field is available only on inputs, but to return the same `dict`
-        # structure, we default "excluded" as False for sectors. In fact, sectors
-        # are never excluded.
-        entries = response.json()
-        _LOGGER.debug(f"Client | Query response: {entries}")
-        items = {}
-        result = {
-            "last_id": entries[-1]["Id"],
-            key_group: items,
-        }
-        try:
-            for entry in entries:
-                if entry["InUse"]:
-                    # Address potential data inconsistency between cloud data and main unit.
-                    # In some installations, they may be out of sync, resulting in the cloud
-                    # providing a sector/input that doesn't actually exist in the main unit.
-                    # To handle this, we default the name to "Unknown" if its description
-                    # isn't found in the cloud data to prevent KeyError.
-                    name = descriptions[query].get(entry["Index"], "Unknown")
-                    item = {
-                        "id": entry.get("Id"),
-                        "index": entry.get("Index"),
-                        "element": entry.get("Element"),
-                        "excluded": entry.get("Excluded", False),
-                        "status": entry.get(status, False),
-                        "name": name,
-                    }
-
-                    items[entry.get("Index")] = item
-        except KeyError as err:
-            raise ParseError(f"Client | Unable to parse query response: {err}") from err
-
-        _LOGGER.debug(f"Client | Query parsed successfully: {result}")
-        return result
+        if query in [q.SECTORS, q.INPUTS]:
+            # Retrieve description or use the cache
+            descriptions = self._get_descriptions()
+
+            # Filter only entries that are used
+            # `excluded` field is available only on inputs, but to return the same `dict`
+            # structure, we default "excluded" as False for sectors. In fact, sectors
+            # are never excluded.
+            entries = response.json()
+            _LOGGER.debug(f"Client | Query response: {entries}")
+            items = {}
+            result = {
+                "last_id": entries[-1]["Id"],
+                key_group: items,
+            }
+            try:
+                for entry in entries:
+                    if entry["InUse"]:
+                        # Address potential data inconsistency between cloud data and main unit.
+                        # In some installations, they may be out of sync, resulting in the cloud
+                        # providing a sector/input that doesn't actually exist in the main unit.
+                        # To handle this, we default the name to "Unknown" if its description
+                        # isn't found in the cloud data to prevent KeyError.
+                        name = descriptions[query].get(entry["Index"], "Unknown")
+                        item = {
+                            "id": entry.get("Id"),
+                            "index": entry.get("Index"),
+                            "element": entry.get("Element"),
+                            "excluded": entry.get("Excluded", False),
+                            "status": entry.get(status, False),
+                            "name": name,
+                        }
+
+                        items[entry.get("Index")] = item
+            except KeyError as err:
+                raise ParseError(f"Client | Unable to parse query response: {err}") from err
+
+            _LOGGER.debug(f"Client | Query parsed successfully: {result}")
+            return result
+
+        if query == q.ALERTS:
+            try:
+                # Check if the response has the expected format
+                msg = response.json()
+                status = msg["PanelLeds"]
+                anomalies = msg["PanelAnomalies"]
+            except (KeyError, ValueError):
+                raise ParseError("Unexpected response format from the server.")
+
+            # Merge the 'status' and 'anomalies' dictionaries
+            merged_dict = {**status, **anomalies}
+
+            # Convert the dict to a snake_case one to simplify the usage in other modules
+            snake_case_dict = {_camel_to_snake_case(k): v for k, v in merged_dict.items()}
+            _LOGGER.debug(f"Client | Status retrieved: {snake_case_dict}")
+
+            return snake_case_dict

src/elmo/query.py

@@ -1,2 +1,3 @@
 SECTORS = 9
 INPUTS = 10
+ALERTS = 11

tests/test_client.py

@@ -1603,8 +1603,8 @@ def test_client_query_error(server, mocker):
         client.query(query.SECTORS)
 
 
-def test_client_get_status(server):
-    """Should raise HTTPError if there is a client error."""
+def test_client_get_alerts_status(server):
+    """Should query a Elmo system to retrieve alerts status."""
     html = """
         {
             "StatusUid": 1,
@@ -1645,14 +1645,16 @@ def test_client_get_status(server):
             }
         }
     """
+
     server.add(responses.POST, "https://example.com/api/statusadv", body=html, status=200)
     client = ElmoClient(base_url="https://example.com", domain="domain")
     client._session_id = "test"
     # Test
-    result = client.get_status()
+    alerts = client.query(query.ALERTS)
     body = server.calls[0].request.body
     assert body == "sessionId=test"
-    assert result == {
+    # Expected output
+    assert alerts == {
         "inputs_led": 2,
         "anomalies_led": 1,
         "alarm_led": 0,
@@ -1681,29 +1683,29 @@ def test_client_get_status(server):
     }
 
 
-def test_client_get_status_http_error(server):
+def test_client_get_alerts_http_error(server):
     """Should raise HTTPError if there is a client error."""
     server.add(responses.POST, "https://example.com/api/statusadv", body="500 Error", status=500)
     client = ElmoClient(base_url="https://example.com", domain="domain")
     client._session_id = "test"
     # Test
     with pytest.raises(HTTPError):
-        client.get_status()
+        client.query(query.ALERTS)
     assert len(server.calls) == 1
 
 
-def test_client_get_status_invalid_json(server):
+def test_client_get_alerts_invalid_json(server):
     """Should raise ParseError if the response is unexpected."""
     server.add(responses.POST, "https://example.com/api/statusadv", body="Invalid JSON", status=200)
     client = ElmoClient(base_url="https://example.com", domain="domain")
     client._session_id = "test"
     # Test
     with pytest.raises(ParseError):
-        client.get_status()
+        client.query(query.ALERTS)
     assert len(server.calls) == 1
 
 
-def test_client_get_status_missing_data(server):
+def test_client_get_alerts_missing_data(server):
     """Should raise ParseError if some response data is missing."""
     html = """
         {
@@ -1727,5 +1729,5 @@ def test_client_get_status_missing_data(server):
     client._session_id = "test"
     # Test
     with pytest.raises(ParseError):
-        client.get_status()
+        client.query(query.ALERTS)
     assert len(server.calls) == 1