mirror of
https://github.com/google/bumble.git
synced 2026-07-14 14:10:50 +00:00
Add RFCOMM and SDP helpers
This commit is contained in:
+50
-1
@@ -19,7 +19,17 @@ import asyncio
|
||||
import pytest
|
||||
|
||||
from . import test_utils
|
||||
from bumble.rfcomm import RFCOMM_Frame, Server, Client, DLC
|
||||
from bumble import core
|
||||
from bumble.rfcomm import (
|
||||
RFCOMM_Frame,
|
||||
Server,
|
||||
Client,
|
||||
DLC,
|
||||
make_service_sdp_records,
|
||||
find_rfcomm_channels,
|
||||
find_rfcomm_channel_with_uuid,
|
||||
RFCOMM_PSM,
|
||||
)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@@ -70,6 +80,45 @@ async def test_basic_connection():
|
||||
assert await queues[0].get() == b'Lorem ipsum dolor sit amet'
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@pytest.mark.asyncio
|
||||
async def test_service_record():
|
||||
HANDLE = 2
|
||||
CHANNEL = 1
|
||||
SERVICE_UUID = core.UUID('00000000-0000-0000-0000-000000000001')
|
||||
|
||||
devices = test_utils.TwoDevices()
|
||||
await devices.setup_connection()
|
||||
|
||||
devices[0].sdp_service_records[HANDLE] = make_service_sdp_records(
|
||||
HANDLE, CHANNEL, SERVICE_UUID
|
||||
)
|
||||
|
||||
assert SERVICE_UUID in (await find_rfcomm_channels(devices.connections[1]))[CHANNEL]
|
||||
assert (
|
||||
await find_rfcomm_channel_with_uuid(devices.connections[1], SERVICE_UUID)
|
||||
== CHANNEL
|
||||
)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@pytest.mark.asyncio
|
||||
async def test_context():
|
||||
devices = test_utils.TwoDevices()
|
||||
await devices.setup_connection()
|
||||
|
||||
server = Server(devices[0])
|
||||
with server:
|
||||
assert server.l2cap_server is not None
|
||||
|
||||
client = Client(devices.connections[1])
|
||||
async with client:
|
||||
assert client.l2cap_channel is not None
|
||||
|
||||
assert client.l2cap_channel is None
|
||||
assert RFCOMM_PSM not in devices[0].l2cap_channel_manager.servers
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
if __name__ == '__main__':
|
||||
test_frames()
|
||||
|
||||
@@ -38,6 +38,7 @@ from .test_utils import TwoDevices
|
||||
# pylint: disable=invalid-name
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
def basic_check(x: DataElement) -> None:
|
||||
serialized = bytes(x)
|
||||
@@ -269,6 +270,20 @@ async def test_service_search_attribute():
|
||||
assert expect.value == actual.value
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_async_context():
|
||||
devices = TwoDevices()
|
||||
await devices.setup_connection()
|
||||
|
||||
client = Client(devices.connections[1])
|
||||
|
||||
async with client:
|
||||
assert client.channel is not None
|
||||
|
||||
assert client.channel is None
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
async def run():
|
||||
test_data_elements()
|
||||
|
||||
Reference in New Issue
Block a user