import asyncio from collections.abc import Callable, 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"] def install_advertisement_source( monkeypatch: MonkeyPatch, observations: list[tuple[float, BLEDevice, str | None]], ) -> dict[str, int]: """A single native scanner that delivers advertisements while it is open.""" lifecycle = {"started": 0, "stopped": 0} class FakeScanner: def __init__( self, detection_callback: Callable[[BLEDevice, AdvertisementData], None] ) -> None: self.callback = detection_callback self.scheduled: list[asyncio.TimerHandle] = [] async def __aenter__(self) -> "FakeScanner": lifecycle["started"] += 1 for delay, device, name in observations: advertisement = AdvertisementData( local_name=name, manufacturer_data={}, service_data={}, service_uuids=[], tx_power=None, rssi=-45, platform_data=(), ) self.scheduled.append(asyncio.get_running_loop().call_later( delay, self.callback, device, advertisement )) return self async def __aexit__(self, *_args: object) -> None: lifecycle["stopped"] += 1 for timer in self.scheduled: timer.cancel() monkeypatch.setattr(scanner_module, "BleakScanner", FakeScanner) monkeypatch.setattr(scanner_module, "BLE_SCAN_INITIAL_WINDOW_SECONDS", 0.02) return lifecycle def test_discovery_keeps_one_scanner_open_for_a_late_k1_name(monkeypatch: MonkeyPatch) -> None: device = BLEDevice("LATE-K1", None, details=object()) lifecycle = install_advertisement_source(monkeypatch, [ (0, device, None), (0.05, device, "XGR-LATE"), ]) async def scenario() -> None: result = await scan(0.4) timing = result["discovery_timing"] assert timing["scan_extended"] is True assert timing["first_candidate_ms"] >= timing["initial_window_ms"] assert timing["scan_elapsed_ms"] < 400 assert len(result["devices"]) == 1 assert result["devices"][0]["k1_name_candidate"] is True assert discovered_device(device.address) is device asyncio.run(scenario()) assert lifecycle == {"started": 1, "stopped": 1} def test_discovery_finishes_after_initial_window_when_k1_is_visible( monkeypatch: MonkeyPatch, ) -> None: device = BLEDevice("EARLY-K1", "XGR-EARLY", details=object()) lifecycle = install_advertisement_source(monkeypatch, [(0, device, "XGR-EARLY")]) result = asyncio.run(scan(0.4)) timing = result["discovery_timing"] assert timing["scan_extended"] is False assert timing["first_candidate_ms"] < timing["initial_window_ms"] assert 20 <= timing["scan_elapsed_ms"] < 400 assert lifecycle == {"started": 1, "stopped": 1} def test_discovery_without_k1_stops_at_deadline_and_does_not_reuse_old_results( monkeypatch: MonkeyPatch, ) -> None: device = BLEDevice("OTHER-DEVICE", "Headphones", details=object()) lifecycle = install_advertisement_source(monkeypatch, [(0, device, "Headphones")]) async def scenario() -> None: result = await scan(0.06) timing = result["discovery_timing"] assert timing["scan_extended"] is True assert "first_candidate_ms" not in timing assert timing["scan_elapsed_ms"] >= 60 assert result["devices"][0]["k1_name_candidate"] is False install_advertisement_source(monkeypatch, []) empty = await scan(0.01) assert empty["devices"] == [] assert empty["discovery_timing"]["scan_extended"] is False assert discovered_device(device.address) is None asyncio.run(scenario()) assert lifecycle == {"started": 1, "stopped": 1} def test_cancelled_discovery_closes_scanner_without_publishing_partial_handles( monkeypatch: MonkeyPatch, ) -> None: device = BLEDevice("PARTIAL", "Headphones", details=object()) lifecycle = install_advertisement_source(monkeypatch, [(0, device, "Headphones")]) async def scenario() -> None: task = asyncio.create_task(scan(1)) await asyncio.sleep(0.04) assert lifecycle["started"] == 1 task.cancel() with pytest.raises(asyncio.CancelledError): await task assert discovered_device(device.address) is None asyncio.run(scenario()) assert lifecycle == {"started": 1, "stopped": 1} 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._discover_k1_advertisements", 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, "_discover_k1_advertisements", 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, "_discover_k1_advertisements", 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, "_discover_k1_advertisements", 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, "_discover_k1_advertisements", 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, "_discover_k1_advertisements", 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, "_discover_k1_advertisements", 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()) @pytest.mark.parametrize("platform,present", [("linux", True), ("linux", False), ("darwin", False)]) def test_status_read_native_cache_check_is_linux_only(monkeypatch, platform, present): from types import SimpleNamespace calls = [] device = BLEDevice("AA:BB:CC:DD:EE:FF", "synthetic", {"path": "/synthetic/bluez"}) async def retrieve(address, details): calls.append((address, details)) return device if present else None monkeypatch.setattr(scanner_module, "sys", SimpleNamespace(platform=platform)) monkeypatch.setattr(scanner_module, "_retrieve_bluez_device", retrieve) assert asyncio.run(scanner_module.status_read_device_is_current(device)) is ( present or platform == "darwin") assert len(calls) == (1 if platform == "linux" else 0)