pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

Reorganize exceptions

Browse Source 
* Add BaseBumbleException as a "real" root error
* Add several core error classes and properly replace builtin errors
  with them
* Add several error classes for specific modules (transport, device)
This commit is contained in:
Josh Wu
2024-06-04 16:11:26 +08:00
parent 090309302f
commit f8a2d4f0e0
26 changed files with 260 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
bumble/transport/__init__.py
View File

@@ -20,7 +20,7 @@ import logging
import os
from typing import Optional
from .common import Transport, AsyncPipeSink, SnoopingTransport
from .common import Transport, AsyncPipeSink, SnoopingTransport, TransportSpecError
from ..snoop import create_snooper
# -----------------------------------------------------------------------------
@@ -180,7 +180,7 @@ async def _open_transport(scheme: str, spec: Optional[str]) -> Transport:
return await open_android_netsim_transport(spec)
raise ValueError('unknown transport scheme')
raise TransportSpecError('unknown transport scheme')
# -----------------------------------------------------------------------------

12
bumble/transport/android_emulator.py
View File

@@ -20,7 +20,13 @@ import grpc.aio
from typing import Optional, Union
from .common import PumpedTransport, PumpedPacketSource, PumpedPacketSink, Transport
from .common import (
PumpedTransport,
PumpedPacketSource,
PumpedPacketSink,
Transport,
TransportSpecError,
)
# pylint: disable=no-name-in-module
from .grpc_protobuf.emulated_bluetooth_pb2_grpc import EmulatedBluetoothServiceStub
@@ -77,7 +83,7 @@ async def open_android_emulator_transport(spec: Optional[str]) -> Transport:
elif ':' in param:
server_host, server_port = param.split(':')
else:
raise ValueError('invalid parameter')
raise TransportSpecError('invalid parameter')
# Connect to the gRPC server
server_address = f'{server_host}:{server_port}'
@@ -94,7 +100,7 @@ async def open_android_emulator_transport(spec: Optional[str]) -> Transport:
service = VhciForwardingServiceStub(channel)
hci_device = HciDevice(service.attachVhci())
else:
raise ValueError('invalid mode')
raise TransportSpecError('invalid mode')
# Create the transport object
class EmulatorTransport(PumpedTransport):

16
bumble/transport/android_netsim.py
View File

