diff --git a/src/ble_reticulum/linux_bluetooth_driver.py b/src/ble_reticulum/linux_bluetooth_driver.py index c14bc01..e25a8e4 100644 --- a/src/ble_reticulum/linux_bluetooth_driver.py +++ b/src/ble_reticulum/linux_bluetooth_driver.py @@ -288,6 +288,13 @@ class PeerConnection: peer_identity: Optional[bytes] = None # 16-byte identity hash +# Wall-clock bound for the _adapter_is_powered() D-Bus query. BlueZ property +# reads are normally well under a second; 5s is generous but keeps a +# hung/absent BlueZ from stalling the discovery loop indefinitely. Exposed as a +# module constant so tests can shorten it without waiting the full 5s. +POWERED_QUERY_TIMEOUT_S = 5.0 + + class LinuxBluetoothDriver(BLEDriverInterface): """ Linux implementation of BLE driver using bleak and bluezero. @@ -680,20 +687,110 @@ def detection_callback(device, advertisement_data): raise # Detect scanner callback corruption + # + # A service-filtered scan seeing nothing is NOT by itself evidence the + # adapter is wedged: at boot (and whenever no Reticulum peer is in + # range) it is normal for the filtered scan to return zero callbacks, + # because the peer's GATT server takes a few seconds to start + # advertising after ours. Declaring that a "corrupted / reboot + # required" stack is a false positive that fires on_error(critical) and + # tears the interface down right in the window the peer is coming up. + # + # Disambiguate with a short UNFILTERED health scan: if the adapter can + # see ANY device at all, it is alive and we are simply waiting for a + # Reticulum peer to advertise. Only a fully blind adapter (unfiltered + # scan also empty) counts as a genuine wedge. if callback_count[0] == 0: - self.consecutive_empty_scans += 1 - self._log(f"⚠️ Scanner corruption detected: 0 callbacks after {scan_time}s scan (streak: {self.consecutive_empty_scans})", "WARNING") - - if self.consecutive_empty_scans >= 3: - self._log("⚠️ CRITICAL: Bleak scanner callbacks not firing", "ERROR") - self._log("⚠️ Bluetooth/BlueZ/D-Bus state is corrupted", "ERROR") - self._log("⚠️ System reboot required to restore BLE scanning", "ERROR") + # Re-check connection state before starting a SECOND scan. The + # pause check at the top of _perform_scan ran before the main scan; + # a connection could have started during the main scan window. If so, + # running the unfiltered health scan now would start another BlueZ + # scan mid-connection and trigger an "Operation already in progress" + # error. Skip the health scan this cycle (it will run next cycle); + # treat the filtered result as-is without changing the streak. + if self._should_pause_scanning(): + self._log("Pausing health scan: connection(s) in progress", "DEBUG") + return + health_devices = await self._adapter_health_check() + if health_devices > 0: + if self.consecutive_empty_scans > 0: + self._log(f"✓ Scanner callbacks resumed after {self.consecutive_empty_scans} empty scans", "INFO") + self.consecutive_empty_scans = 0 + self._log( + f"✓ No Reticulum peer advertising, but adapter is healthy " + f"(unfiltered health scan saw {health_devices} device(s)) - not a wedge", + "INFO" + ) + elif health_devices < 0: + # The health scan itself FAILED (D-Bus/adapter error). A scan + # that cannot even run is direct evidence the adapter is in a + # bad state - unlike a clean zero, which in a quiet RF + # environment may be benign. Escalate regardless of whether we + # ever saw a healthy scan. + self.consecutive_empty_scans += 1 + self._log( + f"⚠️ Unfiltered health scan FAILED (adapter in fault state) " + f"(streak: {self.consecutive_empty_scans})", "WARNING" + ) + if self.consecutive_empty_scans >= 3: + self._log("⚠️ CRITICAL: unfiltered health scan repeatedly failing (adapter wedged)", "ERROR") + self._log("⚠️ Bluetooth/BlueZ/D-Bus state is corrupted", "ERROR") + self._log("⚠️ System reboot required to restore BLE scanning", "ERROR") + if self.on_error: + self.on_error("critical", + f"Adapter health scan has failed for {self.consecutive_empty_scans} " + f"consecutive scans (adapter in fault state). Bluetooth stack requires reboot.", + Exception("BleakScanner health scan failed")) + else: + # Clean zero: unfiltered scan ran but saw no devices at all. + # An empty scan CANNOT by itself distinguish a genuinely quiet + # RF environment (adapter healthy, simply no BLE devices in + # range) from a dead/wedged adapter (no devices because the + # adapter is down). Both look identical by device count. + # + # Resolve it with the adapter's OWN Powered state (independent + # of what it can see): a working adapter in a quiet room reads + # Powered=True; a powered-off/dead one reads Powered=False. + # Only escalate to a "reboot required" critical on POSITIVE + # fault evidence (adapter present but not powered). A powered + # adapter, or an unknown state, only warns - never cry wolf. + self.consecutive_empty_scans += 1 + self._log( + f"⚠️ Scanner blind: 0 Reticulum callbacks AND 0 devices in unfiltered " + f"health scan (streak: {self.consecutive_empty_scans})", "WARNING" + ) - if self.on_error: - self.on_error("critical", - f"Scanner callback failure detected (0 callbacks for {self.consecutive_empty_scans} consecutive scans). " - "Bluetooth stack requires reboot.", - Exception("BleakScanner callbacks not invoked")) + if self.consecutive_empty_scans >= 3: + powered = await self._adapter_is_powered() + if powered is False: + # The adapter is present on the bus but not powered + # while we need it to scan. That is genuine fault + # evidence (not a quiet room - a working adapter in a + # quiet room is powered). This covers both an adapter + # that was never healthy and one that went down. + self._log("⚠️ CRITICAL: adapter is not powered but discovery is running (adapter wedged/powered-off)", "ERROR") + self._log("⚠️ Bluetooth/BlueZ state is inconsistent", "ERROR") + self._log("⚠️ Reboot (or 'bluetoothctl power on') required to restore BLE scanning", "ERROR") + if self.on_error: + self.on_error("critical", + f"Adapter is not powered after {self.consecutive_empty_scans} " + f"consecutive blind scans (0 devices, adapter Powered=False). " + f"Bluetooth stack requires reboot or manual power-on.", + Exception("Adapter not powered during discovery")) + else: + # powered is True (working adapter in a quiet room) or + # None (could not determine state). No positive fault + # evidence - warn without mandating a reboot so a + # healthy, self-recovering adapter is never torn down + # over a quiet environment. + state = "powered" if powered is True else "state unknown" + self._log( + f"⚠️ No BLE devices in {self.consecutive_empty_scans} consecutive scans " + f"but adapter is {state}. Likely a quiet RF environment or a " + f"transient wedge that will recover on its own; not escalating to a " + f"reboot-required critical.", + "WARNING" + ) else: # Reset counter on successful callback if self.consecutive_empty_scans > 0: @@ -736,6 +833,98 @@ def detection_callback(device, advertisement_data): else: self._log(f"✗ {device.address} ({device.name or 'Unknown'}): service UUID mismatch (has {adv_data.service_uuids}, want {self.service_uuid})", "EXTRA") + async def _adapter_health_check(self) -> int: + """Run a short UNFILTERED scan to distinguish "no Reticulum peer in + range" (adapter healthy) from "adapter blind" (genuine wedge). + + The main discovery scan is service-filtered, so it returns zero + callbacks whenever no peer is advertising our Reticulum service - which + is the normal state at boot before the peer's GATT server starts + advertising. That is not evidence of a wedged adapter. To disambiguate, + we run a brief one-shot scan with NO service filter: if the adapter can + see any BLE device at all (including weak ones we would filter out by + RSSI), it is alive and we are simply waiting for a peer. + + Returns the number of distinct devices seen. A negative value (-1) + indicates the health scan itself failed (D-Bus/adapter error) rather + than returning zero devices - the caller treats a failed scan as direct + evidence of an adapter fault (unlike a clean zero, which in a quiet RF + environment may be benign). + """ + try: + devices = await BleakScanner.discover(timeout=1.0) + count = len(devices) + self._log(f"🩺 Adapter health scan: {count} device(s) visible (unfiltered)", "DEBUG") + return count + except Exception as e: + self._log(f"⚠️ Adapter health scan failed: {e}", "WARNING") + return -1 + + async def _adapter_is_powered(self) -> Optional[bool]: + """Query the adapter's OWN BlueZ "Powered" state via D-Bus. + + Why this is needed: an empty unfiltered scan cannot distinguish a + genuinely quiet RF environment (adapter healthy, simply no BLE devices + in range) from a dead/wedged adapter (no devices BECAUSE the adapter is + down). Both look identical by device count. The adapter's own Powered + property is an independent signal that does not depend on what the + adapter can see: + * a working adapter in a quiet room reads Powered=True; + * a dead/powered-off adapter reads Powered=False (or its D-Bus object + is absent). + This is what lets us issue a "reboot required" critical only on genuine + fault evidence instead of on an empty scan. + + Returns True (adapter powered), False (adapter present but not + powered), or None if the state could not be determined (D-Bus + unavailable, no bluez, a query error, or the query timed out). A None + means "no positive fault evidence" - the caller must not escalate to + critical on it. + + The whole query is bounded by a wall-clock timeout: this is called + from inside _perform_scan, so if BlueZ stops replying to D-Bus (e.g. + the adapter is wedged at the D-Bus level), the query must not hang + discovery. A timeout is treated as an unknown state, not a fault. + """ + if not HAS_DBUS: + return None + try: + # Bound the whole D-Bus round-trip so a hung/absent BlueZ cannot + # wedge the discovery loop. BlueZ property reads are normally well + # under a second; 5s is generous but keeps this from stalling the + # scan cycle indefinitely. The bus is connected and disconnected + # inside _query (its own finally), so a wait_for timeout - which + # cancels _query - still releases the connection. + async def _query(): + bus = await MessageBus(bus_type=BusType.SYSTEM).connect() + try: + introspection = await bus.introspect('org.bluez', self.adapter_path) + adapter_obj = bus.get_proxy_object('org.bluez', self.adapter_path, introspection) + adapter_iface = adapter_obj.get_interface('org.bluez.Adapter1') + return await adapter_iface.get_powered() + finally: + try: + bus.disconnect() + except Exception: + pass + powered = await asyncio.wait_for(_query(), timeout=POWERED_QUERY_TIMEOUT_S) + self._log(f"🩺 Adapter powered state: {powered}", "DEBUG") + return bool(powered) + except asyncio.TimeoutError: + # BlueZ did not reply in time. This is NOT proof the adapter is + # broken (the bus may just be busy) - report unknown and let the + # caller treat it as "no positive fault evidence." Crucially, this + # returns control to the scan loop instead of hanging it. + self._log("⚠️ Timed out querying adapter powered state (BlueZ not responding); treating as unknown", "DEBUG") + return None + except Exception as e: + # Could not determine adapter state. This alone is NOT proof the + # adapter is broken (D-Bus may be busy, adapter object not yet + # registered at boot, etc.) - report unknown and let the caller + # treat it as "no positive fault evidence." + self._log(f"⚠️ Could not determine adapter powered state: {e}", "DEBUG") + return None + # ======================================================================== # Advertising (Peripheral Mode) # ======================================================================== diff --git a/tests/test_adapter_health_check_wedge.py b/tests/test_adapter_health_check_wedge.py new file mode 100644 index 0000000..ad7db99 --- /dev/null +++ b/tests/test_adapter_health_check_wedge.py @@ -0,0 +1,460 @@ +""" +Regression test: adapter health-check disambiguation for the scanner-wedge detector. + +**Problem**: The discovery scan is SERVICE-FILTERED (only devices advertising the +Reticulum service UUID trigger the detection callback). At boot, no peer has +started advertising yet, so the filtered scan legitimately returns zero callbacks +for several scans while the peer's GATT server comes up. The old detector treated +3 empty filtered scans as "adapter corrupted / system reboot required" and fired +on_error("critical") - a FALSE POSITIVE that tore the BLE interface down right in +the window the peer was about to come online. (Observed on two Pi Zero 2 W: both +wedge at boot, ~3-5 empty filtered scans, then recover on their own - which a +genuinely wedged adapter can never do without a reboot.) + +**Fix**: When a service-filtered scan returns zero callbacks, re-check that no +connection is in progress (the pause check at the top of _perform_scan ran +before the main scan; a connection could have started during that window, and +a second scan mid-connection would collide with it). Then run a short UNFILTERED +health scan (BleakScanner.discover with no service filter). Outcomes: + * health scan sees >= 1 device -> adapter is ALIVE, just waiting for a + Reticulum peer; reset the streak, do NOT fire critical (the false-positive + case at boot); + * health scan itself raises (D-Bus/adapter error) -> a scan that cannot run + is direct evidence the adapter is in a fault state; escalate to critical; + * health scan runs but sees 0 devices (clean zero) -> an empty scan CANNOT + by itself distinguish a genuinely quiet RF environment (adapter healthy, + no BLE devices in range) from a dead/wedged adapter. Disambiguate with the + adapter's OWN BlueZ "Powered" state (_adapter_is_powered): + - Powered=False (adapter present but not powered) -> genuine fault -> + fire the "reboot required" critical (this catches both an adapter that + was never healthy and one that went down); + - Powered=True (working adapter, quiet room) OR unknown (could not + determine) -> no positive fault evidence -> warn only, never mandating + a reboot, so a healthy self-recovering adapter is never torn down over + a quiet environment (this is the Greptile 3/5 P1 overclaim fix). + +**Test strategy**: Drive the REAL LinuxBluetoothDriver._perform_scan() with a +mocked BleakScanner class and a mocked _adapter_is_powered(). The main scan's +detection callback never fires (empty filtered scan) unless fire_callback=True; +the health scan's BleakScanner.discover and the Powered-state result are +controlled per test. No real Bluetooth is touched. + +The healthy-adapter case is the RED->GREEN assertion: with the OLD code (no +health check) the empty filtered scan would fire on_error("critical") after 3 +scans; with the fix it must not. +""" + +import pytest +import sys +import os +import asyncio +import threading +from unittest.mock import Mock, AsyncMock, patch + +# Add src to path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../src')) + +# Mock RNS module before importing (the driver has no Reticulum dependency, but +# keep the pattern for environments where ble_reticulum/__init__ or a sibling +# module expects RNS constants). Tolerate RNS being absent (e.g. this box). +try: + import RNS + if not hasattr(RNS, 'LOG_INFO'): + RNS.LOG_CRITICAL = 0 + RNS.LOG_ERROR = 1 + RNS.LOG_WARNING = 2 + RNS.LOG_NOTICE = 3 + RNS.LOG_INFO = 4 + RNS.LOG_VERBOSE = 5 + RNS.LOG_DEBUG = 6 + RNS.LOG_EXTREME = 7 + RNS.log = Mock() +except ImportError: + pass + +SERVICE_UUID = "37145b00-442d-4a94-917f-8f42c5da28e3" + + +def _make_driver(adapter_powered=None): + """Build a LinuxBluetoothDriver without running the heavy __init__. + + Only the attributes _perform_scan / _adapter_health_check / + _adapter_is_powered touch are set. Using __new__ avoids the real + constructor, which requires RNS + BlueZ plumbing that is irrelevant to the + wedge-detection logic under test. + + adapter_powered controls the mocked _adapter_is_powered() result: + True -> adapter is powered (working, e.g. quiet room) + False -> adapter present but NOT powered (genuine fault) + None -> state could not be determined (no positive fault evidence) + """ + from ble_reticulum import linux_bluetooth_driver as m + d = m.LinuxBluetoothDriver.__new__(m.LinuxBluetoothDriver) + d._running = True + d.consecutive_empty_scans = 0 + d._log = Mock() + d.on_error = Mock() + d.service_uuid = SERVICE_UUID + d.min_rssi = -60 + # saver => 0.5s main-scan window (keeps tests fast without patching asyncio) + d.power_mode = "saver" + d._should_pause_scanning = Mock(return_value=False) + d.on_device_discovered = Mock() + d._adapter_is_powered = AsyncMock(return_value=adapter_powered) + return d + + +def _make_driver_real_powered(): + """Same as _make_driver but keeps the REAL _adapter_is_powered() (no + AsyncMock) so tests can drive the actual D-Bus probe with a mocked bus. + Sets the minimal attributes the probe reads (_log, adapter_path).""" + from ble_reticulum import linux_bluetooth_driver as m + d = m.LinuxBluetoothDriver.__new__(m.LinuxBluetoothDriver) + d._log = Mock() + d.adapter_path = "/org/bluez/hci0" + return d + + +def _mock_scanner_class(discover_return, fire_callback=False): + """Return a replacement for linux_bluetooth_driver.BleakScanner. + + The main discovery scan does `BleakScanner(detection_callback=..., + service_uuids=...)` then `await scanner.start()/stop()`. By default the + detection callback is never invoked (empty filtered scan). With + fire_callback=True, start() invokes the captured detection_callback once + with a device advertising the Reticulum service UUID, so the scan reports a + real discovery (callback_count > 0). The health check does + `await BleakScanner.discover(timeout=...)` which returns discover_return. + """ + BS = Mock() + BS.discover = AsyncMock(return_value=discover_return) + + def _factory(*args, **kwargs): + cb = kwargs.get("detection_callback") + inst = Mock() + + async def _start(*a, **k): + if fire_callback and cb is not None: + device = Mock() + device.address = "AA:BB:CC:DD:EE:FF" + device.name = "PeerPi" + adv = Mock() + adv.rssi = -50 + adv.service_uuids = [SERVICE_UUID] + adv.manufacturer_data = {} + cb(device, adv) + + inst.start = _start + inst.stop = AsyncMock() + return inst + + BS.side_effect = _factory + return BS + + +class TestAdapterHealthCheckWedgeDetection: + """Empty service-filtered scan must be disambiguated by an unfiltered scan.""" + + @pytest.mark.asyncio + async def test_empty_filtered_but_healthy_adapter_is_not_a_wedge(self): + """ + RED before fix / GREEN after fix. + + The filtered scan sees nothing (peer not advertising yet) but the + unfiltered health scan sees devices -> adapter is alive. Must NOT fire + on_error("critical") and must reset the empty-streak to 0, even across + many scans (this is the boot race that used to false-positive). + """ + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver() + two_devices = [Mock(), Mock()] + # Bind the mock to a local so we can assert on its call count after the + # patch context has exited (m.BleakScanner reverts to the real class + # once the with-block ends, where .discover is a plain function). + BS = _mock_scanner_class(two_devices) + with patch.object(m, "BleakScanner", BS): + for _ in range(5): + await d._perform_scan() + d.on_error.assert_not_called() + assert d.consecutive_empty_scans == 0 + # Health check actually ran on every empty filtered scan. + assert BS.discover.await_count >= 1 + + @pytest.mark.asyncio + async def test_not_powered_adapter_declares_wedge_after_3(self): + """ + Genuine fault: after 3 clean-zero blind scans, the adapter is present + on the bus but NOT powered (Powered=False). That is positive fault + evidence (a working adapter in a quiet room IS powered), so the + "reboot required" critical must fire. Covers both an adapter that was + never healthy and one that went down. + """ + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver(adapter_powered=False) # adapter present but not powered + with patch.object(m, "BleakScanner", _mock_scanner_class([])): + for _ in range(3): + await d._perform_scan() + # After the 3rd blind scan the critical error must fire. + d.on_error.assert_called() + args = d.on_error.call_args[0] + assert args[0] == "critical" + assert d.consecutive_empty_scans >= 3 + # The Powered state was actually consulted before escalating. + d._adapter_is_powered.assert_awaited() + + @pytest.mark.asyncio + async def test_quiet_room_powered_does_not_reboot(self): + """ + Greptile P1 overclaim fix (quiet room): a clean zero on a POWERED + adapter is a working adapter in a room with no BLE devices, NOT a + fault. Must NOT fire the reboot-required critical; it only warns. + This holds even after many scans past the threshold. + """ + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver(adapter_powered=True) # working adapter, quiet room + with patch.object(m, "BleakScanner", _mock_scanner_class([])): + for _ in range(5): # well past the 3-scan threshold + await d._perform_scan() + # Streak incremented, but no critical fired (quiet room, not a fault). + assert d.consecutive_empty_scans >= 3 + d.on_error.assert_not_called() + + @pytest.mark.asyncio + async def test_unknown_powered_state_does_not_reboot(self): + """ + When the Powered state cannot be determined (None), there is no + positive fault evidence - fail safe and warn without mandating a + reboot, so a healthy self-recovering adapter is never torn down. + """ + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver(adapter_powered=None) # state unknown + with patch.object(m, "BleakScanner", _mock_scanner_class([])): + for _ in range(5): + await d._perform_scan() + assert d.consecutive_empty_scans >= 3 + d.on_error.assert_not_called() + + @pytest.mark.asyncio + async def test_successful_scan_resets_streak(self): + """ + A filtered scan that DOES discover a Reticulum device (detection + callback fires) proves the adapter is working: it must reset the + empty-streak and NOT fire any error. This covers the callback-fired + path in _perform_scan. + """ + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver() + d.consecutive_empty_scans = 2 # start from a non-zero streak + BS = _mock_scanner_class([], fire_callback=True) + with patch.object(m, "BleakScanner", BS): + await d._perform_scan() + assert d.consecutive_empty_scans == 0 + d.on_error.assert_not_called() + # The device was forwarded to the discovered-device callback. + d.on_device_discovered.assert_called_once() + # No health scan / powered query needed when the filtered scan works. + assert BS.discover.await_count == 0 + d._adapter_is_powered.assert_not_awaited() + + @pytest.mark.asyncio + async def test_blind_then_healthy_resets_streak(self): + """Two blind scans, then the adapter recovers -> streak resets, no critical.""" + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver() + one_device = [Mock()] + # Scans 1-2: adapter blind (unfiltered empty). Scan 3: adapter recovers. + with patch.object(m, "BleakScanner", _mock_scanner_class([])): + await d._perform_scan() + await d._perform_scan() + assert d.consecutive_empty_scans == 2 + d.on_error.assert_not_called() + with patch.object(m, "BleakScanner", _mock_scanner_class(one_device)): + await d._perform_scan() + assert d.consecutive_empty_scans == 0 + d.on_error.assert_not_called() + + @pytest.mark.asyncio + async def test_health_scan_failure_is_fault_evidence(self): + """If the health scan itself raises, that is direct evidence the adapter + is in a fault state (a scan that cannot run is not a quiet room). It + escalates to critical regardless of the Powered state - fail closed.""" + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver(adapter_powered=True) # even if "powered", scan failing is fault + BS = _mock_scanner_class([]) + BS.discover = AsyncMock(side_effect=RuntimeError("dbus gone")) + with patch.object(m, "BleakScanner", BS): + for _ in range(3): + await d._perform_scan() + d.on_error.assert_called() + assert d.on_error.call_args[0][0] == "critical" + # The failure short-circuits before consulting the Powered state. + d._adapter_is_powered.assert_not_awaited() + + @pytest.mark.asyncio + async def test_connection_started_during_scan_skips_health_scan(self): + """ + Greptile P1 race fix: a connection that starts during the main-scan + window must suppress the follow-up health scan. Re-checking + _should_pause_scanning() before the unfiltered scan prevents a second + BlueZ scan from colliding with an active connection ("Operation already + in progress"). Must NOT call the health scan and must NOT fire critical. + """ + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver() + # Model the real race: the pause check at the TOP of _perform_scan sees + # "not paused" (False) so the main scan proceeds, but a connection starts + # DURING the scan window, so the re-check at the health-scan guard (after + # the main scan) sees "paused" (True). Each scan calls the check twice: + # top, then health-guard. So the sequence is F,T,F,T,... + d._should_pause_scanning = Mock(side_effect=[False, True] * 5) + BS = _mock_scanner_class([]) + with patch.object(m, "BleakScanner", BS): + for _ in range(5): + await d._perform_scan() + # The unfiltered health scan must never have run (guard caught the + # in-progress connection each cycle). + assert BS.discover.await_count == 0 + d.on_error.assert_not_called() + + @pytest.mark.asyncio + async def test_adapter_health_check_returns_count(self): + """_adapter_health_check returns the number of distinct devices seen.""" + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver() + BS = _mock_scanner_class([Mock(), Mock(), Mock()]) + with patch.object(m, "BleakScanner", BS): + count = await d._adapter_health_check() + assert count == 3 + + # And 0 when nothing is visible. + BS0 = _mock_scanner_class([]) + with patch.object(m, "BleakScanner", BS0): + count0 = await d._adapter_health_check() + assert count0 == 0 + + +class TestAdapterIsPowered: + """The real _adapter_is_powered() D-Bus probe: powered / not-powered / + unknown / no-dbus outcomes, and that the bus connection is always closed.""" + + def _patched_bus(self, m, powered_result): + """Patch m.MessageBus so the driver's _adapter_is_powered() D-Bus + sequence yields the given powered value (or raises). + + The driver calls: await connect(); await introspect(); + get_proxy_object() [sync]; get_interface() [sync]; await get_powered(). + Returns (bus, adapter_iface) for assertions and the started patch. + """ + bus = Mock() + bus.connect = AsyncMock(return_value=bus) + # `await bus.introspect(...)` -> Mock (truthy, unused beyond existence) + bus.introspect = AsyncMock(return_value=Mock()) + # `bus.get_proxy_object(...)` [SYNC] -> adapter_obj + adapter_obj = Mock() + bus.get_proxy_object = Mock(return_value=adapter_obj) + # `adapter_obj.get_interface('org.bluez.Adapter1')` [SYNC] -> iface + adapter_iface = Mock() + adapter_obj.get_interface = Mock(return_value=adapter_iface) + # `await adapter_iface.get_powered()` -> powered value (or raises) + if isinstance(powered_result, Exception): + adapter_iface.get_powered = AsyncMock(side_effect=powered_result) + else: + adapter_iface.get_powered = AsyncMock(return_value=powered_result) + + mb = patch.object(m, "MessageBus") + mock_bus_class = mb.start() # start() returns the mock (not the _patch) + mock_bus_class.return_value = bus + return bus, adapter_iface, mb + + @pytest.mark.asyncio + async def test_powered_true(self): + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver_real_powered() + with patch.object(m, "HAS_DBUS", True): + bus, adapter_iface, mb = self._patched_bus(m, True) + try: + result = await d._adapter_is_powered() + finally: + mb.stop() + assert result is True + adapter_iface.get_powered.assert_awaited() + bus.disconnect.assert_called() + + @pytest.mark.asyncio + async def test_powered_false(self): + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver_real_powered() + with patch.object(m, "HAS_DBUS", True): + bus, adapter_iface, mb = self._patched_bus(m, False) + try: + result = await d._adapter_is_powered() + finally: + mb.stop() + assert result is False + bus.disconnect.assert_called() + + @pytest.mark.asyncio + async def test_query_error_is_unknown(self): + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver_real_powered() + with patch.object(m, "HAS_DBUS", True): + bus, _, mb = self._patched_bus(m, RuntimeError("UnknownObject")) + try: + result = await d._adapter_is_powered() + finally: + mb.stop() + # A query failure is "unknown", not a positive fault. + assert result is None + bus.disconnect.assert_called() + + @pytest.mark.asyncio + async def test_no_dbus_is_unknown(self): + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver_real_powered() + with patch.object(m, "HAS_DBUS", False): + result = await d._adapter_is_powered() + # No D-Bus at all: unknown, and no bus connection is attempted. + assert result is None + + @pytest.mark.asyncio + async def test_hung_bluez_times_out_as_unknown(self): + """Greptile P1 (timeout): if BlueZ stops replying, the query must not + hang the discovery loop. The whole D-Bus round-trip is bounded by a + wall-clock timeout; a timeout is treated as an unknown state (no + positive fault evidence) and returns control to the scan loop. + + connect() is made to sleep well past the (patched) timeout so the + query genuinely hangs; wait_for must cancel it and return None quickly. + """ + from ble_reticulum import linux_bluetooth_driver as m + d = _make_driver_real_powered() + + bus = Mock() + + async def _hang_forever(): + # Simulate BlueZ not replying: never completes. + await asyncio.sleep(60.0) + + # connect() hangs; introspect/get_powered are irrelevant (never reached + # before the timeout fires). + bus.connect = _hang_forever + bus.disconnect = Mock() + + import time + mb = patch.object(m, "MessageBus") + mock_bus_class = mb.start() # start() returns the mock (not the _patch) + mock_bus_class.return_value = bus + # Patch the timeout down so the test is fast (no 5s wait). + timeout_patch = patch.object(m, "POWERED_QUERY_TIMEOUT_S", 0.05) + with patch.object(m, "HAS_DBUS", True), timeout_patch: + start = time.monotonic() + try: + result = await d._adapter_is_powered() + finally: + mb.stop() + elapsed = time.monotonic() - start + # Timed out -> unknown, and it did NOT hang (returned well before the + # 60s sleep would have, and around the 0.05s bound, not unbounded). + assert result is None + assert elapsed < 5.0 # sanity: returned promptly, not stuck