format with Black
diff --git a/tests/device_test.py b/tests/device_test.py
index dd23c88..123df29 100644
--- a/tests/device_test.py
+++ b/tests/device_test.py
@@ -25,10 +25,23 @@
from bumble.device import Connection, Device
from bumble.host import Host
from bumble.hci import (
- HCI_ACCEPT_CONNECTION_REQUEST_COMMAND, HCI_COMMAND_STATUS_PENDING, HCI_CREATE_CONNECTION_COMMAND, HCI_SUCCESS,
- Address, HCI_Command_Complete_Event, HCI_Command_Status_Event, HCI_Connection_Complete_Event, HCI_Connection_Request_Event, HCI_Packet
+ HCI_ACCEPT_CONNECTION_REQUEST_COMMAND,
+ HCI_COMMAND_STATUS_PENDING,
+ HCI_CREATE_CONNECTION_COMMAND,
+ HCI_SUCCESS,
+ Address,
+ HCI_Command_Complete_Event,
+ HCI_Command_Status_Event,
+ HCI_Connection_Complete_Event,
+ HCI_Connection_Request_Event,
+ HCI_Packet,
)
-from bumble.gatt import GATT_GENERIC_ACCESS_SERVICE, GATT_CHARACTERISTIC_ATTRIBUTE_TYPE, GATT_DEVICE_NAME_CHARACTERISTIC, GATT_APPEARANCE_CHARACTERISTIC
+from bumble.gatt import (
+ GATT_GENERIC_ACCESS_SERVICE,
+ GATT_CHARACTERISTIC_ATTRIBUTE_TYPE,
+ GATT_DEVICE_NAME_CHARACTERISTIC,
+ GATT_APPEARANCE_CHARACTERISTIC,
+)
# -----------------------------------------------------------------------------
# Logging
@@ -59,70 +72,90 @@
d2.classic_enabled = True
# set public addresses
- d0.public_address = Address('F0:F1:F2:F3:F4:F5', address_type=Address.PUBLIC_DEVICE_ADDRESS)
- d1.public_address = Address('F5:F4:F3:F2:F1:F0', address_type=Address.PUBLIC_DEVICE_ADDRESS)
- d2.public_address = Address('F5:F4:F3:F3:F4:F5', address_type=Address.PUBLIC_DEVICE_ADDRESS)
+ d0.public_address = Address(
+ 'F0:F1:F2:F3:F4:F5', address_type=Address.PUBLIC_DEVICE_ADDRESS
+ )
+ d1.public_address = Address(
+ 'F5:F4:F3:F2:F1:F0', address_type=Address.PUBLIC_DEVICE_ADDRESS
+ )
+ d2.public_address = Address(
+ 'F5:F4:F3:F3:F4:F5', address_type=Address.PUBLIC_DEVICE_ADDRESS
+ )
def d0_flow():
packet = HCI_Packet.from_bytes((yield))
assert packet.name == 'HCI_CREATE_CONNECTION_COMMAND'
assert packet.bd_addr == d1.public_address
- d0.host.on_hci_packet(HCI_Command_Status_Event(
- status = HCI_COMMAND_STATUS_PENDING,
- num_hci_command_packets = 1,
- command_opcode = HCI_CREATE_CONNECTION_COMMAND
- ))
+ d0.host.on_hci_packet(
+ HCI_Command_Status_Event(
+ status=HCI_COMMAND_STATUS_PENDING,
+ num_hci_command_packets=1,
+ command_opcode=HCI_CREATE_CONNECTION_COMMAND,
+ )
+ )
- d1.host.on_hci_packet(HCI_Connection_Request_Event(
- bd_addr = d0.public_address,
- class_of_device = 0,
- link_type = HCI_Connection_Complete_Event.ACL_LINK_TYPE
- ))
+ d1.host.on_hci_packet(
+ HCI_Connection_Request_Event(
+ bd_addr=d0.public_address,
+ class_of_device=0,
+ link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
+ )
+ )
packet = HCI_Packet.from_bytes((yield))
assert packet.name == 'HCI_CREATE_CONNECTION_COMMAND'
assert packet.bd_addr == d2.public_address
- d0.host.on_hci_packet(HCI_Command_Status_Event(
- status = HCI_COMMAND_STATUS_PENDING,
- num_hci_command_packets = 1,
- command_opcode = HCI_CREATE_CONNECTION_COMMAND
- ))
+ d0.host.on_hci_packet(
+ HCI_Command_Status_Event(
+ status=HCI_COMMAND_STATUS_PENDING,
+ num_hci_command_packets=1,
+ command_opcode=HCI_CREATE_CONNECTION_COMMAND,
+ )
+ )
- d2.host.on_hci_packet(HCI_Connection_Request_Event(
- bd_addr = d0.public_address,
- class_of_device = 0,
- link_type = HCI_Connection_Complete_Event.ACL_LINK_TYPE
- ))
+ d2.host.on_hci_packet(
+ HCI_Connection_Request_Event(
+ bd_addr=d0.public_address,
+ class_of_device=0,
+ link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
+ )
+ )
assert (yield) == None
-
+
def d1_flow():
packet = HCI_Packet.from_bytes((yield))
assert packet.name == 'HCI_ACCEPT_CONNECTION_REQUEST_COMMAND'
- d1.host.on_hci_packet(HCI_Command_Complete_Event(
- num_hci_command_packets = 1,
- command_opcode = HCI_ACCEPT_CONNECTION_REQUEST_COMMAND,
- return_parameters = b"\x00"
- ))
+ d1.host.on_hci_packet(
+ HCI_Command_Complete_Event(
+ num_hci_command_packets=1,
+ command_opcode=HCI_ACCEPT_CONNECTION_REQUEST_COMMAND,
+ return_parameters=b"\x00",
+ )
+ )
- d1.host.on_hci_packet(HCI_Connection_Complete_Event(
- status = HCI_SUCCESS,
- connection_handle = 0x100,
- bd_addr = d0.public_address,
- link_type = HCI_Connection_Complete_Event.ACL_LINK_TYPE,
- encryption_enabled = True,
- ))
+ d1.host.on_hci_packet(
+ HCI_Connection_Complete_Event(
+ status=HCI_SUCCESS,
+ connection_handle=0x100,
+ bd_addr=d0.public_address,
+ link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
+ encryption_enabled=True,
+ )
+ )
- d0.host.on_hci_packet(HCI_Connection_Complete_Event(
- status = HCI_SUCCESS,
- connection_handle = 0x100,
- bd_addr = d1.public_address,
- link_type = HCI_Connection_Complete_Event.ACL_LINK_TYPE,
- encryption_enabled = True,
- ))
+ d0.host.on_hci_packet(
+ HCI_Connection_Complete_Event(
+ status=HCI_SUCCESS,
+ connection_handle=0x100,
+ bd_addr=d1.public_address,
+ link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
+ encryption_enabled=True,
+ )
+ )
assert (yield) == None
@@ -130,27 +163,33 @@
packet = HCI_Packet.from_bytes((yield))
assert packet.name == 'HCI_ACCEPT_CONNECTION_REQUEST_COMMAND'
- d2.host.on_hci_packet(HCI_Command_Complete_Event(
- num_hci_command_packets = 1,
- command_opcode = HCI_ACCEPT_CONNECTION_REQUEST_COMMAND,
- return_parameters = b"\x00"
- ))
+ d2.host.on_hci_packet(
+ HCI_Command_Complete_Event(
+ num_hci_command_packets=1,
+ command_opcode=HCI_ACCEPT_CONNECTION_REQUEST_COMMAND,
+ return_parameters=b"\x00",
+ )
+ )
- d2.host.on_hci_packet(HCI_Connection_Complete_Event(
- status = HCI_SUCCESS,
- connection_handle = 0x101,
- bd_addr = d0.public_address,
- link_type = HCI_Connection_Complete_Event.ACL_LINK_TYPE,
- encryption_enabled = True,
- ))
+ d2.host.on_hci_packet(
+ HCI_Connection_Complete_Event(
+ status=HCI_SUCCESS,
+ connection_handle=0x101,
+ bd_addr=d0.public_address,
+ link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
+ encryption_enabled=True,
+ )
+ )
- d0.host.on_hci_packet(HCI_Connection_Complete_Event(
- status = HCI_SUCCESS,
- connection_handle = 0x101,
- bd_addr = d2.public_address,
- link_type = HCI_Connection_Complete_Event.ACL_LINK_TYPE,
- encryption_enabled = True,
- ))
+ d0.host.on_hci_packet(
+ HCI_Connection_Complete_Event(
+ status=HCI_SUCCESS,
+ connection_handle=0x101,
+ bd_addr=d2.public_address,
+ link_type=HCI_Connection_Complete_Event.ACL_LINK_TYPE,
+ encryption_enabled=True,
+ )
+ )
assert (yield) == None
@@ -158,13 +197,19 @@
d1.host.set_packet_sink(Sink(d1_flow()))
d2.host.set_packet_sink(Sink(d2_flow()))
- [c01, c02, a10, a20, a01] = await asyncio.gather(*[
- asyncio.create_task(d0.connect(d1.public_address, transport=BT_BR_EDR_TRANSPORT)),
- asyncio.create_task(d0.connect(d2.public_address, transport=BT_BR_EDR_TRANSPORT)),
- asyncio.create_task(d1.accept(peer_address=d0.public_address)),
- asyncio.create_task(d2.accept()),
- asyncio.create_task(d0.accept(peer_address=d1.public_address)),
- ])
+ [c01, c02, a10, a20, a01] = await asyncio.gather(
+ *[
+ asyncio.create_task(
+ d0.connect(d1.public_address, transport=BT_BR_EDR_TRANSPORT)
+ ),
+ asyncio.create_task(
+ d0.connect(d2.public_address, transport=BT_BR_EDR_TRANSPORT)
+ ),
+ asyncio.create_task(d1.accept(peer_address=d0.public_address)),
+ asyncio.create_task(d2.accept()),
+ asyncio.create_task(d0.accept(peer_address=d1.public_address)),
+ ]
+ )
assert type(c01) == Connection
assert type(c02) == Connection
@@ -205,5 +250,5 @@
# -----------------------------------------------------------------------------
if __name__ == '__main__':
- logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())
+ logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())
asyncio.run(run_test_device())
