Add Device::add_default_services()

This will allow a test to:
a: add services to a device
b: reset services via `Server()`
c: add the default services back
This commit is contained in:
Alan Rosenthal
2022-11-30 16:15:26 -05:00
committed by Alan Rosenthal
parent 4417eb636c
commit a8eff737e6
2 changed files with 28 additions and 4 deletions

View File

@@ -797,9 +797,7 @@ class Device(CompositeEventEmitter):
# Register the SDP server with the L2CAP Channel Manager # Register the SDP server with the L2CAP Channel Manager
self.sdp_server.register(self.l2cap_channel_manager) self.sdp_server.register(self.l2cap_channel_manager)
# Add a GAP Service if requested self.add_default_services(generic_access_service)
if generic_access_service:
self.gatt_server.add_service(GenericAccessService(self.name))
self.l2cap_channel_manager.register_fixed_channel(ATT_CID, self.on_gatt_pdu) self.l2cap_channel_manager.register_fixed_channel(ATT_CID, self.on_gatt_pdu)
# Forward some events # Forward some events
@@ -1886,6 +1884,11 @@ class Device(CompositeEventEmitter):
def add_services(self, services): def add_services(self, services):
self.gatt_server.add_services(services) self.gatt_server.add_services(services)
def add_default_services(self, generic_access_service=True):
# Add a GAP Service if requested
if generic_access_service:
self.gatt_server.add_service(GenericAccessService(self.name))
async def notify_subscriber(self, connection, attribute, value=None, force=False): async def notify_subscriber(self, connection, attribute, value=None, force=False):
await self.gatt_server.notify_subscriber(connection, attribute, value, force) await self.gatt_server.notify_subscriber(connection, attribute, value, force)

View File

@@ -28,7 +28,7 @@ from bumble.hci import (
HCI_ACCEPT_CONNECTION_REQUEST_COMMAND, HCI_COMMAND_STATUS_PENDING, HCI_CREATE_CONNECTION_COMMAND, HCI_SUCCESS, 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 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
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
# Logging # Logging
@@ -182,6 +182,27 @@ async def run_test_device():
await test_device_connect_parallel() await test_device_connect_parallel()
# -----------------------------------------------------------------------------
def test_gatt_services_with_gas():
device = Device(host=Host(None, None))
# there should be one service and two chars, therefore 5 attributes
assert len(device.gatt_server.attributes) == 5
assert device.gatt_server.attributes[0].uuid == GATT_GENERIC_ACCESS_SERVICE
assert device.gatt_server.attributes[1].type == GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
assert device.gatt_server.attributes[2].uuid == GATT_DEVICE_NAME_CHARACTERISTIC
assert device.gatt_server.attributes[3].type == GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
assert device.gatt_server.attributes[4].uuid == GATT_APPEARANCE_CHARACTERISTIC
# -----------------------------------------------------------------------------
def test_gatt_services_without_gas():
device = Device(host=Host(None, None), generic_access_service=False)
# there should be no services
assert len(device.gatt_server.attributes) == 0
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
if __name__ == '__main__': if __name__ == '__main__':
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper()) logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())