1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
import sys
from typing import AsyncGenerator
import pytest
from bumble import data_types
from bumble.core import AdvertisingData, DataType
from bumble.device import Device, DeviceConfiguration
from bumble.gatt import Service
from bumble.hci import Address
from bumble.transport import open_transport
from bumble.transport.common import Transport
from bleak import BleakScanner
from bleak.backends.device import BLEDevice
@pytest.fixture
async def hci_transport(
request: pytest.FixtureRequest,
) -> AsyncGenerator[Transport, None]:
"""Create a bumble HCI Transport."""
hci_transport_name: str | None = request.config.getoption("--bleak-hci-transport")
bluez_vhci_enabled: bool = request.config.getoption("--bleak-bluez-vhci")
if hci_transport_name is not None and bluez_vhci_enabled:
raise pytest.UsageError(
"Cannot use --bleak-hci-transport and --bleak-bluez-vhci together"
)
elif bluez_vhci_enabled:
if sys.platform != "linux":
pytest.skip("--bleak-bluez-vhci is only supported on Linux")
from tests.integration.bluez_controller import open_transport_with_bluez_vhci
async with open_transport_with_bluez_vhci() as hci_transport:
yield hci_transport
elif hci_transport_name is not None:
async with await open_transport(hci_transport_name) as hci_transport:
yield hci_transport
else:
pytest.skip(
"No HCI transport provided (use --bleak-hci-transport or --bleak-bluez-vhci)"
)
@pytest.fixture
def bumble_peripheral(hci_transport: Transport) -> Device:
"""
Create a BLE peripheral device with bumble.
"""
config = DeviceConfiguration(
name="Bleak",
# use random static address to avoid device caching issues, when characteristics change between test runs
address=Address.generate_static_address(),
advertising_interval_min=200,
advertising_interval_max=200,
)
return Device.from_config_with_hci(
config,
hci_transport.source,
hci_transport.sink,
)
def add_default_advertising_data(
bumble_peripheral: Device,
additional_adv_data: list[DataType] | None = None,
) -> None:
"""Add default advertising data to bumble peripheral."""
adv_data: list[DataType] = [
data_types.Flags(
AdvertisingData.Flags.LE_GENERAL_DISCOVERABLE_MODE
| AdvertisingData.Flags.BR_EDR_NOT_SUPPORTED
),
data_types.CompleteLocalName(bumble_peripheral.name),
]
if additional_adv_data:
adv_data.extend(additional_adv_data)
bumble_peripheral.advertising_data = bytes(AdvertisingData(adv_data))
async def configure_and_power_on_bumble_peripheral(
bumble_peripheral: Device,
additional_adv_data: list[DataType] | None = None,
services: list[Service] | None = None,
) -> None:
"""Configure and power on the bumble peripheral."""
add_default_advertising_data(bumble_peripheral, additional_adv_data)
if services:
bumble_peripheral.add_services(services)
await bumble_peripheral.power_on()
await bumble_peripheral.start_advertising()
async def find_ble_device(bumble_peripheral: Device) -> BLEDevice:
"""Find the BLE device corresponding to the bumble peripheral."""
device = await BleakScanner.find_device_by_name(bumble_peripheral.name)
if device is None:
raise RuntimeError("failed to discover device, is Bumble working?")
return device
|
