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
|
import asyncio
from bumble.att import Attribute, AttributeValue
from bumble.device import Connection, Device
from bumble.gatt import (
GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
Characteristic,
Descriptor,
Service,
)
from bleak import BleakClient, BleakScanner
from bleak.backends.characteristic import BleakGATTCharacteristic
from tests.integration.conftest import add_default_advertising_data
TEST_SERVICE_UUID = "9d513f40-5c89-42dc-9688-2cfa30f2d9e7"
TEST_CHARACTERISTIC_UUID = "e809cb2f-34e3-42a1-ba92-22db2495cd6a"
async def test_notification_sent_before_write_response(
bumble_peripheral: Device,
) -> None:
"""
Regression test for <https://github.com/hbldh/bleak/issues/1885>.
"""
notifications_enabled = False
def on_cccd_read(connection: Connection) -> bytes:
return b"\x01\x00" if notifications_enabled else b"\x00\x00"
async def on_cccd_write(connection: Connection, value: bytes) -> None:
nonlocal notifications_enabled
notifications_enabled = value == b"\x01\x00"
# This is simulating an unusual peripheral that sends a notification
# immediately upon receiving the CCCD write, before sending the write
# response.
# TODO: Type hints in bumble need to be fixed to be able to remove the pyright ignore
await bumble_peripheral.notify_subscribers( # pyright: ignore[reportUnknownMemberType]
test_characteristic, b"test", force=True
)
cccd_value = AttributeValue[bytes](on_cccd_read, on_cccd_write)
test_characteristic = Characteristic[bytes](
TEST_CHARACTERISTIC_UUID,
Characteristic.Properties.NOTIFY,
Attribute.Permissions(0),
descriptors=[
Descriptor(
GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
Attribute.Permissions.WRITEABLE | Attribute.Permissions.READABLE,
cccd_value,
)
],
)
bumble_peripheral.add_service(Service(TEST_SERVICE_UUID, [test_characteristic]))
add_default_advertising_data(bumble_peripheral)
await bumble_peripheral.power_on()
await bumble_peripheral.start_advertising()
device = await BleakScanner.find_device_by_address(
bumble_peripheral.static_address.to_string(), cb={"use_bdaddr": True}
)
assert device is not None, "Could not find bumble peripheral device"
async with BleakClient(device, services=[TEST_SERVICE_UUID]) as client:
notification_queue: asyncio.Queue[bytes] = asyncio.Queue()
def on_notification(_: BleakGATTCharacteristic, data: bytearray) -> None:
notification_queue.put_nowait(bytes(data))
await client.start_notify(
TEST_CHARACTERISTIC_UUID, on_notification, bluez={"use_start_notify": True}
)
# In BlueZ, the notification is not received when using "AcquireNotify"
# causing this to timeout.
data = await asyncio.wait_for(notification_queue.get(), timeout=3)
assert data == b"test"
|