@@ -31,6 +31,8 @@ from .common import (
PumpedPacketSource,
PumpedPacketSink,
Transport,
TransportSpecError,
TransportInitError,
)
# pylint: disable=no-name-in-module
@@ -135,7 +137,7 @@ async def open_android_netsim_controller_transport(
server_host: Optional[str], server_port: int, options: Dict[str, str]
) -> Transport:
if not server_port:
raise ValueError('invalid port')
raise TransportSpecError('invalid port')
if server_host == '_' or not server_host:
server_host = 'localhost'
@@ -288,7 +290,7 @@ async def open_android_netsim_host_transport_with_address(
instance_number = 0 if options is None else int(options.get('instance', '0'))
server_port = find_grpc_port(instance_number)
if not server_port:
raise RuntimeError('gRPC server port not found')
raise TransportInitError('gRPC server port not found')
# Connect to the gRPC server
server_address = f'{server_host}:{server_port}'
@@ -326,7 +328,7 @@ async def open_android_netsim_host_transport_with_channel(
if response_type == 'error':
logger.warning(f'received error: {response.error}')
raise RuntimeError(response.error)
raise TransportInitError(response.error)
if response_type == 'hci_packet':
return (
@@ -334,7 +336,7 @@ async def open_android_netsim_host_transport_with_channel(
+ response.hci_packet.packet
)
raise ValueError('unsupported response type')
raise TransportSpecError('unsupported response type')
async def write(self, packet):
await self.hci_device.write(
@@ -429,7 +431,7 @@ async def open_android_netsim_transport(spec: Optional[str]) -> Transport:
options: Dict[str, str] = {}
for param in params[params_offset:]:
if '=' not in param:
raise ValueError('invalid parameter, expected <name>=<value>')
raise TransportSpecError('invalid parameter, expected <name>=<value>')
option_name, option_value = param.split('=')
options[option_name] = option_value
@@ -440,7 +442,7 @@ async def open_android_netsim_transport(spec: Optional[str]) -> Transport:
)
if mode == 'controller':
if host is None:
raise ValueError('<host>:<port> missing')
raise TransportSpecError('<host>:<port> missing')
return await open_android_netsim_controller_transport(host, port, options)
raise ValueError('invalid mode option')
raise TransportSpecError('invalid mode option')

29
bumble/transport/common.py
View File

@@ -23,6 +23,7 @@ import logging
import io
from typing import Any, ContextManager, Tuple, Optional, Protocol, Dict
from bumble import core
from bumble import hci
from bumble.colors import color
from bumble.snoop import Snooper
@@ -49,10 +50,16 @@ HCI_PACKET_INFO: Dict[int, Tuple[int, int, str]] = {
# -----------------------------------------------------------------------------
# Errors
# -----------------------------------------------------------------------------
class TransportLostError(Exception):
"""
The Transport has been lost/disconnected.
"""
class TransportLostError(core.BaseBumbleError, RuntimeError):
"""The Transport has been lost/disconnected."""
class TransportInitError(core.BaseBumbleError, RuntimeError):
"""Error raised when the transport cannot be initialized."""
class TransportSpecError(core.BaseBumbleError, ValueError):
"""Error raised when the transport spec is invalid."""
# -----------------------------------------------------------------------------
@@ -132,7 +139,9 @@ class PacketParser:
packet_type
) or self.extended_packet_info.get(packet_type)
if self.packet_info is None:
raise ValueError(f'invalid packet type {packet_type}')
raise core.InvalidPacketError(
f'invalid packet type {packet_type}'
)
self.state = PacketParser.NEED_LENGTH
self.bytes_needed = self.packet_info[0] + self.packet_info[1]
elif self.state == PacketParser.NEED_LENGTH:
@@ -178,19 +187,19 @@ class PacketReader:
# Get the packet info based on its type
packet_info = HCI_PACKET_INFO.get(packet_type[0])
if packet_info is None:
raise ValueError(f'invalid packet type {packet_type[0]} found')
raise core.InvalidPacketError(f'invalid packet type {packet_type[0]} found')
# Read the header (that includes the length)
header_size = packet_info[0] + packet_info[1]
header = self.source.read(header_size)
if len(header) != header_size:
raise ValueError('packet too short')
raise core.InvalidPacketError('packet too short')
# Read the body
body_length = struct.unpack_from(packet_info[2], header, packet_info[1])[0]
body = self.source.read(body_length)
if len(body) != body_length:
raise ValueError('packet too short')
raise core.InvalidPacketError('packet too short')
return packet_type + header + body
@@ -211,7 +220,7 @@ class AsyncPacketReader:
# Get the packet info based on its type
packet_info = HCI_PACKET_INFO.get(packet_type[0])
if packet_info is None:
raise ValueError(f'invalid packet type {packet_type[0]} found')
raise core.InvalidPacketError(f'invalid packet type {packet_type[0]} found')
# Read the header (that includes the length)
header_size = packet_info[0] + packet_info[1]
@@ -420,7 +429,7 @@ class SnoopingTransport(Transport):
return SnoopingTransport(
transport, exit_stack.enter_context(snooper), exit_stack.pop_all().close
)
raise RuntimeError('unexpected code path') # Satisfy the type checker
raise core.UnreachableError() # Satisfy the type checker
class Source:
sink: TransportSink

4
bumble/transport/pyusb.py
View File

@@ -29,7 +29,7 @@ from usb.core import USBError
from usb.util import CTRL_TYPE_CLASS, CTRL_RECIPIENT_OTHER
from usb.legacy import REQ_SET_FEATURE, REQ_CLEAR_FEATURE, CLASS_HUB
from .common import Transport, ParserSource
from .common import Transport, ParserSource, TransportInitError
from .. import hci
from ..colors import color
@@ -259,7 +259,7 @@ async def open_pyusb_transport(spec: str) -> Transport:
device = None
if device is None:
raise ValueError('device not found')
raise TransportInitError('device not found')
logger.debug(f'USB Device: {device}')
# Power Cycle the device

7
bumble/transport/usb.py
View File

@@ -24,10 +24,9 @@ import platform
import usb1
from bumble.transport.common import Transport, ParserSource
from bumble.transport.common import Transport, ParserSource, TransportInitError
from bumble import hci
from bumble.colors import color
from bumble.utils import AsyncRunner
# -----------------------------------------------------------------------------
@@ -442,7 +441,7 @@ async def open_usb_transport(spec: str) -> Transport:
if found is None:
context.close()
raise ValueError('device not found')
raise TransportInitError('device not found')
logger.debug(f'USB Device: {found}')
@@ -507,7 +506,7 @@ async def open_usb_transport(spec: str) -> Transport:
endpoints = find_endpoints(found)
if endpoints is None:
raise ValueError('no compatible interface found for device')
raise TransportInitError('no compatible interface found for device')
(configuration, interface, setting, acl_in, acl_out, events_in) = endpoints
logger.debug(
f'selected endpoints: configuration={configuration}, '