Fixes for 3DCM020200r015 firmware

src/sfrbox_api/bridge.py

@@ -1,4 +1,5 @@
 """SFR Box bridge."""
+
 from __future__ import annotations
 
 import logging
@@ -32,6 +33,7 @@
 
 
 _LOGGER = logging.getLogger(__name__)
+_T = TypeVar("_T")
 _R = TypeVar("_R")
 _P = ParamSpec("_P")
 
@@ -87,13 +89,15 @@ async def _get_token(self) -> str:
         if not (self._username and self._password):
             raise SFRBoxAuthenticationError("Credentials not set")
         element = await self._send_get("auth", "getToken")
+        assert element is not None  # noqa: S101
         if (method := element.get("method")) not in {"all", "passwd"}:
             raise SFRBoxAuthenticationError(
                 f"Password authentication is not allowed, valid methods: `{method}`"
             )
         token = element.get("token", "")
         hash = compute_hash(token, self._username, self._password)
         element = await self._send_get("auth", "checkToken", token=token, hash=hash)
+        assert element is not None  # noqa: S101
         return element.get("token", "")
 
     def _check_response(self, response: httpx.Response) -> XmlElement:
@@ -107,10 +111,15 @@ def _check_response(self, response: httpx.Response) -> XmlElement:
             response.text,
         )
         response.raise_for_status()
+        response_text: str = response.text
+        if "</firewall>" in response_text and "<dsl" in response_text:
+            # There is a bug in firmware 3DCM020200r015
+            response_text = response_text.replace("</firewall>", "/>")
+
         try:
-            element: XmlElement = DefusedElementTree.fromstring(response.text)
+            element: XmlElement = DefusedElementTree.fromstring(response_text)
         except Exception as exc:
