diff --git a/src/sfrbox_api/bridge.py b/src/sfrbox_api/bridge.py index 563d9f7..2f52b21 100644 --- a/src/sfrbox_api/bridge.py +++ b/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,6 +89,7 @@ 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}`" @@ -94,6 +97,7 @@ async def _get_token(self) -> str: 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 "" in response_text and "", "/>") + 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) diff --git a/src/sfrbox_api/models.py b/src/sfrbox_api/models.py index 342babf..d07a2d1 100644 --- a/src/sfrbox_api/models.py +++ b/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. diff --git a/tests/fixtures/dsl.getInfo.3DCM020200r015.xml b/tests/fixtures/dsl.getInfo.3DCM020200r015.xml new file mode 100644 index 0000000..e62d39f --- /dev/null +++ b/tests/fixtures/dsl.getInfo.3DCM020200r015.xml @@ -0,0 +1,5 @@ + + + + diff --git a/tests/fixtures/ftth.getInfo.3DCM020200r015.xml b/tests/fixtures/ftth.getInfo.3DCM020200r015.xml new file mode 100644 index 0000000..f45403e --- /dev/null +++ b/tests/fixtures/ftth.getInfo.3DCM020200r015.xml @@ -0,0 +1,3 @@ + + + diff --git a/tests/fixtures/system.getInfo.3DCM020200r015.xml b/tests/fixtures/system.getInfo.3DCM020200r015.xml new file mode 100644 index 0000000..7c426fd --- /dev/null +++ b/tests/fixtures/system.getInfo.3DCM020200r015.xml @@ -0,0 +1,4 @@ + + + + diff --git a/tests/test_bridge.py b/tests/test_bridge.py index 321a1ff..010a6a9 100644 --- a/tests/test_bridge.py +++ b/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: