Add REST reboot support and hosts endpoint fallback

Author: espipj

README.md

@@ -135,7 +135,7 @@ The client supports two API variants:
 - `ApiMode.REST`: newer `/api/v1/*` API used by newer firmwares
 - `ApiMode.AUTO` (default): tries legacy first, then falls back to REST when the legacy endpoint is unavailable
 
-When REST mode is active, high-level helpers like `get_device_info()` and `get_hosts()` are supported. XPath-based methods (`get_value_by_xpath`, `set_value_by_xpath`, `get_values_by_xpaths`) are legacy-only.
+When REST mode is active, high-level helpers like `get_device_info()`, `get_hosts()` and `reboot()` are supported. XPath-based methods (`get_value_by_xpath`, `set_value_by_xpath`, `get_values_by_xpaths`) are legacy-only.
 
 ### Determine the EncryptionMethod
 

sagemcom_api/client.py

@@ -415,6 +415,116 @@ async def __rest_login(self):
         )
         return True
 
+    @staticmethod
+    def __first_value(data: dict[str, Any], *keys: str) -> Any:
+        """Return the first non-None value from data for the given keys."""
+        for key in keys:
+            if key in data and data[key] is not None:
+                return data[key]
+        return None
+
+    @staticmethod
+    def __to_bool(value: Any, default: bool = True) -> bool:
+        """Convert mixed payload boolean values to bool."""
+        if value is None:
+            return default
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, (int, float)):
+            return bool(value)
+        if isinstance(value, str):
+            return value.strip().lower() in ("1", "true", "yes", "on", "up")
+        return default
+
+    def __build_rest_device(
+        self, entry: dict[str, Any], interface_type: str | None
+    ) -> Device:
+        """Map a REST host entry to Device."""
+        detected_interface = self.__first_value(
+            entry,
+            "interface_type",
+            "interfaceType",
+            "interface",
+            "connectionType",
+            "connection_type",
+            "type",
+        )
+        if interface_type is None and isinstance(detected_interface, str):
+            normalized = detected_interface.lower()
+            if "wifi" in normalized or "wireless" in normalized or "wlan" in normalized:
+                interface_type = "wifi"
+            elif "ethernet" in normalized or "eth" in normalized or "lan" in normalized:
+                interface_type = "ethernet"
+            else:
+                interface_type = detected_interface
+
+        return Device(
+            uid=self.__first_value(entry, "id", "uid"),
+            phys_address=self.__first_value(
+                entry, "macAddress", "mac_address", "phys_address"
+            ),
+            ip_address=self.__first_value(entry, "ipAddress", "ip_address"),
+            host_name=self.__first_value(entry, "hostname", "host_name", "name"),
+            user_host_name=self.__first_value(
+                entry, "friendlyname", "friendly_name", "user_host_name"
+            ),
+            active=self.__to_bool(
+                self.__first_value(entry, "active", "isActive"), True
+            ),
+            interface_type=interface_type,
+            detected_device_type=self.__first_value(
+                entry, "devicetype", "deviceType", "detected_device_type"
+            ),
+        )
+
+    def __extract_rest_home_hosts(self, data: Any) -> list[Device]:
+        """Parse /api/v1/home hosts payload."""
+        if isinstance(data, list):
+            if not data:
+                return []
+            home = data[0]
+        elif isinstance(data, dict):
+            home = data
+        else:
+            raise UnknownException("Invalid response from /api/v1/home")
+
+        if not isinstance(home, dict):
+            raise UnknownException("Invalid response from /api/v1/home")
+
+        devices: list[Device] = []
+        for entry in home.get("wirelessListDevice", []):
+            if isinstance(entry, dict):
+                devices.append(self.__build_rest_device(entry, "wifi"))
+
+        for entry in home.get("ethernetListDevice", []):
+            if isinstance(entry, dict):
+                devices.append(self.__build_rest_device(entry, "ethernet"))
+
+        return devices
+
+    def __extract_rest_hosts(self, data: Any) -> list[Device]:
+        """Parse /api/v1/hosts payload."""
+        hosts: list[dict[str, Any]]
+        if isinstance(data, list):
+            hosts = [entry for entry in data if isinstance(entry, dict)]
+        elif isinstance(data, dict):
+            raw_hosts = self.__first_value(
+                data,
+                "hosts",
+                "Hosts",
+                "list",
+                "listDevice",
+                "list_device",
+                "devices",
+            )
+            if not isinstance(raw_hosts, list):
+                raise UnknownException("Invalid response from /api/v1/hosts")
+            hosts = [entry for entry in raw_hosts if isinstance(entry, dict)]
+        else:
+            raise UnknownException("Invalid response from /api/v1/hosts")
+
+        return [self.__build_rest_device(entry, None) for entry in hosts]
+
     def __should_fallback_to_rest(self, exception: Exception) -> bool:
         """Return True when legacy API failure indicates a REST-only router."""
         if isinstance(exception, UnsupportedHostException):
@@ -663,40 +773,23 @@ async def get_device_info(self) -> DeviceInfo:
     async def get_hosts(self, only_active: bool | None = False) -> list[Device]:
         """Retrieve hosts connected to Sagemcom F@st device."""
         if self._active_api_mode == ApiMode.REST:
-            data = await self.__rest_request("GET", "/api/v1/home")
-            if not data or not isinstance(data, list):
-                raise UnknownException("Invalid response from /api/v1/home")
-
-            home = data[0]
+            rest_errors: list[Exception] = []
             devices: list[Device] = []
 
