import asyncio from collections.abc import Iterator from pathlib import Path from typing import Literal import pytest from bleak.backends.device import BLEDevice from bleak.backends.scanner import AdvertisementData from pytest import MonkeyPatch import k1link.device_plugins.xgrids_k1.ble.scanner as scanner_module from k1link.device_plugins.xgrids_k1.ble.runtime_arbiter import ( BleRuntimeBusy, bind_ble_runtime_owner_loop, configure_ble_runtime_process_lease, reset_ble_runtime_arbiter_for_tests, run_ble_operation, ) from k1link.device_plugins.xgrids_k1.ble.scanner import ( advertisement_record, capture_discovered_device, connected_device_capture, connected_device_recovery_name, connected_device_recovery_snapshot, demote_connected_device_handle_after_gatt_failure, discover_known_device_capture_for_status_read, discovered_device, discovered_device_selection, mark_captured_device_gatt_validated, pin_connected_device_handle, retrieve_connected_device_capture, retrieve_known_device_capture_for_status_read, scan, ) RecoveryOperationKind = Literal["status-read", "wifi-provision", "ap-enable"] async def _retrieve_connected_inside_ble_lease( macos_uuid: str, *, device_session_id: str, operation_kind: RecoveryOperationKind = "status-read", ) -> scanner_module.CapturedDiscoveredDevice | None: return await run_ble_operation( operation_kind, hard_timeout_seconds=2.0, operation=lambda _progress: retrieve_connected_device_capture( macos_uuid, device_session_id=device_session_id, ), ) @pytest.fixture(autouse=True) def reset_runtime_handle_lease(tmp_path: Path) -> Iterator[None]: scanner_module.reset_runtime_handles_for_tests() reset_ble_runtime_arbiter_for_tests() configure_ble_runtime_process_lease(tmp_path) yield scanner_module.reset_runtime_handles_for_tests() reset_ble_runtime_arbiter_for_tests() class _FakeCoreBluetoothIdentifier: def __init__(self, value: str) -> None: self.value = value def UUIDString(self) -> str: # noqa: N802 - mirrors NSUUID return self.value class _FakeCoreBluetoothPeripheral: def __init__(self, identifier: _FakeCoreBluetoothIdentifier, name: str) -> None: self._identifier = identifier self._name = name def identifier(self) -> _FakeCoreBluetoothIdentifier: return self._identifier def name(self) -> str: return self._name class _FakeCentralManager: def __init__( self, responses: list[list[_FakeCoreBluetoothPeripheral]], ) -> None: self.responses = responses self.requests: list[list[object]] = [] def retrievePeripheralsWithIdentifiers_( # noqa: N802 - mirrors CoreBluetooth self, identifiers: list[object], ) -> list[_FakeCoreBluetoothPeripheral]: self.requests.append(identifiers) return self.responses.pop(0) class _FakeCoreBluetoothManager: def __init__( self, loop: asyncio.AbstractEventLoop, central_manager: _FakeCentralManager, ) -> None: self.event_loop = loop self.central_manager = central_manager self.ready_calls = 0 async def wait_until_ready(self) -> None: self.ready_calls += 1 def _active_status_read_runtime(owner_epoch: int = 7) -> dict[str, object]: return { "owner_epoch": owner_epoch, "owner_loop_bound": True, "active_operation_kind": "status-read", "cleanup_pending": False, "poisoned": False, } def test_advertisement_record_marks_k1_candidate() -> None: device = BLEDevice("TEST-UUID", "Unknown", details=None) advertisement = AdvertisementData( local_name="Lixel-K1-1234", manufacturer_data={123: b"\x01\x02"}, service_data={"service": b"\xaa"}, service_uuids=["B", "a"], tx_power=-4, rssi=-52, platform_data=(), ) record = advertisement_record(device, advertisement) assert record["k1_name_candidate"] is True assert record["id_kind"] == "corebluetooth_uuid" assert record["manufacturer_data_hex"] == {"123": "0102"} assert record["service_data_hex"] == {"service": "aa"} assert record["service_uuids"] == ["B", "a"] def test_advertisement_record_marks_synthetic_xgr_name_as_k1_candidate() -> None: device = BLEDevice("00000000-0000-0000-0000-000000000001", "XGR-TEST01", None) advertisement = AdvertisementData( local_name="XGR-TEST01", manufacturer_data={}, service_data={}, service_uuids=[], tx_power=0, rssi=-60, platform_data=(), ) record = advertisement_record(device, advertisement) assert record["k1_name_candidate"] is True def test_scan_retains_the_live_corebluetooth_handle( monkeypatch: MonkeyPatch, ) -> None: device = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) advertisement = AdvertisementData( local_name="XGR-LIVE", manufacturer_data={}, service_data={}, service_uuids=[], tx_power=0, rssi=-41, platform_data=(), ) async def fake_discover(**_kwargs: object) -> dict[str, tuple[BLEDevice, AdvertisementData]]: return {"LIVE-UUID": (device, advertisement)} monkeypatch.setattr( "k1link.device_plugins.xgrids_k1.ble.scanner.BleakScanner.discover", fake_discover, ) async def scenario() -> object: result = await scan(1.0) assert result["devices"][0]["macos_uuid"] == "LIVE-UUID" assert discovered_device("LIVE-UUID") is device assert discovered_device_selection("LIVE-UUID").from_fresh_scan is True return result asyncio.run(scenario()) def test_scan_start_invalidates_previous_lease_and_failure_leaves_it_empty( monkeypatch: MonkeyPatch, ) -> None: old_device = BLEDevice("OLD-UUID", "XGR-OLD", details=object()) old_advertisement = AdvertisementData( local_name="XGR-OLD", manufacturer_data={}, service_data={}, service_uuids=[], tx_power=0, rssi=-41, platform_data=(), ) async def scenario() -> None: async def initial_discover( **_kwargs: object, ) -> dict[str, tuple[BLEDevice, AdvertisementData]]: return {old_device.address: (old_device, old_advertisement)} monkeypatch.setattr(scanner_module.BleakScanner, "discover", initial_discover) await scan(1.0) assert discovered_device(old_device.address) is old_device failing_scan_started = asyncio.Event() release_failing_scan = asyncio.Event() async def failing_discover(**_kwargs: object) -> object: failing_scan_started.set() await release_failing_scan.wait() raise RuntimeError("synthetic BLE scan failure") monkeypatch.setattr(scanner_module.BleakScanner, "discover", failing_discover) scan_task = asyncio.create_task(scan(1.0)) await failing_scan_started.wait() # Starting a new explicit scan revokes the prior generation before I/O. assert discovered_device(old_device.address) is None assert discovered_device_selection(old_device.address).from_fresh_scan is False release_failing_scan.set() with pytest.raises(RuntimeError, match="synthetic BLE scan failure"): await scan_task assert discovered_device(old_device.address) is None assert discovered_device_selection(old_device.address).from_fresh_scan is False asyncio.run(scenario()) def test_process_arbiter_rejects_overlapping_scan(monkeypatch: MonkeyPatch) -> None: older_device = BLEDevice("OLDER-UUID", "XGR-OLDER", details=object()) older_advertisement = AdvertisementData( local_name="XGR-OLDER", manufacturer_data={}, service_data={}, service_uuids=[], tx_power=0, rssi=-51, platform_data=(), ) async def scenario() -> None: older_scan_started = asyncio.Event() release_older_scan = asyncio.Event() busy_scan_admitted = False async def blocked_discover( **_kwargs: object, ) -> dict[str, tuple[BLEDevice, AdvertisementData]]: older_scan_started.set() await release_older_scan.wait() return {older_device.address: (older_device, older_advertisement)} monkeypatch.setattr(scanner_module.BleakScanner, "discover", blocked_discover) older_task = asyncio.create_task(scan(1.0)) await older_scan_started.wait() def record_busy_admission() -> None: nonlocal busy_scan_admitted busy_scan_admitted = True with pytest.raises(BleRuntimeBusy) as raised: await scan(1.0, on_admitted=record_busy_admission) assert raised.value.reason_code == "ble-runtime-busy" assert raised.value.active_operation_kind == "scan" assert busy_scan_admitted is False release_older_scan.set() await older_task assert discovered_device(older_device.address) is older_device asyncio.run(scenario()) def test_owner_epoch_rebind_rejects_late_handle_publish() -> None: old_device = BLEDevice("OLD-UUID", "XGR-OLD", details=object()) late_device = BLEDevice("LATE-UUID", "XGR-LATE", details=object()) first_loop = asyncio.new_event_loop() try: first_epoch, generation = first_loop.run_until_complete(_publish_owned_handle(old_device)) finally: first_loop.close() second_loop = asyncio.new_event_loop() try: second_epoch = second_loop.run_until_complete( _rebind_and_reject_late_publish( generation, first_epoch, old_device, late_device, ) ) assert second_epoch > first_epoch assert scanner_module._runtime_handles == {} # noqa: SLF001 finally: second_loop.close() async def _publish_owned_handle(device: BLEDevice) -> tuple[int, int]: owner_epoch = bind_ble_runtime_owner_loop() generation = scanner_module._begin_scan_generation(owner_epoch) # noqa: SLF001 scanner_module._publish_scan_handles( # noqa: SLF001 generation, owner_epoch, {device.address: device}, ) assert discovered_device(device.address) is device return owner_epoch, generation async def _rebind_and_reject_late_publish( generation: int, old_owner_epoch: int, old_device: BLEDevice, late_device: BLEDevice, ) -> int: new_owner_epoch = bind_ble_runtime_owner_loop() assert discovered_device(old_device.address) is None scanner_module._publish_scan_handles( # noqa: SLF001 generation, old_owner_epoch, {old_device.address: old_device, late_device.address: late_device}, ) assert discovered_device(late_device.address) is None return new_owner_epoch def test_runtime_handle_from_latest_scan_does_not_expire_by_age( monkeypatch: MonkeyPatch, ) -> None: clock = [100.0] handle = BLEDevice("LIVE-UUID", "XGR-K1", details=object()) monkeypatch.setattr(scanner_module, "monotonic", lambda: clock[0]) scanner_module._runtime_handles[handle.address] = handle # noqa: SLF001 scanner_module._runtime_handle_observed_at_monotonic = clock[0] # noqa: SLF001 clock[0] += scanner_module.BLE_DISCOVERY_CANDIDATE_LEASE_TTL_SECONDS + 0.001 assert discovered_device(handle.address) is handle clock[0] = 100.0 + scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS + 0.001 assert discovered_device(handle.address) is handle def test_connected_handle_is_session_scoped_and_survives_wall_clock_age( monkeypatch: MonkeyPatch, ) -> None: monotonic_clock = [100.0] suspend_aware_clock = [1_000.0] device = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) advertisement = AdvertisementData( local_name="XGR-LIVE", manufacturer_data={}, service_data={}, service_uuids=[], tx_power=0, rssi=-41, platform_data=(), ) discoveries = [ {device.address: (device, advertisement)}, {}, ] async def fake_discover(**_kwargs: object) -> dict[str, tuple[BLEDevice, AdvertisementData]]: return discoveries.pop(0) monkeypatch.setattr(scanner_module.BleakScanner, "discover", fake_discover) monkeypatch.setattr(scanner_module, "monotonic", lambda: monotonic_clock[0]) monkeypatch.setattr( scanner_module, "suspend_aware_time", lambda: suspend_aware_clock[0], ) async def scenario() -> None: await scan(1.0) captured = capture_discovered_device(device.address) assert captured is not None pin_connected_device_handle(captured, device_session_id="device-session-a") await scan(1.0) assert discovered_device_selection( device.address ) == scanner_module.DiscoveredDeviceSelection( device=None, from_fresh_scan=True, ) retained_capture = connected_device_capture( device.address, device_session_id="device-session-a", ) assert retained_capture is not None assert retained_capture.device is captured.device assert retained_capture.source == "retained-session" owner_epoch_for_loop = scanner_module.ble_runtime_owner_epoch_for_current_loop monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", lambda: None, ) assert discovered_device_selection( device.address ) == scanner_module.DiscoveredDeviceSelection( device=None, from_fresh_scan=False, ) monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", owner_epoch_for_loop, ) retained_capture = connected_device_capture( device.address, device_session_id="device-session-a", ) assert retained_capture is not None assert retained_capture.device is captured.device assert ( connected_device_capture( device.address, device_session_id="device-session-b", ) is None ) assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", ) == { "status": "retained", "scope": "owner-epoch-device-session", "advertised_now": False, "gatt_validated_recently": False, } monotonic_clock[0] += scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS + 0.001 suspend_aware_clock[0] += ( scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS + 0.001 ) assert connected_device_capture( device.address, device_session_id="device-session-a", ) is not None assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", ) == { "status": "retained", "scope": "owner-epoch-device-session", "advertised_now": False, "gatt_validated_recently": False, } asyncio.run(scenario()) def test_scan_and_retained_handle_survive_macos_sleep_until_explicit_invalidation( monkeypatch: MonkeyPatch, ) -> None: monotonic_clock = [100.0] suspend_aware_clock = [1_000.0] device = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) advertisement = AdvertisementData( local_name="XGR-LIVE", manufacturer_data={}, service_data={}, service_uuids=[], tx_power=0, rssi=-41, platform_data=(), ) async def fake_discover( **_kwargs: object, ) -> dict[str, tuple[BLEDevice, AdvertisementData]]: return {device.address: (device, advertisement)} monkeypatch.setattr(scanner_module.BleakScanner, "discover", fake_discover) monkeypatch.setattr(scanner_module, "monotonic", lambda: monotonic_clock[0]) monkeypatch.setattr( scanner_module, "suspend_aware_time", lambda: suspend_aware_clock[0], ) async def scenario() -> None: await scan(1.0) captured = capture_discovered_device(device.address) assert captured is not None assert mark_captured_device_gatt_validated(captured) is True pin_connected_device_handle(captured, device_session_id="device-session-a") assert discovered_device(device.address) is device retained_capture = connected_device_capture( device.address, device_session_id="device-session-a", ) assert retained_capture is not None assert retained_capture.device is captured.device assert retained_capture.source == "retained-session" assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", ) == { "status": "retained", "scope": "owner-epoch-device-session", "advertised_now": True, "gatt_validated_recently": True, } # macOS monotonic time may stop while the laptop sleeps. Wall time is # deliberately the only clock advanced here. GATT validation has the # shortest lease, so it is revoked before the scan/retained handles. suspend_aware_clock[0] += ( scanner_module.BLE_GATT_VALIDATION_RECENCY_TTL_SECONDS + 0.001 ) assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", ) == { "status": "retained", "scope": "owner-epoch-device-session", "advertised_now": True, "gatt_validated_recently": False, } suspend_aware_clock[0] += ( scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS - scanner_module.BLE_GATT_VALIDATION_RECENCY_TTL_SECONDS ) assert discovered_device(device.address) is device assert scanner_module.captured_device_handle(captured) is device assert connected_device_capture( device.address, device_session_id="device-session-a", ) is not None assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", ) == { "status": "retained", "scope": "owner-epoch-device-session", "advertised_now": True, "gatt_validated_recently": False, } asyncio.run(scenario()) def test_slow_gatt_handoff_refreshes_capture_and_retained_lease_together( monkeypatch: MonkeyPatch, ) -> None: monotonic_clock = [100.0] suspend_aware_clock = [1_000.0] monkeypatch.setattr(scanner_module, "monotonic", lambda: monotonic_clock[0]) monkeypatch.setattr( scanner_module, "suspend_aware_time", lambda: suspend_aware_clock[0], ) device = BLEDevice("SLOW-UUID", "XGR-SLOW", details=object()) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, scan_generation=7, captured_at_monotonic=monotonic_clock[0], captured_at_suspend_aware=suspend_aware_clock[0], ) # A real connect/baseline can outlive the original scan lease. Its # exact successful GATT result is the new freshness boundary. elapsed = scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS + 2.0 monotonic_clock[0] += elapsed suspend_aware_clock[0] += elapsed assert scanner_module.captured_device_handle(captured) is device assert mark_captured_device_gatt_validated(captured) is True pin_connected_device_handle(captured, device_session_id="device-session-a") retained = connected_device_capture( device.address, device_session_id="device-session-a", ) assert retained is not None assert retained.device is device assert retained.source == "retained-session" assert retained.captured_at_monotonic == monotonic_clock[0] assert retained.captured_at_suspend_aware == suspend_aware_clock[0] assert scanner_module.captured_device_handle(retained) is device asyncio.run(scenario()) def test_validated_expired_retained_capture_repins_to_new_session( monkeypatch: MonkeyPatch, ) -> None: monotonic_clock = [100.0] suspend_aware_clock = [1_000.0] monkeypatch.setattr(scanner_module, "monotonic", lambda: monotonic_clock[0]) monkeypatch.setattr( scanner_module, "suspend_aware_time", lambda: suspend_aware_clock[0], ) device = BLEDevice("ROTATE-UUID", "XGR-ROTATE", details=object()) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, scan_generation=11, captured_at_monotonic=monotonic_clock[0], captured_at_suspend_aware=suspend_aware_clock[0], ) pin_connected_device_handle(captured, device_session_id="old-session") elapsed = scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS + 2.0 monotonic_clock[0] += elapsed suspend_aware_clock[0] += elapsed assert connected_device_capture( device.address, device_session_id="old-session", ) is not None # The exact new GATT baseline, not the old scan timestamp, is the # admission boundary for rotating this same transport into a new # facade device session. assert mark_captured_device_gatt_validated(captured) is True pin_connected_device_handle(captured, device_session_id="new-session") assert connected_device_capture( device.address, device_session_id="old-session", ) is None repinned = connected_device_capture( device.address, device_session_id="new-session", ) assert repinned is not None assert repinned.device is device assert repinned.captured_at_monotonic == monotonic_clock[0] assert repinned.captured_at_suspend_aware == suspend_aware_clock[0] assert connected_device_recovery_snapshot( device.address, device_session_id="new-session", )["gatt_validated_recently"] is True asyncio.run(scenario()) @pytest.mark.parametrize("source", ["retrieved-session", "retrieved-durable"]) def test_retrieved_capture_cannot_be_pinned_before_exact_gatt_validation( source: str, ) -> None: async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() device = BLEDevice("UNVALIDATED-UUID", "XGR-UNVALIDATED", details=object()) now = scanner_module._freshness_now() # noqa: SLF001 captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, captured_at_monotonic=now.monotonic, captured_at_suspend_aware=now.suspend_aware, source=source, # type: ignore[arg-type] ) with pytest.raises(RuntimeError, match="requires exact GATT validation"): pin_connected_device_handle(captured, device_session_id="new-session") assert connected_device_recovery_snapshot( device.address, device_session_id="new-session", )["status"] == "unavailable" asyncio.run(scenario()) def test_gatt_handoff_token_retains_and_matches_exact_device_object() -> None: async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() validated_device = BLEDevice("HANDOFF-UUID", "XGR-HANDOFF", details=object()) different_device = BLEDevice("HANDOFF-UUID", "XGR-HANDOFF", details=object()) now = scanner_module._freshness_now() # noqa: SLF001 validated = scanner_module.CapturedDiscoveredDevice( device=validated_device, macos_uuid=validated_device.address, owner_epoch=owner_epoch, scan_generation=13, captured_at_monotonic=now.monotonic, captured_at_suspend_aware=now.suspend_aware, source="retrieved-durable", ) impostor = scanner_module.CapturedDiscoveredDevice( device=different_device, macos_uuid=different_device.address, owner_epoch=owner_epoch, scan_generation=13, captured_at_monotonic=now.monotonic, captured_at_suspend_aware=now.suspend_aware, source="retrieved-durable", ) assert mark_captured_device_gatt_validated(validated) is True with pytest.raises(RuntimeError, match="requires exact GATT validation"): pin_connected_device_handle(impostor, device_session_id="device-session-a") pin_connected_device_handle(validated, device_session_id="device-session-a") retained = connected_device_capture( validated_device.address, device_session_id="device-session-a", ) assert retained is not None assert retained.device is validated_device asyncio.run(scenario()) @pytest.mark.parametrize( "operation_kind", ["status-read", "wifi-provision", "ap-enable"], ) def test_retained_handle_survives_age_and_can_be_explicitly_retrieved( monkeypatch: MonkeyPatch, operation_kind: RecoveryOperationKind, ) -> None: monotonic_clock = [100.0] suspend_aware_clock = [1_000.0] monkeypatch.setattr(scanner_module, "monotonic", lambda: monotonic_clock[0]) monkeypatch.setattr( scanner_module, "suspend_aware_time", lambda: suspend_aware_clock[0], ) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() identifier = _FakeCoreBluetoothIdentifier("RETRIEVE-UUID") original_peripheral = _FakeCoreBluetoothPeripheral(identifier, "XGR-OLD") retrieved_peripheral = _FakeCoreBluetoothPeripheral(identifier, "XGR-NEW") central = _FakeCentralManager([[retrieved_peripheral]]) manager = _FakeCoreBluetoothManager(asyncio.get_running_loop(), central) original_device = BLEDevice( identifier.value, "XGR-OLD", details=(original_peripheral, manager), ) captured = scanner_module.CapturedDiscoveredDevice( device=original_device, macos_uuid=original_device.address, owner_epoch=owner_epoch, scan_generation=9, captured_at_monotonic=monotonic_clock[0], captured_at_suspend_aware=suspend_aware_clock[0], ) pin_connected_device_handle(captured, device_session_id="device-session-a") elapsed = scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS + 0.001 monotonic_clock[0] += elapsed suspend_aware_clock[0] += elapsed assert scanner_module.captured_device_handle(captured) is original_device assert connected_device_capture( original_device.address, device_session_id="device-session-a", ) is not None assert connected_device_recovery_snapshot( original_device.address, device_session_id="device-session-a", ) == { "status": "retained", "scope": "owner-epoch-device-session", "advertised_now": False, "gatt_validated_recently": False, } retrieved = await _retrieve_connected_inside_ble_lease( original_device.address, device_session_id="device-session-a", operation_kind=operation_kind, ) assert retrieved is not None assert retrieved.source == "retrieved-session" assert retrieved.device is not original_device assert retrieved.device.details == (retrieved_peripheral, manager) assert scanner_module.captured_device_handle(retrieved) is retrieved.device assert central.requests == [[identifier]] assert manager.ready_calls == 1 assert connected_device_recovery_snapshot( original_device.address, device_session_id="device-session-a", ) == { "status": "retained", "scope": "owner-epoch-device-session", "advertised_now": False, "gatt_validated_recently": False, } asyncio.run(scenario()) def test_gatt_failure_invalidates_session_token_until_new_explicit_scan() -> None: async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() identifier = _FakeCoreBluetoothIdentifier("RETRY-UUID") original_peripheral = _FakeCoreBluetoothPeripheral(identifier, "XGR-OLD") retrieved_peripheral = _FakeCoreBluetoothPeripheral(identifier, "XGR-NEW") central = _FakeCentralManager([[], [retrieved_peripheral]]) manager = _FakeCoreBluetoothManager(asyncio.get_running_loop(), central) original_device = BLEDevice( identifier.value, "XGR-OLD", details=(original_peripheral, manager), ) captured = scanner_module.CapturedDiscoveredDevice( device=original_device, macos_uuid=original_device.address, owner_epoch=owner_epoch, scan_generation=3, ) pin_connected_device_handle(captured, device_session_id="device-session-a") assert demote_connected_device_handle_after_gatt_failure(captured) is True assert ( await _retrieve_connected_inside_ble_lease( original_device.address, device_session_id="device-session-a", ) is None ) assert connected_device_recovery_snapshot( original_device.address, device_session_id="device-session-a", )["status"] == "unavailable" assert central.requests == [] asyncio.run(scenario()) def test_retrieval_rejects_uuid_session_and_owner_mismatch( monkeypatch: MonkeyPatch, ) -> None: async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() identifier = _FakeCoreBluetoothIdentifier("EXACT-UUID") peripheral = _FakeCoreBluetoothPeripheral(identifier, "XGR-EXACT") central = _FakeCentralManager([[peripheral]]) manager = _FakeCoreBluetoothManager(asyncio.get_running_loop(), central) device = BLEDevice( identifier.value, "XGR-EXACT", details=(peripheral, manager), ) captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, ) pin_connected_device_handle(captured, device_session_id="device-session-a") assert ( await _retrieve_connected_inside_ble_lease( "OTHER-UUID", device_session_id="device-session-a", ) is None ) assert ( await _retrieve_connected_inside_ble_lease( device.address, device_session_id="device-session-b", ) is None ) owner_epoch_lookup = scanner_module.ble_runtime_owner_epoch_for_current_loop monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", lambda: owner_epoch + 1, ) assert ( await _retrieve_connected_inside_ble_lease( device.address, device_session_id="device-session-a", ) is None ) monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", owner_epoch_lookup, ) assert central.requests == [] assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", )["status"] == "retained" asyncio.run(scenario()) def test_connected_retrieval_requires_active_owned_ble_lease() -> None: async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() identifier = _FakeCoreBluetoothIdentifier("LEASE-UUID") peripheral = _FakeCoreBluetoothPeripheral(identifier, "XGR-LEASE") central = _FakeCentralManager([[peripheral]]) manager = _FakeCoreBluetoothManager(asyncio.get_running_loop(), central) device = BLEDevice( identifier.value, "XGR-LEASE", details=(peripheral, manager), ) captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, ) pin_connected_device_handle(captured, device_session_id="device-session-a") assert ( await retrieve_connected_device_capture( device.address, device_session_id="device-session-a", ) is None ) assert central.requests == [] assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", )["status"] == "retained" asyncio.run(scenario()) def test_connected_retrieval_rechecks_same_operation_kind_after_native_await( monkeypatch: MonkeyPatch, ) -> None: async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() identifier = _FakeCoreBluetoothIdentifier("KIND-UUID") peripheral = _FakeCoreBluetoothPeripheral(identifier, "XGR-KIND") central = _FakeCentralManager([[peripheral]]) manager = _FakeCoreBluetoothManager(asyncio.get_running_loop(), central) device = BLEDevice(identifier.value, "XGR-KIND", details=(peripheral, manager)) captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, ) pin_connected_device_handle(captured, device_session_id="device-session-a") runtime = { "owner_epoch": owner_epoch, "owner_loop_bound": True, "active_operation_kind": "status-read", "cleanup_pending": False, "poisoned": False, } monkeypatch.setattr(scanner_module, "ble_runtime_snapshot", lambda: dict(runtime)) async def change_active_operation() -> None: manager.ready_calls += 1 runtime["active_operation_kind"] = "wifi-provision" manager.wait_until_ready = change_active_operation # type: ignore[method-assign] assert ( await retrieve_connected_device_capture( device.address, device_session_id="device-session-a", ) is None ) assert central.requests == [[identifier]] # A changed lease never consumes or replaces the exact process token. runtime["active_operation_kind"] = None assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", )["status"] == "retained" asyncio.run(scenario()) def test_connected_device_recovery_name_is_exact_but_not_presence( monkeypatch: MonkeyPatch, ) -> None: monotonic_clock = [100.0] suspend_aware_clock = [1_000.0] monkeypatch.setattr(scanner_module, "monotonic", lambda: monotonic_clock[0]) monkeypatch.setattr( scanner_module, "suspend_aware_time", lambda: suspend_aware_clock[0], ) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() device = BLEDevice("NAME-UUID", " XGR-EXACT ", details=object()) captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, captured_at_monotonic=monotonic_clock[0], captured_at_suspend_aware=suspend_aware_clock[0], ) pin_connected_device_handle(captured, device_session_id="device-session-a") elapsed = scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS + 1.0 monotonic_clock[0] += elapsed suspend_aware_clock[0] += elapsed assert connected_device_capture( device.address, device_session_id="device-session-a", ) is not None assert connected_device_recovery_name( device.address, device_session_id="device-session-a", ) == "XGR-EXACT" assert connected_device_recovery_name( device.address, device_session_id="other-session", ) is None snapshot = connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", ) assert snapshot["status"] == "retained" assert snapshot["advertised_now"] is False assert snapshot["gatt_validated_recently"] is False asyncio.run(scenario()) def test_durable_uuid_retrieval_is_available_only_inside_explicit_status_read( monkeypatch: MonkeyPatch, ) -> None: macos_uuid = "11111111-2222-4333-8444-555555555555" async def scenario() -> None: identifier = _FakeCoreBluetoothIdentifier(macos_uuid) peripheral = _FakeCoreBluetoothPeripheral(identifier, "XGR-DURABLE") central = _FakeCentralManager([[peripheral]]) manager = _FakeCoreBluetoothManager(asyncio.get_running_loop(), central) monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", lambda: 7, ) runtime = _active_status_read_runtime() monkeypatch.setattr(scanner_module, "ble_runtime_snapshot", lambda: runtime) monkeypatch.setattr( scanner_module, "_new_corebluetooth_retrieval_context", lambda requested: (manager, identifier) if requested == macos_uuid else None, ) captured = await retrieve_known_device_capture_for_status_read(macos_uuid) assert captured is not None assert captured.macos_uuid == macos_uuid assert captured.source == "retrieved-durable" assert captured.owner_epoch == 7 assert captured.device.details == (peripheral, manager) assert central.requests == [[identifier]] assert manager.ready_calls == 1 # Retrieval is only an opaque transport object. It is neither added to # discovery nor pinned as a current device session by this helper. assert discovered_device(macos_uuid) is None assert connected_device_recovery_snapshot( macos_uuid, device_session_id="any-session", )["status"] == "unavailable" asyncio.run(scenario()) def test_durable_uuid_retrieval_rejects_non_status_operation_before_native_call( monkeypatch: MonkeyPatch, ) -> None: calls: list[str] = [] monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", lambda: 7, ) monkeypatch.setattr( scanner_module, "ble_runtime_snapshot", lambda: { **_active_status_read_runtime(), "active_operation_kind": "wifi-provision", }, ) monkeypatch.setattr( scanner_module, "_new_corebluetooth_retrieval_context", lambda requested: calls.append(requested), ) assert ( asyncio.run( retrieve_known_device_capture_for_status_read( "11111111-2222-4333-8444-555555555555" ) ) is None ) assert calls == [] def test_exact_uuid_advertisement_scan_captures_without_publishing_discovery( monkeypatch: MonkeyPatch, ) -> None: macos_uuid = "AAAAAAAA-BBBB-4CCC-8DDD-EEEEEEEEEEEE" native_details = object() candidate = BLEDevice( macos_uuid.lower(), "XGR-ADVERTISEMENT", details=native_details, ) calls: list[tuple[str, float]] = [] async def find_exact(address: str, *, timeout: float) -> BLEDevice: calls.append((address, timeout)) return candidate monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", lambda: 7, ) monkeypatch.setattr( scanner_module, "ble_runtime_snapshot", lambda: _active_status_read_runtime(), ) monkeypatch.setattr( scanner_module.BleakScanner, "find_device_by_address", find_exact, ) captured = asyncio.run( discover_known_device_capture_for_status_read( macos_uuid, timeout_seconds=30.0, ) ) assert captured is not None assert captured.macos_uuid == macos_uuid assert captured.device.address == macos_uuid assert captured.device.details is native_details assert captured.source == "retrieved-durable" assert calls == [(macos_uuid, 30.0)] assert discovered_device(macos_uuid) is None assert connected_device_recovery_snapshot( macos_uuid, device_session_id="any-session", )["status"] == "unavailable" def test_exact_uuid_advertisement_scan_rejects_wrong_uuid( monkeypatch: MonkeyPatch, ) -> None: macos_uuid = "AAAAAAAA-BBBB-4CCC-8DDD-EEEEEEEEEEEE" wrong = BLEDevice( "11111111-2222-4333-8444-555555555555", "XGR-WRONG", details=object(), ) async def return_wrong(_address: str, *, timeout: float) -> BLEDevice: assert timeout == 30.0 return wrong monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", lambda: 7, ) monkeypatch.setattr( scanner_module, "ble_runtime_snapshot", lambda: _active_status_read_runtime(), ) monkeypatch.setattr( scanner_module.BleakScanner, "find_device_by_address", return_wrong, ) assert ( asyncio.run( discover_known_device_capture_for_status_read( macos_uuid, timeout_seconds=30.0, ) ) is None ) assert discovered_device(macos_uuid) is None def test_exact_uuid_advertisement_scan_rejects_timeout_and_owner_change( monkeypatch: MonkeyPatch, ) -> None: macos_uuid = "AAAAAAAA-BBBB-4CCC-8DDD-EEEEEEEEEEEE" candidate = BLEDevice(macos_uuid, "XGR-LATE", details=object()) runtime_snapshots = [ _active_status_read_runtime(), _active_status_read_runtime(owner_epoch=8), ] responses: list[BLEDevice | None] = [None, candidate] async def find_exact(_address: str, *, timeout: float) -> BLEDevice | None: assert timeout == 30.0 return responses.pop(0) monkeypatch.setattr( scanner_module, "ble_runtime_owner_epoch_for_current_loop", lambda: 7, ) monkeypatch.setattr( scanner_module.BleakScanner, "find_device_by_address", find_exact, ) monkeypatch.setattr( scanner_module, "ble_runtime_snapshot", lambda: _active_status_read_runtime(), ) assert ( asyncio.run( discover_known_device_capture_for_status_read( macos_uuid, timeout_seconds=30.0, ) ) is None ) monkeypatch.setattr( scanner_module, "ble_runtime_snapshot", lambda: runtime_snapshots.pop(0), ) assert ( asyncio.run( discover_known_device_capture_for_status_read( macos_uuid, timeout_seconds=30.0, ) ) is None ) assert responses == [] assert discovered_device(macos_uuid) is None def test_pending_gatt_handoff_expires_across_macos_sleep( monkeypatch: MonkeyPatch, ) -> None: monotonic_clock = [100.0] suspend_aware_clock = [1_000.0] monkeypatch.setattr(scanner_module, "monotonic", lambda: monotonic_clock[0]) monkeypatch.setattr( scanner_module, "suspend_aware_time", lambda: suspend_aware_clock[0], ) device = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, scan_generation=7, captured_at_monotonic=monotonic_clock[0], captured_at_suspend_aware=suspend_aware_clock[0], ) assert mark_captured_device_gatt_validated(captured) is True suspend_aware_clock[0] += ( scanner_module.BLE_GATT_VALIDATION_HANDOFF_TTL_SECONDS + 0.001 ) pin_connected_device_handle(captured, device_session_id="device-session-a") snapshot = connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", ) assert snapshot["status"] == "retained" assert snapshot["gatt_validated_recently"] is False asyncio.run(scenario()) def test_later_fresh_scan_does_not_replace_retained_corebluetooth_object( monkeypatch: MonkeyPatch, ) -> None: first = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) second = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) advertisement = AdvertisementData( local_name="XGR-LIVE", manufacturer_data={}, service_data={}, service_uuids=[], tx_power=0, rssi=-38, platform_data=(), ) discoveries = [ {first.address: (first, advertisement)}, {second.address: (second, advertisement)}, ] async def fake_discover(**_kwargs: object) -> dict[str, tuple[BLEDevice, AdvertisementData]]: return discoveries.pop(0) monkeypatch.setattr(scanner_module.BleakScanner, "discover", fake_discover) async def scenario() -> None: await scan(1.0) captured = capture_discovered_device(first.address) assert captured is not None assert captured.device is first pin_connected_device_handle(captured, device_session_id="device-session-a") await scan(1.0) retained = connected_device_capture( first.address, device_session_id="device-session-a", ) assert retained is not None assert retained is not captured assert retained.device is first assert retained.scan_generation == captured.scan_generation next_capture = capture_discovered_device(first.address) assert next_capture is not None assert next_capture.device is second assert next_capture.scan_generation > captured.scan_generation assert connected_device_recovery_snapshot( first.address, device_session_id="device-session-a", ) == { "status": "retained", "scope": "owner-epoch-device-session", "advertised_now": True, "gatt_validated_recently": False, } asyncio.run(scenario()) def test_exact_gatt_validation_handoff_marks_newly_pinned_recovery_token() -> None: device = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() captured = scanner_module.CapturedDiscoveredDevice( device=device, macos_uuid=device.address, owner_epoch=owner_epoch, scan_generation=7, captured_at_monotonic=100.0, ) # The reviewed baseline read completes immediately before the facade # creates its new device-session. Validation follows the exact object # across only that bounded synchronous handoff. assert mark_captured_device_gatt_validated(captured) is True pin_connected_device_handle(captured, device_session_id="device-session-a") assert connected_device_recovery_snapshot( device.address, device_session_id="device-session-a", )["gatt_validated_recently"] is True asyncio.run(scenario()) def test_failed_gatt_demotes_only_the_exact_retained_corebluetooth_object() -> None: first = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) replacement = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() stale_capture = scanner_module.CapturedDiscoveredDevice( device=first, macos_uuid=first.address, owner_epoch=owner_epoch, scan_generation=7, ) fresh_capture = scanner_module.CapturedDiscoveredDevice( device=replacement, macos_uuid=replacement.address, owner_epoch=owner_epoch, scan_generation=8, ) pin_connected_device_handle( stale_capture, device_session_id="device-session-a", ) assert demote_connected_device_handle_after_gatt_failure(stale_capture) is True assert ( connected_device_capture( first.address, device_session_id="device-session-a", ) is None ) # A late failure from the old generation must not revoke a replacement # object already admitted for the same CoreBluetooth UUID/session. pin_connected_device_handle( fresh_capture, device_session_id="device-session-a", ) assert demote_connected_device_handle_after_gatt_failure(stale_capture) is False retained_replacement = connected_device_capture( replacement.address, device_session_id="device-session-a", ) assert retained_replacement is not None assert retained_replacement.device is replacement assert retained_replacement.scan_generation == fresh_capture.scan_generation assert retained_replacement.source == "retained-session" asyncio.run(scenario()) def test_failed_gatt_invalidates_unpinned_selection_until_new_scan_generation() -> None: first = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) replacement = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() failed = scanner_module.CapturedDiscoveredDevice( device=first, macos_uuid=first.address, owner_epoch=owner_epoch, scan_generation=7, ) assert demote_connected_device_handle_after_gatt_failure(failed) is False assert scanner_module.captured_device_handle(failed) is None next_scan = scanner_module.CapturedDiscoveredDevice( device=replacement, macos_uuid=replacement.address, owner_epoch=owner_epoch, scan_generation=8, ) assert scanner_module.captured_device_handle(next_scan) is replacement asyncio.run(scenario()) def test_invalidated_visible_row_requires_a_new_exact_scan_object() -> None: first = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) replacement = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() first_generation = scanner_module._begin_scan_generation(owner_epoch) # noqa: SLF001 scanner_module._publish_scan_handles( # noqa: SLF001 first_generation, owner_epoch, {first.address: first}, ) first_capture = capture_discovered_device(first.address) assert first_capture is not None assert first_capture.device is first assert mark_captured_device_gatt_validated(first_capture) is True pin_connected_device_handle( first_capture, device_session_id="verify-provisional-session", ) # A losing read-only Verify retires its provisional device session but # may leave the facade's row visible for product continuity. The native # object behind that same generation must nevertheless stay revoked. assert scanner_module.invalidate_connected_device_session( first.address, device_session_id="verify-provisional-session", ) is True invalidated_selection = discovered_device_selection(first.address) assert invalidated_selection.from_fresh_scan is True assert invalidated_selection.device is None assert capture_discovered_device(first.address) is None assert scanner_module.captured_device_handle(first_capture) is None second_generation = scanner_module._begin_scan_generation(owner_epoch) # noqa: SLF001 scanner_module._publish_scan_handles( # noqa: SLF001 second_generation, owner_epoch, {replacement.address: replacement}, ) replacement_capture = capture_discovered_device(replacement.address) assert replacement_capture is not None assert replacement_capture.device is replacement assert replacement_capture.scan_generation == second_generation assert scanner_module.captured_device_handle(replacement_capture) is replacement # Neither the old object nor a capture manufactured for another owner # epoch gains authority from the new scan. assert scanner_module.captured_device_handle(first_capture) is None foreign_capture = scanner_module.CapturedDiscoveredDevice( device=replacement, macos_uuid=replacement.address, owner_epoch=owner_epoch + 1, scan_generation=second_generation, ) assert scanner_module.captured_device_handle(foreign_capture) is None asyncio.run(scenario()) def test_exact_session_invalidation_does_not_clear_new_scan_handle() -> None: retained = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) newly_scanned = BLEDevice("LIVE-UUID", "XGR-LIVE", details=object()) async def scenario() -> None: owner_epoch = bind_ble_runtime_owner_loop() captured = scanner_module.CapturedDiscoveredDevice( device=retained, macos_uuid=retained.address, owner_epoch=owner_epoch, scan_generation=7, ) assert mark_captured_device_gatt_validated(captured) is True pin_connected_device_handle(captured, device_session_id="device-session-a") next_generation = scanner_module._begin_scan_generation(owner_epoch) # noqa: SLF001 scanner_module._publish_scan_handles( # noqa: SLF001 next_generation, owner_epoch, {newly_scanned.address: newly_scanned}, ) replacement_capture = scanner_module.capture_discovered_device(newly_scanned.address) assert replacement_capture is not None assert replacement_capture.device is newly_scanned assert scanner_module.invalidate_connected_device_session( retained.address, device_session_id="device-session-a", ) is True assert connected_device_capture( retained.address, device_session_id="device-session-a", ) is None assert scanner_module.captured_device_handle(captured) is None with pytest.raises(RuntimeError, match="invalidated by session teardown"): pin_connected_device_handle(captured, device_session_id="stale-repin") with scanner_module._runtime_handle_lock: # noqa: SLF001 assert ( # noqa: SLF001 scanner_module._runtime_handles[newly_scanned.address] is newly_scanned ) assert scanner_module.captured_device_handle(replacement_capture) is newly_scanned assert mark_captured_device_gatt_validated(replacement_capture) is True pin_connected_device_handle( replacement_capture, device_session_id="device-session-b", ) assert connected_device_capture( newly_scanned.address, device_session_id="device-session-b", ) is not None asyncio.run(scenario())