diff --git a/tests/test_scanner.py b/tests/test_scanner.py index 1b516e67..fa2c7f34 100644 --- a/tests/test_scanner.py +++ b/tests/test_scanner.py @@ -1711,3 +1711,76 @@ async def test_on_scanner_start_callback( # Verify the callback was called assert len(manager.scanner_start_calls) == 1 assert manager.scanner_start_calls[0] is scanner + + +@pytest.mark.parametrize("exc", [FileNotFoundError("no dbus"), BleakError("nope")]) +def test_create_bleak_scanner_wraps_init_error(exc: Exception) -> None: + """``create_bleak_scanner`` wraps FileNotFoundError/BleakError as RuntimeError.""" + with ( + patch.object(scanner, "IS_LINUX", True), + patch.object(scanner, "IS_MACOS", False), + patch( + "habluetooth.scanner.OriginalBleakScanner", + side_effect=exc, + ), + pytest.raises(RuntimeError, match="Failed to initialize Bluetooth"), + ): + create_bleak_scanner(None, BluetoothScanningMode.ACTIVE, "hci0") + + +@pytest.mark.asyncio +@pytest.mark.skipif(NOT_POSIX) +@pytest.mark.parametrize("exc", [TimeoutError("slow"), BleakError("nope")]) +async def test_async_stop_scanner_logs_when_scanner_stop_raises( + caplog: pytest.LogCaptureFixture, exc: Exception +) -> None: + """``_async_stop_scanner`` logs and clears the scanner when ``.stop()`` raises.""" + mock_scanner = MagicMock() + mock_scanner.start = AsyncMock() + mock_scanner.stop = AsyncMock(side_effect=exc) + with patch("habluetooth.scanner.OriginalBleakScanner", return_value=mock_scanner): + scanner = HaScanner(BluetoothScanningMode.ACTIVE, "hci0", "AA:BB:CC:DD:EE:FF") + scanner.async_setup() + await scanner.async_start() + caplog.clear() + await scanner.async_stop() + assert "Error stopping scanner" in caplog.text + assert scanner.scanner is None + + +@pytest.mark.asyncio +@pytest.mark.skipif(NOT_POSIX) +async def test_async_force_stop_discovery_logs_on_timeout( + caplog: pytest.LogCaptureFixture, +) -> None: + """Force-stop logs an error when ``stop_discovery`` times out.""" + ha_scanner = HaScanner(BluetoothScanningMode.ACTIVE, "hci0", "AA:BB:CC:DD:EE:FF") + ha_scanner.async_setup() + with patch("habluetooth.scanner.stop_discovery", side_effect=TimeoutError("slow")): + await ha_scanner._async_force_stop_discovery() + assert "Timeout force stopping scanner" in caplog.text + await ha_scanner.async_stop() + + +@pytest.mark.asyncio +@pytest.mark.skipif(NOT_POSIX) +async def test_async_force_stop_discovery_logs_on_unexpected_error( + caplog: pytest.LogCaptureFixture, +) -> None: + """Force-stop logs an error when ``stop_discovery`` raises an unexpected error.""" + ha_scanner = HaScanner(BluetoothScanningMode.ACTIVE, "hci0", "AA:BB:CC:DD:EE:FF") + ha_scanner.async_setup() + with patch("habluetooth.scanner.stop_discovery", side_effect=BleakError("boom")): + await ha_scanner._async_force_stop_discovery() + assert "Failed to force stop scanner" in caplog.text + await ha_scanner.async_stop() + + +@pytest.mark.asyncio +async def test_get_allocations_returns_none_without_slot_manager() -> None: + """``HaScanner.get_allocations`` returns None when manager has no slot manager.""" + ha_scanner = HaScanner(BluetoothScanningMode.ACTIVE, "hci0", "AA:BB:CC:DD:EE:FF") + ha_scanner.async_setup() + with patch.object(get_manager(), "slot_manager", None): + assert ha_scanner.get_allocations() is None + await ha_scanner.async_stop()