Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/console.py
Original file line number Diff line number Diff line change
Expand Up @@ -579,7 +579,7 @@ async def rssi_monitor_loop(self):

async def command(self, command):
try:
(keyword, *params) = command.strip().split(' ')
keyword, *params = command.strip().split(' ')
keyword = keyword.replace('-', '_').lower()
handler = getattr(self, f'do_{keyword}', None)
if handler:
Expand Down
6 changes: 2 additions & 4 deletions apps/show.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,9 +63,7 @@ def __init__(self, source):
raise ValueError(
'not a valid snoop file, unexpected identification pattern'
)
(self.version_number, self.data_link_type) = struct.unpack(
'>II', source.read(8)
)
self.version_number, self.data_link_type = struct.unpack('>II', source.read(8))
if self.data_link_type not in (self.DATALINK_H4, self.DATALINK_H1):
raise ValueError(f'datalink type {self.data_link_type} not supported')

Expand Down Expand Up @@ -175,7 +173,7 @@ def read_next_packet():

while not packet_reader.at_end:
try:
(timestamp, direction, packet) = read_next_packet()
timestamp, direction, packet = read_next_packet()
if packet:
tracer.trace(hci.HCI_Packet.from_bytes(packet), direction, timestamp)
else:
Expand Down
4 changes: 2 additions & 2 deletions apps/usb_probe.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ def show_device_details(device):
suffix = (
f'/{alternate_setting}' if interface.getNumSettings() > 1 else ''
)
(class_string, subclass_string) = get_class_info(
class_string, subclass_string = get_class_info(
setting.getClass(), setting.getSubClass(), setting.getProtocol()
)
details = f'({class_string}, {subclass_string})'
Expand Down Expand Up @@ -191,7 +191,7 @@ def main(verbose: bool, manufacturer: str, product: str, hci_only: bool):

device_id = (device.getVendorID(), device.getProductID())

(device_class_string, device_subclass_string) = get_class_info(
device_class_string, device_subclass_string = get_class_info(
device_class, device_subclass, device_protocol
)

Expand Down
2 changes: 1 addition & 1 deletion bumble/a2dp.py
Original file line number Diff line number Diff line change
Expand Up @@ -459,7 +459,7 @@ class VendorSpecificMediaCodecInformation(MediaCodecInformation):

@staticmethod
def from_bytes(data: bytes) -> VendorSpecificMediaCodecInformation:
(vendor_id, codec_id) = struct.unpack_from('<IH', data, 0)
vendor_id, codec_id = struct.unpack_from('<IH', data, 0)
return VendorSpecificMediaCodecInformation(vendor_id, codec_id, data[6:])

def __bytes__(self) -> bytes:
Expand Down
4 changes: 2 additions & 2 deletions bumble/crypto/builtin.py
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ def encrypt(self, plaintext: bytes) -> bytes:
raise core.InvalidArgumentError(f'wrong block length {len(plaintext)}')

rounds = len(self._ke) - 1
(s1, s2, s3) = [1, 2, 3]
s1, s2, s3 = [1, 2, 3]
a = [0, 0, 0, 0]

# Convert plaintext to (ints ^ key)
Expand Down Expand Up @@ -205,7 +205,7 @@ def decrypt(self, cipher_text: bytes) -> bytes:
raise core.InvalidArgumentError(f'wrong block length {len(cipher_text)}')

rounds = len(self._kd) - 1
(s1, s2, s3) = [3, 2, 1]
s1, s2, s3 = [3, 2, 1]
a = [0, 0, 0, 0]

# Convert ciphertext to (ints ^ key)
Expand Down
2 changes: 1 addition & 1 deletion bumble/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -5442,7 +5442,7 @@ async def create_cs_config(
role: int = hci.CsRole.INITIATOR,
rtt_type: int = hci.RttType.AA_ONLY,
cs_sync_phy: int = hci.CsSyncPhy.LE_1M,
channel_map: bytes = b'\x54\x55\x55\x54\x55\x55\x55\x55\x55\x15',
channel_map: bytes = b'\x54\x55\x55\x54\x55\x55\x55\x55\x55\x05',
channel_map_repetition: int = 0x01,
channel_selection_type: int = hci.HCI_LE_CS_Create_Config_Command.ChannelSelectionType.ALGO_3B,
ch3c_shape: int = hci.HCI_LE_CS_Create_Config_Command.Ch3cShape.HAT,
Expand Down
1 change: 1 addition & 0 deletions bumble/drivers/rtk.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
Based on various online bits of information, including the Linux kernel.
(see `drivers/bluetooth/btrtl.c`)
"""

from __future__ import annotations

import asyncio
Expand Down
2 changes: 1 addition & 1 deletion bumble/gatt_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -219,7 +219,7 @@ def get_descriptor_attribute(
if not characteristics:
return None

(_, characteristic_value) = characteristics
_, characteristic_value = characteristics

return next(
(
Expand Down
2 changes: 1 addition & 1 deletion bumble/pandora/host.py
Original file line number Diff line number Diff line change
Expand Up @@ -639,7 +639,7 @@ async def Inquiry(
await self.device.start_discovery(auto_restart=False)
try:
while inquiry_result := await inquiry_queue.get():
(address, class_of_device, eir_data, rssi) = inquiry_result
address, class_of_device, eir_data, rssi = inquiry_result
# FIXME: if needed, add support for `page_scan_repetition_mode` and `clock_offset` in Bumble
yield InquiryResponse(
address=bytes(reversed(bytes(address))),
Expand Down
2 changes: 1 addition & 1 deletion bumble/pandora/l2cap.py
Original file line number Diff line number Diff line change
Expand Up @@ -300,7 +300,7 @@ def craft_pandora_channel(
return PandoraChannel(cookie=cookie)

def lookup_channel(self, pandora_channel: PandoraChannel) -> L2capChannel:
(connection_handle, source_cid) = json.loads(
connection_handle, source_cid = json.loads(
pandora_channel.cookie.value
).values()

Expand Down
2 changes: 1 addition & 1 deletion bumble/profiles/aics.py
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,7 @@ class GainSettingsProperties:

@classmethod
def from_bytes(cls, data: bytes):
(gain_settings_unit, gain_settings_minimum, gain_settings_maximum) = (
gain_settings_unit, gain_settings_minimum, gain_settings_maximum = (
struct.unpack('BBB', data)
)
return GainSettingsProperties(
Expand Down
2 changes: 1 addition & 1 deletion bumble/rfcomm.py
Original file line number Diff line number Diff line change
Expand Up @@ -870,7 +870,7 @@ def on_disc_frame(self, _frame: RFCOMM_Frame) -> None:
)

def on_uih_frame(self, frame: RFCOMM_Frame) -> None:
(mcc_type, c_r, value) = RFCOMM_Frame.parse_mcc(frame.information)
mcc_type, c_r, value = RFCOMM_Frame.parse_mcc(frame.information)

if mcc_type == MccType.PN:
pn = RFCOMM_MCC_PN.from_bytes(value)
Expand Down
2 changes: 1 addition & 1 deletion bumble/smp.py
Original file line number Diff line number Diff line change
Expand Up @@ -1720,7 +1720,7 @@ def on_smp_pairing_random_command_secure_connections(
# Compute the MacKey and LTK
a = self.ia + bytes([self.iat])
b = self.ra + bytes([self.rat])
(mac_key, self.ltk) = crypto.f5(self.dh_key, self.na, self.nb, a, b)
mac_key, self.ltk = crypto.f5(self.dh_key, self.na, self.nb, a, b)

# Compute the DH Key checks
if self.pairing_method in (
Expand Down
20 changes: 20 additions & 0 deletions tests/device_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
# -----------------------------------------------------------------------------
import asyncio
import functools
import inspect
import logging
import os
from unittest import mock
Expand Down Expand Up @@ -693,6 +694,25 @@ async def test_power_on_default_static_address_should_not_be_any():
assert devices[0].static_address != Address.ANY_RANDOM


# -----------------------------------------------------------------------------
def test_cs_channel_map_excludes_forbidden_channels():
forbidden = {0, 1, 23, 24, 25, 76, 77, 78, 79}
default_map = (
inspect.signature(Device.create_cs_config).parameters['channel_map'].default
)

enabled = {
byte_idx * 8 + bit
for byte_idx, byte in enumerate(default_map)
for bit in range(8)
if byte & (1 << bit)
}

assert enabled.isdisjoint(
forbidden
), f"Default channel_map enables forbidden CS channels: {enabled & forbidden}"


# -----------------------------------------------------------------------------
def test_gatt_services_with_gas_and_gatt():
device = Device(host=Host(None, None))
Expand Down
5 changes: 1 addition & 4 deletions tests/gatt_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -1191,12 +1191,9 @@ def test_characteristic_property_from_string_assert():
with pytest.raises(TypeError) as e_info:
Characteristic.Properties.from_string("BROADCAST,HELLO")

assert (
str(e_info.value)
== """Characteristic.Properties::from_string() error:
assert str(e_info.value) == """Characteristic.Properties::from_string() error:
Expected a string containing any of the keys, separated by , or |: BROADCAST,READ,WRITE_WITHOUT_RESPONSE,WRITE,NOTIFY,INDICATE,AUTHENTICATED_SIGNED_WRITES,EXTENDED_PROPERTIES
Got: BROADCAST,HELLO"""
)


# -----------------------------------------------------------------------------
Expand Down