-            for entry in home.get("wirelessListDevice", []):
-                devices.append(
-                    Device(
-                        uid=entry.get("id"),
-                        phys_address=entry.get("macAddress"),
-                        ip_address=entry.get("ipAddress"),
-                        host_name=entry.get("hostname"),
-                        user_host_name=entry.get("friendlyname"),
-                        active=entry.get("active", True),
-                        interface_type="wifi",
-                        detected_device_type=entry.get("devicetype"),
-                    )
-                )
-
-            for entry in home.get("ethernetListDevice", []):
-                devices.append(
-                    Device(
-                        uid=entry.get("id"),
-                        phys_address=entry.get("macAddress"),
-                        ip_address=entry.get("ipAddress"),
-                        host_name=entry.get("hostname"),
-                        user_host_name=entry.get("friendlyname"),
-                        active=entry.get("active", True),
-                        interface_type="ethernet",
-                        detected_device_type=entry.get("devicetype"),
-                    )
-                )
+            for endpoint, parser in (
+                ("/api/v1/home", self.__extract_rest_home_hosts),
+                ("/api/v1/hosts", self.__extract_rest_hosts),
+            ):
+                try:
+                    data = await self.__rest_request("GET", endpoint)
+                    devices = parser(data)
+                    break
+                except (UnknownException, UnsupportedHostException) as exception:
+                    rest_errors.append(exception)
+            else:
+                if rest_errors:
+                    raise rest_errors[-1]
+                raise UnknownException("Unable to retrieve hosts using REST endpoints")
 
             if only_active:
                 return [d for d in devices if d.active is True]
@@ -746,7 +839,9 @@ async def get_port_mappings(self) -> list[PortMapping]:
     )
     async def reboot(self):
         """Reboot Sagemcom F@st device."""
-        self.__ensure_legacy_api()
+        if self._active_api_mode == ApiMode.REST:
+            return await self.__rest_request("POST", "/api/v1/device/reboot")
+
         action = {
             "id": 0,
             "method": "reboot",

tests/unit/test_client_basic.py

@@ -234,3 +234,96 @@ async def test_get_hosts_rest_mode():
     assert devices[0].interface_type == "wifi"
     assert devices[1].host_name == "lan-device"
     assert devices[1].interface_type == "ethernet"
+
+
+@pytest.mark.asyncio
+async def test_get_hosts_rest_fallbacks_to_hosts_endpoint():
+    """/api/v1/hosts should be used when /api/v1/home response is invalid."""
+    mock_session = MagicMock(spec=ClientSession)
+    mock_session.close = AsyncMock()
+
+    login_response = AsyncMock()
+    login_response.status = 204
+    login_response.text = AsyncMock(return_value="")
+    login_response.__aenter__ = AsyncMock(return_value=login_response)
+    login_response.__aexit__ = AsyncMock(return_value=None)
+
+    home_response = AsyncMock()
+    home_response.status = 200
+    home_response.json = AsyncMock(return_value=[{"unexpected": "shape"}])
+    home_response.__aenter__ = AsyncMock(return_value=home_response)
+    home_response.__aexit__ = AsyncMock(return_value=None)
+
+    hosts_payload = [
+        {
+            "id": 7,
+            "hostname": "tablet",
+            "friendlyname": "tablet",
+            "macAddress": "aa:aa:aa:aa:aa:aa",
+            "ipAddress": "192.168.1.50",
+            "active": "true",
+            "interfaceType": "wireless",
+            "devicetype": "TABLET",
+        }
+    ]
+    hosts_response = AsyncMock()
+    hosts_response.status = 200
+    hosts_response.json = AsyncMock(return_value=hosts_payload)
+    hosts_response.__aenter__ = AsyncMock(return_value=hosts_response)
+    hosts_response.__aexit__ = AsyncMock(return_value=None)
+
+    mock_session.request.side_effect = [login_response, home_response, hosts_response]
+
+    client = SagemcomClient(
+        host="192.168.1.1",
+        username="admin",
+        password="admin",
+        session=mock_session,
+        api_mode=ApiMode.REST,
+    )
+
+    await client.login()
+    devices = await client.get_hosts(only_active=True)
+
+    assert len(devices) == 1
+    assert devices[0].host_name == "tablet"
+    assert devices[0].interface_type == "wifi"
+    assert devices[0].active is True
+
+
+@pytest.mark.asyncio
+async def test_reboot_rest_mode():
+    """reboot should call REST endpoint on REST firmware."""
+    mock_session = MagicMock(spec=ClientSession)
+    mock_session.close = AsyncMock()
+
+    login_response = AsyncMock()
+    login_response.status = 204
+    login_response.text = AsyncMock(return_value="")
+    login_response.__aenter__ = AsyncMock(return_value=login_response)
+    login_response.__aexit__ = AsyncMock(return_value=None)
+
+    reboot_response = AsyncMock()
+    reboot_response.status = 204
+    reboot_response.text = AsyncMock(return_value="")
+    reboot_response.__aenter__ = AsyncMock(return_value=reboot_response)
+    reboot_response.__aexit__ = AsyncMock(return_value=None)
+
+    mock_session.request.side_effect = [login_response, reboot_response]
+
+    client = SagemcomClient(
+        host="192.168.1.1",
+        username="admin",
+        password="admin",
+        session=mock_session,
+        api_mode=ApiMode.REST,
+    )
+
+    await client.login()
+    result = await client.reboot()
+
+    assert result is None
+    assert mock_session.request.call_count == 2
+    reboot_call = mock_session.request.call_args_list[1]
+    assert reboot_call.args[0] == "POST"
+    assert reboot_call.args[1].endswith("/api/v1/device/reboot")