Add WS API for getting an OTBR's extended address (#89384)

* Add WS API for getting an OTBR's extended address

* Bump python-otbr-api to 1.0.8

* Really add require_admin decorator to otbr WS API

homeassistant/components/otbr/__init__.py

@@ -78,6 +78,11 @@ class OTBRData:
         """Create an active operational dataset."""
         return await self.api.create_active_dataset(dataset)
 
+    @_handle_otbr_error
+    async def get_extended_address(self) -> bytes:
+        """Get extended address (EUI-64)."""
+        return await self.api.get_extended_address()
+
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the Open Thread Border Router component."""

homeassistant/components/otbr/manifest.json

@@ -8,5 +8,5 @@
   "documentation": "https://www.home-assistant.io/integrations/otbr",
   "integration_type": "service",
   "iot_class": "local_polling",
-  "requirements": ["python-otbr-api==1.0.5"]
+  "requirements": ["python-otbr-api==1.0.8"]
 }

homeassistant/components/otbr/websocket_api.py

@@ -18,6 +18,7 @@ def async_setup(hass: HomeAssistant) -> None:
     """Set up the OTBR Websocket API."""
     websocket_api.async_register_command(hass, websocket_info)
     websocket_api.async_register_command(hass, websocket_create_network)
+    websocket_api.async_register_command(hass, websocket_get_extended_address)
 
 
 @websocket_api.websocket_command(
@@ -96,3 +97,29 @@ async def websocket_create_network(
         return
 
     connection.send_result(msg["id"])
+
+
+@websocket_api.websocket_command(
+    {
+        "type": "otbr/get_extended_address",
+    }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+async def websocket_get_extended_address(
+    hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
+) -> None:
+    """Get extended address (EUI-64)."""
+    if DOMAIN not in hass.data:
+        connection.send_error(msg["id"], "not_loaded", "No OTBR API loaded")
+        return
+
+    data: OTBRData = hass.data[DOMAIN]
+
+    try:
+        extended_address = await data.get_extended_address()
+    except HomeAssistantError as exc:
+        connection.send_error(msg["id"], "get_extended_address_failed", str(exc))
+        return
+
+    connection.send_result(msg["id"], {"extended_address": extended_address.hex()})

homeassistant/components/thread/manifest.json

@@ -7,6 +7,6 @@
   "documentation": "https://www.home-assistant.io/integrations/thread",
   "integration_type": "service",
   "iot_class": "local_polling",
-  "requirements": ["python-otbr-api==1.0.5", "pyroute2==0.7.5"],
+  "requirements": ["python-otbr-api==1.0.8", "pyroute2==0.7.5"],
   "zeroconf": ["_meshcop._udp.local."]
 }

requirements_all.txt

@@ -2097,7 +2097,7 @@ python-nest==4.2.0
 
 # homeassistant.components.otbr
 # homeassistant.components.thread
-python-otbr-api==1.0.5
+python-otbr-api==1.0.8
 
 # homeassistant.components.picnic
 python-picnic-api==1.1.0

requirements_test_all.txt

@@ -1496,7 +1496,7 @@ python-nest==4.2.0
 
 # homeassistant.components.otbr
 # homeassistant.components.thread
-python-otbr-api==1.0.5
+python-otbr-api==1.0.8
 
 # homeassistant.components.picnic
 python-picnic-api==1.1.0

tests/components/otbr/test_websocket_api.py

@@ -234,3 +234,75 @@ async def test_get_info_fetch_fails_3(
     assert msg["id"] == 5
     assert not msg["success"]
     assert msg["error"]["code"] == "set_enabled_failed"
+
+
+async def test_get_extended_address(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test get extended address."""
+
+    with patch(
+        "python_otbr_api.OTBR.get_extended_address",
+        return_value=bytes.fromhex("4EF6C4F3FF750626"),
+    ):
+        await websocket_client.send_json(
+            {
+                "id": 5,
+                "type": "otbr/get_extended_address",
+            }
+        )
+        msg = await websocket_client.receive_json()
+
+    assert msg["id"] == 5
+    assert msg["success"]
+    assert msg["result"] == {"extended_address": "4EF6C4F3FF750626".lower()}
+
+
+async def test_get_extended_address_no_entry(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test get extended address."""
+    await async_setup_component(hass, "otbr", {})
+    websocket_client = await hass_ws_client(hass)
+    await websocket_client.send_json(
+        {
+            "id": 5,
+            "type": "otbr/get_extended_address",
+        }
+    )
+
+    msg = await websocket_client.receive_json()
+    assert msg["id"] == 5
+    assert not msg["success"]
+    assert msg["error"]["code"] == "not_loaded"
+
+
+async def test_get_extended_address_fetch_fails(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    otbr_config_entry,
+    websocket_client,
+) -> None:
+    """Test get extended address."""
+    await async_setup_component(hass, "otbr", {})
+
+    with patch(
+        "python_otbr_api.OTBR.get_extended_address",
+        side_effect=python_otbr_api.OTBRError,
+    ):
+        await websocket_client.send_json(
+            {
+                "id": 5,
+                "type": "otbr/get_extended_address",
+            }
+        )
+        msg = await websocket_client.receive_json()
+
+    assert msg["id"] == 5
+    assert not msg["success"]
+    assert msg["error"]["code"] == "get_extended_address_failed"