feat: callback on scanner start success (#298)

Author: bdraco

src/habluetooth/base_scanner.py

@@ -123,6 +123,15 @@ def __init__(
         self._connect_failures: dict[str, int] = {}
         self._connect_in_progress: dict[str, int] = {}
 
+    def _on_start_success(self) -> None:
+        """
+        Called when the scanner successfully starts.
+
+        Notifies the manager that this scanner has started.
+        """
+        if self._manager:
+            self._manager.on_scanner_start(self)
+
     def _clear_connection_history(self) -> None:
         """Clear the connection history for a scanner."""
         self._connect_failures.clear()

src/habluetooth/manager.py

@@ -215,6 +215,22 @@ def supports_passive_scan(self) -> bool:
         """Return if passive scan is supported."""
         return any(adapter[ADAPTER_PASSIVE_SCAN] for adapter in self._adapters.values())
 
+    def is_operating_degraded(self) -> bool:
+        """
+        Return if the manager is operating in degraded mode.
+
+        On Linux, we're in degraded mode if mgmt control is not available.
+        This typically means we don't have NET_ADMIN/NET_RAW capabilities.
+        """
+        return IS_LINUX and self._mgmt_ctl is None
+
+    def on_scanner_start(self, scanner: BaseHaScanner) -> None:
+        """
+        Called when a scanner starts.
+
+        Subclasses can override this to perform custom actions when a scanner starts.
+        """
+
     def async_scanner_count(self, connectable: bool = True) -> int:
         """Return the number of scanners."""
         if connectable:

src/habluetooth/scanner.py

@@ -456,6 +456,7 @@ async def _async_start_attempt(self, attempt: int) -> bool:
             self._start_future = None
 
         self._log_start_success(attempt)
+        self._on_start_success()
         return True
 
     def _log_adapter_init_wait(self, attempt: int) -> None:

tests/conftest.py

@@ -224,3 +224,68 @@ def register_non_connectable_scanner() -> Generator[None, None, None]:
     cancel = manager.async_register_scanner(remote_scanner)
     yield
     cancel()
+
+
+class MockBluetoothManagerWithCallbacks(BluetoothManager):
+    """Mock bluetooth manager that tracks scanner start callbacks."""
+
+    def __init__(self, *args, **kwargs):
+        """Initialize the mock manager."""
+        super().__init__(*args, **kwargs)
+        self.scanner_start_calls = []
+
+    def on_scanner_start(self, scanner):
+        """Track scanner start calls."""
+        self.scanner_start_calls.append(scanner)
+        super().on_scanner_start(scanner)
+
+
+@pytest.fixture
+def mock_manager_with_scanner_callbacks() -> (
+    Generator[MockBluetoothManagerWithCallbacks, None, None]
+):
+    """Provide a mock BluetoothManager that tracks scanner start callbacks."""
+    mock_bluetooth_adapters = FakeBluetoothAdapters()
+    manager = MockBluetoothManagerWithCallbacks(
+        mock_bluetooth_adapters,
+        slot_manager=MagicMock(),
+    )
+
+    # Save the original manager
+    original_manager = get_manager()
+
+    # Set our mock manager as the global manager
+    set_manager(manager)
+
+    try:
+        yield manager
+    finally:
+        # Restore the original manager
+        set_manager(original_manager)
+
+
+@pytest_asyncio.fixture
+async def async_mock_manager_with_scanner_callbacks() -> (
+    AsyncGenerator[MockBluetoothManagerWithCallbacks, None]
+):
+    """Provide an async mock BluetoothManager that tracks scanner start callbacks."""
+    mock_bluetooth_adapters = FakeBluetoothAdapters()
+    manager = MockBluetoothManagerWithCallbacks(
+        mock_bluetooth_adapters,
+        slot_manager=MagicMock(),
+    )
+
+    # Setup the manager
+    await manager.async_setup()
+
+    # Save the original manager
+    original_manager = get_manager()
+
+    # Set our mock manager as the global manager
+    set_manager(manager)
+
+    try:
+        yield manager
+    finally:
+        # Restore the original manager
+        set_manager(original_manager)

tests/test_base_scanner.py

@@ -6,7 +6,7 @@
 import time
 from datetime import timedelta
 from typing import Any
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import ANY, MagicMock, Mock, patch
 
 import pytest
 from bleak.backends.device import BLEDevice
@@ -17,13 +17,15 @@
 from habluetooth import (
     BaseHaRemoteScanner,
     BaseHaScanner,
+    BluetoothManager,
     BluetoothScannerDevice,
     BluetoothScanningMode,
     HaBluetoothConnector,
     HaScannerDetails,
     HaScannerModeChange,
     HaScannerType,
     get_manager,
+    set_manager,
 )
 from habluetooth.const import (
     CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
@@ -44,6 +46,7 @@
     patch_bluetooth_time,
     utcnow,
 )
+from .conftest import FakeBluetoothAdapters, MockBluetoothManagerWithCallbacks
 
 
 class FakeScanner(BaseHaRemoteScanner):
@@ -1190,3 +1193,91 @@ def test_score_with_connections_in_progress_and_slots():
     # -50 (RSSI) - 10.1 (connection in progress) - 7.6 (last slot)
     assert score == -50 - 10.1 - 7.6
     assert score == -67.7
+
+
+@pytest.mark.asyncio
+async def test_on_scanner_start_callback_remote_scanner(
+    async_mock_manager_with_scanner_callbacks: MockBluetoothManagerWithCallbacks,
+) -> None:
+    """Test that on_scanner_start is called when a remote scanner starts."""
+    manager = async_mock_manager_with_scanner_callbacks
+
+    # Create a fake remote scanner
+    scanner = FakeScanner(
+        source="esp32_proxy",
+        adapter="esp32_proxy",
+        connector=None,
+        connectable=True,
+    )
+
+    # Simulate scanner start success
+    scanner._on_start_success()
+
+    # Verify the callback was called
+    assert len(manager.scanner_start_calls) == 1
+    assert manager.scanner_start_calls[0] is scanner
+
+
+@pytest.mark.asyncio
+async def test_on_scanner_start_multiple_scanners(
+    async_mock_manager_with_scanner_callbacks: MockBluetoothManagerWithCallbacks,
+) -> None:
+    """Test that on_scanner_start is called for multiple scanners."""
+    manager = async_mock_manager_with_scanner_callbacks
+
+    # Create multiple scanners
+    scanner1 = FakeScanner(
+        source="scanner1",
+        adapter="scanner1",
+        connector=None,
+        connectable=True,
+    )
+
+    scanner2 = FakeScanner(
+        source="scanner2",
+        adapter="scanner2",
+        connector=None,
+        connectable=True,
+    )
+
+    # Simulate both scanners starting
+    scanner1._on_start_success()
+    scanner2._on_start_success()
+
+    # Verify both callbacks were called
+    assert len(manager.scanner_start_calls) == 2
+    assert scanner1 in manager.scanner_start_calls
+    assert scanner2 in manager.scanner_start_calls
+
+
+@pytest.mark.asyncio
+async def test_scanner_without_manager() -> None:
+    """Test that _on_start_success handles scanner without manager gracefully."""
+    # Set a temporary manager for scanner creation
+    mock_bluetooth_adapters = FakeBluetoothAdapters()
+    temp_manager = BluetoothManager(
+        mock_bluetooth_adapters,
+        slot_manager=Mock(),
+    )
+
+    original_manager = get_manager()
+    set_manager(temp_manager)
+
+    try:
+        # Create a scanner
+        scanner = FakeScanner(
+            source="test",
+            adapter="test",
+            connector=None,
+            connectable=True,
+        )
+
+        # Clear the manager to simulate no manager scenario
+        scanner._manager = None  # type: ignore[assignment]
+
+        # Should not raise an exception
+        scanner._on_start_success()
+
+    finally:
+        # Restore original manager
+        set_manager(original_manager)

tests/test_manager.py

@@ -4,7 +4,7 @@
 import time
 from datetime import timedelta
 from typing import Any
-from unittest.mock import ANY, patch
+from unittest.mock import ANY, AsyncMock, Mock, patch
 
 import pytest
 from bleak_retry_connector import AllocationChange, Allocations, BleakSlotManager
@@ -1328,3 +1328,81 @@ class TestBluetoothManager2(BluetoothManager):
 
     TestBluetoothManager2(bluetooth_adapters, slot_manager)
     assert "does not implement _discover_service_info" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_is_operating_degraded_on_linux_with_mgmt() -> None:
+    """Test is_operating_degraded returns False on Linux with mgmt control."""
+    mock_bluetooth_adapters = FakeBluetoothAdapters()
+    manager = BluetoothManager(
+        mock_bluetooth_adapters,
+        slot_manager=Mock(),
+    )
+
+    with (
+        patch("habluetooth.manager.IS_LINUX", True),
+        patch.object(manager, "_mgmt_ctl", Mock()),
+    ):
+        # Mock mgmt_ctl being available
+        assert manager.is_operating_degraded() is False
+
+
+@pytest.mark.asyncio
+async def test_is_operating_degraded_on_linux_without_mgmt() -> None:
+    """Test is_operating_degraded returns True on Linux without mgmt control."""
+    mock_bluetooth_adapters = FakeBluetoothAdapters()
+    manager = BluetoothManager(
+        mock_bluetooth_adapters,
+        slot_manager=Mock(),
+    )
+
+    with patch("habluetooth.manager.IS_LINUX", True):
+        # mgmt_ctl is None by default
+        assert manager._mgmt_ctl is None
+        assert manager.is_operating_degraded() is True
+
+
+@pytest.mark.asyncio
+async def test_is_operating_degraded_on_non_linux() -> None:
+    """Test is_operating_degraded returns False on non-Linux systems."""
+    mock_bluetooth_adapters = FakeBluetoothAdapters()
+    manager = BluetoothManager(
+        mock_bluetooth_adapters,
+        slot_manager=Mock(),
+    )
+
+    with patch("habluetooth.manager.IS_LINUX", False):
+        # Should return False regardless of mgmt_ctl state
+        assert manager.is_operating_degraded() is False
+
+        # Even with mgmt_ctl set
+        manager._mgmt_ctl = Mock()
+        assert manager.is_operating_degraded() is False
+
+
+@pytest.mark.asyncio
+async def test_is_operating_degraded_after_permission_error() -> None:
+    """Test is_operating_degraded after mgmt setup fails with permission error."""
+    mock_bluetooth_adapters = FakeBluetoothAdapters()
+    manager = BluetoothManager(
+        mock_bluetooth_adapters,
+        slot_manager=Mock(),
+    )
+
+    with (
+        patch("habluetooth.manager.IS_LINUX", True),
+        patch("habluetooth.manager.MGMTBluetoothCtl") as mock_mgmt_class,
+    ):
+        # Make setup fail with permission error
+        mock_mgmt_instance = Mock()
+        mock_mgmt_instance.setup = AsyncMock(
+            side_effect=PermissionError("No permission")
+        )
+        mock_mgmt_class.return_value = mock_mgmt_instance
+
+        # Setup should handle the error and set mgmt_ctl to None
+        await manager.async_setup()
+
+        # Should be in degraded mode
+        assert manager._mgmt_ctl is None
+        assert manager.is_operating_degraded() is True

tests/test_scanner.py

@@ -42,7 +42,7 @@
     patch_bluetooth_time,
     utcnow,
 )
-from .conftest import FakeBluetoothAdapters
+from .conftest import FakeBluetoothAdapters, MockBluetoothManagerWithCallbacks
 
 DEVICE_FOUND = 0x0012
 ADV_MONITOR_DEVICE_FOUND = 0x002F
@@ -1611,3 +1611,33 @@ def test_ha_scanner_get_allocations_updates_dynamically() -> None:
         assert allocations is not None
         assert allocations.free == 1
         assert len(allocations.allocated) == 2
+
+
+@pytest.mark.asyncio
+async def test_on_scanner_start_callback(
+    async_mock_manager_with_scanner_callbacks: MockBluetoothManagerWithCallbacks,
+) -> None:
+    """Test that on_scanner_start is called when a local scanner starts."""
+    manager = async_mock_manager_with_scanner_callbacks
+
+    # Create a local scanner (it will get the manager from get_manager())
+    scanner = HaScanner(
+        mode=BluetoothScanningMode.ACTIVE,
+        adapter="hci0",
+        address="00:00:00:00:00:00",
+    )
+
+    # Register scanner with manager
+    manager.async_register_scanner(scanner)
+
+    # Setup the scanner
+    scanner.async_setup()
+
+    # Directly call _on_start_success to test the callback
+    # (In real usage, this is called by HaScanner._async_start_attempt
+    # after successful start)
+    scanner._on_start_success()
+
+    # Verify the callback was called
+    assert len(manager.scanner_start_calls) == 1
+    assert manager.scanner_start_calls[0] is scanner