Add test for async_address_reachability_diagnostics API

tests/components/apprise/test_notify.py

@@ -6,11 +6,10 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
 from homeassistant.core import HomeAssistant
 from homeassistant.setup import async_setup_component
 
-BASE_COMPONENT = "notify"
-
 
 @pytest.fixture(autouse=True)
 def reset_log_level():
@@ -26,25 +25,25 @@ async def test_apprise_config_load_fail01(hass: HomeAssistant) -> None:
     """Test apprise configuration failures 1."""
 
     config = {
-        BASE_COMPONENT: {"name": "test", "platform": "apprise", "config": "/path/"}
+        NOTIFY_DOMAIN: {"name": "test", "platform": "apprise", "config": "/path/"}
     }
 
     with patch(
         "homeassistant.components.apprise.notify.apprise.AppriseConfig.add",
         return_value=False,
     ):
-        assert await async_setup_component(hass, BASE_COMPONENT, config)
+        assert await async_setup_component(hass, NOTIFY_DOMAIN, config)
         await hass.async_block_till_done()
 
         # Test that our service failed to load
-        assert not hass.services.has_service(BASE_COMPONENT, "test")
+        assert not hass.services.has_service(NOTIFY_DOMAIN, "test")
 
 
 async def test_apprise_config_load_fail02(hass: HomeAssistant) -> None:
     """Test apprise configuration failures 2."""
 
     config = {
-        BASE_COMPONENT: {"name": "test", "platform": "apprise", "config": "/path/"}
+        NOTIFY_DOMAIN: {"name": "test", "platform": "apprise", "config": "/path/"}
     }
 
     with (
@@ -57,11 +56,11 @@ async def test_apprise_config_load_fail02(hass: HomeAssistant) -> None:
             return_value=True,
         ),
     ):
-        assert await async_setup_component(hass, BASE_COMPONENT, config)
+        assert await async_setup_component(hass, NOTIFY_DOMAIN, config)
         await hass.async_block_till_done()
 
         # Test that our service failed to load
-        assert not hass.services.has_service(BASE_COMPONENT, "test")
+        assert not hass.services.has_service(NOTIFY_DOMAIN, "test")
 
 
 async def test_apprise_config_load_okay(hass: HomeAssistant, tmp_path: Path) -> None:
@@ -73,20 +72,20 @@ async def test_apprise_config_load_okay(hass: HomeAssistant, tmp_path: Path) ->
     f = d / "apprise"
     f.write_text("mailto://user:pass@example.com/")
 
-    config = {BASE_COMPONENT: {"name": "test", "platform": "apprise", "config": str(f)}}
+    config = {NOTIFY_DOMAIN: {"name": "test", "platform": "apprise", "config": str(f)}}
 
-    assert await async_setup_component(hass, BASE_COMPONENT, config)
+    assert await async_setup_component(hass, NOTIFY_DOMAIN, config)
     await hass.async_block_till_done()
 
     # Valid configuration was loaded; our service is good
-    assert hass.services.has_service(BASE_COMPONENT, "test")
+    assert hass.services.has_service(NOTIFY_DOMAIN, "test")
 
 
 async def test_apprise_url_load_fail(hass: HomeAssistant) -> None:
     """Test apprise url failure."""
 
     config = {
-        BASE_COMPONENT: {
+        NOTIFY_DOMAIN: {
             "name": "test",
             "platform": "apprise",
             "url": "mailto://user:pass@example.com",
@@ -96,18 +95,18 @@ async def test_apprise_url_load_fail(hass: HomeAssistant) -> None:
         "homeassistant.components.apprise.notify.apprise.Apprise.add",
         return_value=False,
     ):
-        assert await async_setup_component(hass, BASE_COMPONENT, config)
+        assert await async_setup_component(hass, NOTIFY_DOMAIN, config)
         await hass.async_block_till_done()
 
         # Test that our service failed to load
-        assert not hass.services.has_service(BASE_COMPONENT, "test")
+        assert not hass.services.has_service(NOTIFY_DOMAIN, "test")
 
 
 async def test_apprise_notification(hass: HomeAssistant) -> None:
     """Test apprise notification."""
 
     config = {
-        BASE_COMPONENT: {
+        NOTIFY_DOMAIN: {
             "name": "test",
             "platform": "apprise",
             "url": "mailto://user:pass@example.com",
@@ -124,18 +123,18 @@ async def test_apprise_notification(hass: HomeAssistant) -> None:
         obj.add.return_value = True
         obj.notify.return_value = True
         mock_apprise.return_value = obj
-        assert await async_setup_component(hass, BASE_COMPONENT, config)
+        assert await async_setup_component(hass, NOTIFY_DOMAIN, config)
         await hass.async_block_till_done()
 
         # Test the existence of our service
-        assert hass.services.has_service(BASE_COMPONENT, "test")
+        assert hass.services.has_service(NOTIFY_DOMAIN, "test")
 
         # Test the call to our underlining notify() call
-        await hass.services.async_call(BASE_COMPONENT, "test", data)
+        await hass.services.async_call(NOTIFY_DOMAIN, "test", data)
         await hass.async_block_till_done()
 
         # Validate calls were made under the hood correctly
-        obj.add.assert_called_once_with(config[BASE_COMPONENT]["url"])
+        obj.add.assert_called_once_with(config[NOTIFY_DOMAIN]["url"])
         obj.notify.assert_called_once_with(
             body=data["message"], title=data["title"], tag=None
         )
@@ -145,7 +144,7 @@ async def test_apprise_multiple_notification(hass: HomeAssistant) -> None:
     """Test apprise notification."""
 
     config = {
-        BASE_COMPONENT: {
+        NOTIFY_DOMAIN: {
             "name": "test",
             "platform": "apprise",
             "url": [
@@ -165,14 +164,14 @@ async def test_apprise_multiple_notification(hass: HomeAssistant) -> None:
         obj.add.return_value = True
         obj.notify.return_value = True
         mock_apprise.return_value = obj
-        assert await async_setup_component(hass, BASE_COMPONENT, config)
+        assert await async_setup_component(hass, NOTIFY_DOMAIN, config)
         await hass.async_block_till_done()
 
         # Test the existence of our service
-        assert hass.services.has_service(BASE_COMPONENT, "test")
+        assert hass.services.has_service(NOTIFY_DOMAIN, "test")
 
         # Test the call to our underlining notify() call
-        await hass.services.async_call(BASE_COMPONENT, "test", data)
+        await hass.services.async_call(NOTIFY_DOMAIN, "test", data)
         await hass.async_block_till_done()
 
         # Validate 2 calls were made under the hood
@@ -196,7 +195,7 @@ async def test_apprise_notification_with_target(
     f.write_text("devops=mailto://user:pass@example.com/\r\n")
     f.write_text("system,alert=syslog://\r\n")
 
-    config = {BASE_COMPONENT: {"name": "test", "platform": "apprise", "config": str(f)}}
+    config = {NOTIFY_DOMAIN: {"name": "test", "platform": "apprise", "config": str(f)}}
 
     # Our Message, only notify the services tagged with "devops"
     data = {"title": "Test Title", "message": "Test Message", "target": ["devops"]}
@@ -208,14 +207,14 @@ async def test_apprise_notification_with_target(
         apprise_obj.add.return_value = True
         apprise_obj.notify.return_value = True
         mock_apprise.return_value = apprise_obj
-        assert await async_setup_component(hass, BASE_COMPONENT, config)
+        assert await async_setup_component(hass, NOTIFY_DOMAIN, config)
         await hass.async_block_till_done()
 
         # Test the existence of our service
-        assert hass.services.has_service(BASE_COMPONENT, "test")
+        assert hass.services.has_service(NOTIFY_DOMAIN, "test")
 
         # Test the call to our underlining notify() call
-        await hass.services.async_call(BASE_COMPONENT, "test", data)
+        await hass.services.async_call(NOTIFY_DOMAIN, "test", data)
         await hass.async_block_till_done()
 
         # Validate calls were made under the hood correctly

tests/components/bluetooth/test_api.py

@@ -10,9 +10,11 @@ from homeassistant.components.bluetooth import (
     MONOTONIC_TIME,
     BaseHaRemoteScanner,
     BluetoothChange,
+    BluetoothReachabilityIntent,
     BluetoothScanningMode,
     BluetoothServiceInfo,
     HaBluetoothConnector,
+    async_address_reachability_diagnostics,
     async_clear_advertisement_history,
     async_request_active_scan,
     async_scanner_by_source,
@@ -112,6 +114,57 @@ async def test_async_scanner_devices_by_address_connectable(
     cancel()
 
 
+@pytest.mark.usefixtures("enable_bluetooth")
+async def test_async_address_reachability_diagnostics(hass: HomeAssistant) -> None:
+    """Test the address reachability diagnostics passthrough."""
+    # An address that was never seen reports as unknown.
+    assert "unknown" in async_address_reachability_diagnostics(
+        hass, "44:44:33:11:23:99", BluetoothReachabilityIntent.CONNECTION
+    )
+
+    manager = _get_manager()
+
+    class FakeInjectableScanner(BaseHaRemoteScanner):
+        def inject_advertisement(
+            self, device: BLEDevice, advertisement_data: AdvertisementData
+        ) -> None:
+            """Inject an advertisement."""
+            self._async_on_advertisement(
+                device.address,
+                advertisement_data.rssi,
+                device.name,
+                advertisement_data.service_uuids,
+                advertisement_data.service_data,
+                advertisement_data.manufacturer_data,
+                advertisement_data.tx_power,
+                {"scanner_specific_data": "test"},
+                MONOTONIC_TIME(),
+            )
+
+    connector = HaBluetoothConnector(MockBleakClient, "esp32", lambda: True)
+    scanner = FakeInjectableScanner("esp32", "esp32", connector, True)
+    unsetup = scanner.async_setup()
+    cancel = manager.async_register_scanner(scanner)
+    switchbot_device = generate_ble_device("44:44:33:11:23:45", "wohand", {})
+    switchbot_device_adv = generate_advertisement_data(local_name="wohand", rssi=-80)
+    scanner.inject_advertisement(switchbot_device, switchbot_device_adv)
+
+    connection_diag = async_address_reachability_diagnostics(
+        hass, "44:44:33:11:23:45", BluetoothReachabilityIntent.CONNECTION
+    )
+    assert "in connectable history" in connection_diag
+    assert "esp32" in connection_diag
+
+    # An advertisement intent does not report connectable paths or slots.
+    advertisement_diag = async_address_reachability_diagnostics(
+        hass, "44:44:33:11:23:45", BluetoothReachabilityIntent.PASSIVE_ADVERTISEMENT
+    )
+    assert "advertising" in advertisement_diag
+
+    unsetup()
+    cancel()
+
+
 @pytest.mark.usefixtures("enable_bluetooth")
 async def test_async_scanner_devices_by_address_non_connectable(
     hass: HomeAssistant,

tests/components/signal_messenger/test_notify.py

@@ -12,6 +12,7 @@ import pytest
 from requests_mock.mocker import Mocker
 import voluptuous as vol
 
+from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
 from homeassistant.core import HomeAssistant
 from homeassistant.setup import async_setup_component
 
@@ -25,13 +26,11 @@ from .conftest import (
     SignalNotificationService,
 )
 
-BASE_COMPONENT = "notify"
-
 
 async def test_signal_messenger_init(hass: HomeAssistant) -> None:
     """Test that service loads successfully."""
     config = {
-        BASE_COMPONENT: {
+        NOTIFY_DOMAIN: {
             "name": "test",
             "platform": "signal_messenger",
             "url": "http://127.0.0.1:8080",
@@ -41,10 +40,10 @@ async def test_signal_messenger_init(hass: HomeAssistant) -> None:
     }
 
     with patch("pysignalclirestapi.SignalCliRestApi.send_message", return_value=None):
-        assert await async_setup_component(hass, BASE_COMPONENT, config)
+        assert await async_setup_component(hass, NOTIFY_DOMAIN, config)
         await hass.async_block_till_done()
 
-        assert hass.services.has_service(BASE_COMPONENT, "test")
+        assert hass.services.has_service(NOTIFY_DOMAIN, "test")
 
 
 def test_send_message(