Support HomeKit Controller Thread Provisioning (#87809)

tests/components/homekit_controller/test_connection.py

@@ -2,6 +2,7 @@
 
 import dataclasses
 
+from aiohomekit.controller import TransportType
 import pytest
 
 from homeassistant.components.homekit_controller.const import (
@@ -10,7 +11,9 @@ from homeassistant.components.homekit_controller.const import (
     IDENTIFIER_LEGACY_ACCESSORY_ID,
     IDENTIFIER_LEGACY_SERIAL_NUMBER,
 )
+from homeassistant.components.thread import async_add_dataset
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import device_registry as dr
 
 from .common import setup_accessories_from_file, setup_platform, setup_test_accessories
@@ -176,3 +179,132 @@ async def test_migrate_ble_unique_id(hass: HomeAssistant) -> None:
     await hass.async_block_till_done()
 
     assert config_entry.unique_id == "02:03:ef:02:03:ef"
+
+
+async def test_thread_provision_no_creds(hass: HomeAssistant) -> None:
+    """Test that we don't migrate to thread when there are no creds available."""
+    accessories = await setup_accessories_from_file(hass, "nanoleaf_strip_nl55.json")
+
+    fake_controller = await setup_platform(hass)
+    await fake_controller.add_paired_device(accessories, "02:03:EF:02:03:EF")
+    config_entry = MockConfigEntry(
+        version=1,
+        domain="homekit_controller",
+        entry_id="TestData",
+        data={"AccessoryPairingID": "02:03:EF:02:03:EF"},
+        title="test",
+        unique_id="02:03:ef:02:03:ef",
+    )
+    config_entry.add_to_hass(hass)
+
+    fake_controller.transport_type = TransportType.BLE
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            "button",
+            "press",
+            {
+                "entity_id": "button.nanoleaf_strip_3b32_provision_preferred_thread_credentials"
+            },
+            blocking=True,
+        )
+
+
+async def test_thread_provision(hass: HomeAssistant) -> None:
+    """Test that a when a thread provision works the config entry is updated."""
+    await async_add_dataset(
+        hass,
+        "Tests",
+        "0E080000000000010000000300000F35060004001FFFE0020811111111222222220708FDAD70BF"
+        "E5AA15DD051000112233445566778899AABBCCDDEEFF030E4F70656E54687265616444656D6F01"
+        "0212340410445F2B5CA6F2A93A55CE570A70EFEECB0C0402A0F7F8",
+    )
+
+    accessories = await setup_accessories_from_file(hass, "nanoleaf_strip_nl55.json")
+
+    fake_controller = await setup_platform(hass)
+    await fake_controller.add_paired_device(accessories, "00:00:00:00:00:00")
+    config_entry = MockConfigEntry(
+        version=1,
+        domain="homekit_controller",
+        entry_id="TestData",
+        data={"AccessoryPairingID": "00:00:00:00:00:00"},
+        title="test",
+        unique_id="00:00:00:00:00:00",
+    )
+    config_entry.add_to_hass(hass)
+
+    fake_controller.transport_type = TransportType.BLE
+
+    # Needs a COAP transport to do migration
+    fake_controller.transports = {TransportType.COAP: fake_controller}
+
+    # Fake discovery won't have an address/port - set one so the migration works
+    discovery = fake_controller.discoveries["00:00:00:00:00:00"]
+    discovery.description.address = "127.0.0.1"
+    discovery.description.port = 53
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    await hass.services.async_call(
+        "button",
+        "press",
+        {
+            "entity_id": "button.nanoleaf_strip_3b32_provision_preferred_thread_credentials"
+        },
+        blocking=True,
+    )
+
+    assert config_entry.data["Connection"] == "CoAP"
+
+
+async def test_thread_provision_migration_failed(hass: HomeAssistant) -> None:
+    """Test that when a device 'migrates' but doesn't show up in CoAP, we remain in BLE mode."""
+    await async_add_dataset(
+        hass,
+        "Tests",
+        "0E080000000000010000000300000F35060004001FFFE0020811111111222222220708FDAD70BF"
+        "E5AA15DD051000112233445566778899AABBCCDDEEFF030E4F70656E54687265616444656D6F01"
+        "0212340410445F2B5CA6F2A93A55CE570A70EFEECB0C0402A0F7F8",
+    )
+
+    accessories = await setup_accessories_from_file(hass, "nanoleaf_strip_nl55.json")
+
+    fake_controller = await setup_platform(hass)
+    await fake_controller.add_paired_device(accessories, "00:00:00:00:00:00")
+    config_entry = MockConfigEntry(
+        version=1,
+        domain="homekit_controller",
+        entry_id="TestData",
+        data={"AccessoryPairingID": "00:00:00:00:00:00", "Connection": "BLE"},
+        title="test",
+        unique_id="00:00:00:00:00:00",
+    )
+    config_entry.add_to_hass(hass)
+
+    fake_controller.transport_type = TransportType.BLE
+
+    # Needs a COAP transport to do migration
+    fake_controller.transports = {TransportType.COAP: fake_controller}
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    # Make sure not disoverable via CoAP
+    del fake_controller.discoveries["00:00:00:00:00:00"]
+
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            "button",
+            "press",
+            {
+                "entity_id": "button.nanoleaf_strip_3b32_provision_preferred_thread_credentials"
+            },
+            blocking=True,
+        )
+
+    assert config_entry.data["Connection"] == "BLE"