-            raise SFRBoxError(f"Failed to parse response: {response.text}") from exc
+            raise SFRBoxError(f"Failed to parse response: {response_text}") from exc
         stat = element.get("stat", "")
         if (
             stat == "fail"
@@ -124,7 +133,7 @@ def _check_response(self, response: httpx.Response) -> XmlElement:
                 raise SFRBoxAuthenticationError(f"Api call failed: [{code}] {msg}")
             raise SFRBoxApiError(f"Api call failed: [{code}] {msg}")
         if stat != "ok":
-            raise SFRBoxError(f"Response was not ok: {response.text}")
+            raise SFRBoxError(f"Response was not ok: {response_text}")
         return element
 
     @_with_error_wrapping
@@ -137,10 +146,14 @@ async def _send_get_simple(
         return element
 
     @_with_error_wrapping
-    async def _send_get(self, namespace: str, method: str, **kwargs: str) -> XmlElement:
+    async def _send_get(
+        self, namespace: str, method: str, **kwargs: str
+    ) -> XmlElement | None:
         params = httpx.QueryParams(method=f"{namespace}.{method}", **kwargs)
         response = await self._client.get(f"http://{self._ip}/api/1.0/", params=params)
         element = self._check_response(response)
+        if len(element) == 0:
+            return None
         result = element.find(namespace)
         if result is None:
             raise SFRBoxError(
@@ -163,30 +176,38 @@ async def _send_post(
         )
         self._check_response(response)
 
-    async def dsl_get_info(self) -> DslInfo:
+    def _create_class(
+        self, cls: type[_T], xml_response: XmlElement | None
+    ) -> _T | None:
+        """Crée la classe."""
+        if xml_response is None:
+            return None
+        return cls(**xml_response.attrib)
+
+    async def dsl_get_info(self) -> DslInfo | None:
         """Renvoie les informations sur le lien ADSL."""
         xml_response = await self._send_get("dsl", "getInfo")
-        return DslInfo(**xml_response.attrib)  # type: ignore[arg-type]
+        return self._create_class(DslInfo, xml_response)
 
-    async def ftth_get_info(self) -> FtthInfo:
+    async def ftth_get_info(self) -> FtthInfo | None:
         """Renvoie les informations sur le lien FTTH."""
         xml_response = await self._send_get("ftth", "getInfo")
-        return FtthInfo(**xml_response.attrib)
+        return self._create_class(FtthInfo, xml_response)
 
-    async def system_get_info(self) -> SystemInfo:
+    async def system_get_info(self) -> SystemInfo | None:
         """Renvoie les informations sur le système."""
         xml_response = await self._send_get("system", "getInfo")
-        return SystemInfo(**xml_response.attrib)  # type: ignore[arg-type]
+        return self._create_class(SystemInfo, xml_response)
 
     async def system_reboot(self) -> None:
         """Redémarrer la BOX."""
         token = await self._ensure_token()
         await self._send_post("system", "reboot", token=token)
 
-    async def wan_get_info(self) -> WanInfo:
+    async def wan_get_info(self) -> WanInfo | None:
         """Renvoie les informations génériques sur la connexion internet."""
         xml_response = await self._send_get("wan", "getInfo")
-        return WanInfo(**xml_response.attrib)  # type: ignore[arg-type]
+        return self._create_class(WanInfo, xml_response)
 
     async def wlan_get_client_list(self) -> WlanClientList:
         """Liste des clients WiFi."""
@@ -201,6 +222,7 @@ async def wlan_get_info(self) -> WlanInfo:
         """Renvoie les informations sur le WiFi."""
         token = await self._ensure_token()
         xml_response = await self._send_get("wlan", "getInfo", token=token)
+        assert xml_response is not None  # noqa: S101
         wl0_element = xml_response.find("wl0")
         assert wl0_element is not None  # noqa: S101
         wl0 = WlanWl0Info(**wl0_element.attrib)

src/sfrbox_api/models.py

@@ -1,4 +1,5 @@
 """SFR Box models."""
+
 from typing import Optional
 from typing import Union
 
@@ -11,7 +12,9 @@
     from pydantic.dataclasses import dataclass  # type: ignore[no-redef]
 
 
-def _empty_to_none(v: Union[int, str, None]) -> Optional[Union[int, str, None]]:
+def _empty_to_none(
+    v: Union[float, int, str, None]
+) -> Optional[Union[float, int, str, None]]:
     return None if v == "" else v
 
 
@@ -53,18 +56,20 @@ class DslInfo:
     """Débit flux descendant."""
     rate_up: int
     """Débit flux montant."""
-    line_status: str
+    line_status: Optional[str] = None
     """Etat détaillé du lien.
 
     = (No Defect|Of Frame|Loss Of Signal|Loss Of Power|Loss Of Signal Quality|Unknown)
-    (firmware >= 3.3.2)"""
-    training: str
+    (firmware >= 3.3.2)
+    Note: ne semble pas être disponible dans la box 8"""
+    training: Optional[str] = None
     """Etat de négociation avec le DSLAM.
 
     = (Idle|G.994 Training|G.992 Started|G.922 Channel Analysis|G.992 Message Exchange|
     G.993 Started|G.993 Channel Analysis|G.993 Message Exchange|Showtime|Unknown)
 
-    (firmware >= 3.3.2)"""
+    (firmware >= 3.3.2)
+    Note: ne semble pas être disponible dans la box 8"""
 
 
 @dataclass
@@ -130,10 +135,12 @@ class SystemInfo:
     """Tension de l'alimentation exprimé en mV.
 
     (firmware >= 3.5.0)"""
-    temperature: Optional[int] = None
+    temperature: Optional[Union[float, int]] = None
     """Température de la BOX exprimé en m°C.
 
-    (firmware >= 3.5.0)"""
+    (firmware >= 3.5.0)
+    Note: il semblerait que la température de la BOX soit
+    exprimée en °C pour les box 8"""
     serial_number: Optional[str] = None
     """Numéro de série de l'IAD.
 

tests/fixtures/dsl.getInfo.3DCM020200r015.xml

@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<rsp stat="ok" version="1.0">
+  <dsl linemode="G.DMT" uptime="4857" counter="1" crc="0" status="up" noise_down="4.5" noise_up="4.2" attenuation_down="3.2" attenuation_up="5.2" rate_down="8000" rate_up="800"
+  </firewall>
+</rsp>

tests/fixtures/ftth.getInfo.3DCM020200r015.xml

@@ -0,0 +1,3 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<rsp stat="ok" version="1.0">
+</rsp>

tests/fixtures/system.getInfo.3DCM020200r015.xml

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<rsp stat="ok" version="1.0">
+  <system product_id="ALGD1-UBE-r0" serial_number="MU1B01140006020043" mac_addr="***hidden***" net_mode="router" net_infra="fttb" uptime="1563441" version_mainfirmware="3DCM020200r015" version_rescuefirmware="3DCM020200r015" version_bootloader="3.00" version_dsldriver="" current_datetime="20240905130854" refclient="" idur="RNCUAOL" alimvoltage="12251" temperature="57.5" />
+</rsp>

tests/test_bridge.py

@@ -1,4 +1,5 @@
 """Test cases for the __main__ module."""
+
 import pathlib
 import re
 import time
@@ -104,6 +105,33 @@ async def test_authenticate_method_not_allowed() -> None:
             await box.authenticate(password="password")  # noqa: S106
 
 
+@respx.mock
+@pytest.mark.asyncio
+async def test_dsl_getinfo_3dcm020200r015() -> None:
+    """It exits with a status code of zero."""
+    respx.get("http://192.168.0.1/api/1.0/?method=dsl.getInfo").respond(
+        text=_load_fixture("dsl.getInfo.3DCM020200r015.xml")
+    )
+    async with httpx.AsyncClient() as client:
+        box = SFRBox(ip="192.168.0.1", client=client)
+        info = await box.dsl_get_info()
+        assert info == DslInfo(
+            linemode="G.DMT",
+            uptime=4857,
+            counter=1,
+            crc=0,
+            status="up",
+            noise_down=4.5,
+            noise_up=4.2,
+            attenuation_down=3.2,
+            attenuation_up=5.2,
+            rate_down=8000,
+            rate_up=800,
+            line_status=None,
+            training=None,
+        )
+
+
 @respx.mock
 @pytest.mark.asyncio
 async def test_dsl_getinfo() -> None:
@@ -144,6 +172,19 @@ async def test_ftth_getinfo() -> None:
         assert info == FtthInfo(status="down", wanfibre="out")
 
 
+@respx.mock
+@pytest.mark.asyncio
+async def test_ftth_getinfo_3dcm020200r015() -> None:
+    """It exits with a status code of zero."""
+    respx.get("http://192.168.0.1/api/1.0/?method=ftth.getInfo").respond(
+        text=_load_fixture("ftth.getInfo.3DCM020200r015.xml")
+    )
+    async with httpx.AsyncClient() as client:
+        box = SFRBox(ip="192.168.0.1", client=client)
+        info = await box.ftth_get_info()
+        assert info is None
+
+
 @respx.mock
 @pytest.mark.asyncio
 async def test_system_getinfo() -> None:
@@ -202,6 +243,35 @@ async def test_system_getinfo_3_5_8() -> None:
         )
 
 
+@respx.mock
+@pytest.mark.asyncio
+async def test_system_getinfo_3dcm020200r015() -> None:
+    """It exits with a status code of zero."""
+    respx.get("http://192.168.0.1/api/1.0/?method=system.getInfo").respond(
+        text=_load_fixture("system.getInfo.3DCM020200r015.xml")
+    )
+    async with httpx.AsyncClient() as client:
+        box = SFRBox(ip="192.168.0.1", client=client)
+        info = await box.system_get_info()
+        assert info == SystemInfo(
+            product_id="ALGD1-UBE-r0",
+            mac_addr="***hidden***",
+            net_mode="router",
+            net_infra="fttb",
+            uptime=1563441,
+            version_mainfirmware="3DCM020200r015",
+            version_rescuefirmware="3DCM020200r015",
+            version_bootloader="3.00",
+            version_dsldriver="",
+            current_datetime="20240905130854",
+            refclient="",
+            idur="RNCUAOL",
+            alimvoltage=12251,
+            temperature=57.5,
+            serial_number="MU1B01140006020043",
+        )
+
+
 @respx.mock
 @pytest.mark.asyncio
 async def test_system_reboot() -> None: