diff options
| -rw-r--r-- | tests/multi_bluetooth/ble_characteristic.py | 164 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_characteristic.py.exp | 24 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_gap_advertise.py | 46 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_gap_advertise.py.exp | 5 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_gap_connect.py | 94 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_gap_connect.py.exp | 15 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_gatt_data_transfer.py | 168 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_gatt_data_transfer.py.exp | 30 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_gattc_discover_services.py | 99 | ||||
| -rw-r--r-- | tests/multi_bluetooth/ble_gattc_discover_services.py.exp | 11 |
10 files changed, 656 insertions, 0 deletions
diff --git a/tests/multi_bluetooth/ble_characteristic.py b/tests/multi_bluetooth/ble_characteristic.py new file mode 100644 index 000000000..2f6fe404f --- /dev/null +++ b/tests/multi_bluetooth/ble_characteristic.py @@ -0,0 +1,164 @@ +# Test characteristic read/write/notify from both GATTS and GATTC. + +from micropython import const +import time, machine, bluetooth + +TIMEOUT_MS = 5000 + +_IRQ_CENTRAL_CONNECT = const(1 << 0) +_IRQ_CENTRAL_DISCONNECT = const(1 << 1) +_IRQ_GATTS_WRITE = const(1 << 2) +_IRQ_PERIPHERAL_CONNECT = const(1 << 6) +_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7) +_IRQ_GATTC_SERVICE_RESULT = const(1 << 8) +_IRQ_GATTC_CHARACTERISTIC_RESULT = const(1 << 9) +_IRQ_GATTC_READ_RESULT = const(1 << 11) +_IRQ_GATTC_WRITE_STATUS = const(1 << 12) +_IRQ_GATTC_NOTIFY = const(1 << 13) + +SERVICE_UUID = bluetooth.UUID("A5A5A5A5-FFFF-9999-1111-5A5A5A5A5A5A") +CHAR_UUID = bluetooth.UUID("00000000-1111-2222-3333-444444444444") +CHAR = ( + CHAR_UUID, + bluetooth.FLAG_READ | bluetooth.FLAG_WRITE | bluetooth.FLAG_NOTIFY, +) +SERVICE = ( + SERVICE_UUID, + (CHAR,), +) +SERVICES = (SERVICE,) + +last_event = None +last_data = None +value_handle = 0 + + +def irq(event, data): + global last_event, last_data, value_handle + last_event = event + last_data = data + if event == _IRQ_CENTRAL_CONNECT: + print("_IRQ_CENTRAL_CONNECT") + elif event == _IRQ_CENTRAL_DISCONNECT: + print("_IRQ_CENTRAL_DISCONNECT") + elif event == _IRQ_GATTS_WRITE: + print("_IRQ_GATTS_WRITE", ble.gatts_read(data[-1])) + elif event == _IRQ_PERIPHERAL_CONNECT: + print("_IRQ_PERIPHERAL_CONNECT") + elif event == _IRQ_PERIPHERAL_DISCONNECT: + print("_IRQ_PERIPHERAL_DISCONNECT") + elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT: + # conn_handle, def_handle, value_handle, properties, uuid = data + if data[-1] == CHAR_UUID: + print("_IRQ_GATTC_SERVICE_RESULT", data[-1]) + value_handle = data[2] + elif event == _IRQ_GATTC_READ_RESULT: + print("_IRQ_GATTC_READ_RESULT", data[-1]) + elif event == _IRQ_GATTC_WRITE_STATUS: + print("_IRQ_GATTC_WRITE_STATUS", data[-1]) + elif event == _IRQ_GATTC_NOTIFY: + print("_IRQ_GATTC_NOTIFY", data[-1]) + + +def wait_for_event(event, timeout_ms): + t0 = time.ticks_ms() + while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms: + if isinstance(event, int): + if last_event == event: + break + elif event(): + break + machine.idle() + + +# Acting in peripheral role. +def instance0(): + multitest.globals(BDADDR=ble.config("mac")) + ((char_handle,),) = ble.gatts_register_services(SERVICES) + print("gap_advertise") + ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY") + multitest.next() + try: + # Write initial characteristic value. + ble.gatts_write(char_handle, "periph0") + + # Wait for central to connect to us. + wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS) + if last_event != _IRQ_CENTRAL_CONNECT: + return + conn_handle, _, _ = last_data + + # Wait for a write to the characteristic from the central. + wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS) + + # Wait a bit, then write the characteristic and notify it. + time.sleep_ms(1000) + ble.gatts_write(char_handle, "periph1") + ble.gatts_notify(conn_handle, char_handle) + + # Wait for a write to the characteristic from the central. + wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS) + + # Wait a bit, then notify a new value on the characteristic. + time.sleep_ms(1000) + ble.gatts_notify(conn_handle, char_handle, "periph2") + + # Wait for the central to disconnect. + wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS) + finally: + ble.active(0) + + +# Acting in central role. +def instance1(): + multitest.next() + try: + # Connect to peripheral and then disconnect. + print("gap_connect") + ble.gap_connect(0, BDADDR) + wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS) + if last_event != _IRQ_PERIPHERAL_CONNECT: + return + conn_handle, _, _ = last_data + + # Discover characteristics. + ble.gattc_discover_characteristics(conn_handle, 1, 65535) + wait_for_event(lambda: value_handle, TIMEOUT_MS) + + # Issue read of characteristic, should get initial value. + print("gattc_read") + ble.gattc_read(conn_handle, value_handle) + wait_for_event(_IRQ_GATTC_READ_RESULT, TIMEOUT_MS) + + # Write to the characteristic, and ask for a response. + print("gattc_write") + ble.gattc_write(conn_handle, value_handle, "central0", 1) + wait_for_event(_IRQ_GATTC_WRITE_STATUS, TIMEOUT_MS) + + # Wait for a notify, then read new value. + wait_for_event(_IRQ_GATTC_NOTIFY, TIMEOUT_MS) + print("gattc_read") + ble.gattc_read(conn_handle, value_handle) + wait_for_event(_IRQ_GATTC_READ_RESULT, TIMEOUT_MS) + + # Write to the characteristic, and ask for a response. + print("gattc_write") + ble.gattc_write(conn_handle, value_handle, "central1", 1) + wait_for_event(_IRQ_GATTC_WRITE_STATUS, TIMEOUT_MS) + + # Wait for a notify (should have new data), then read old value (should be unchanged). + wait_for_event(_IRQ_GATTC_NOTIFY, TIMEOUT_MS) + print("gattc_read") + ble.gattc_read(conn_handle, value_handle) + wait_for_event(_IRQ_GATTC_READ_RESULT, TIMEOUT_MS) + + # Disconnect from peripheral. + print("gap_disconnect:", ble.gap_disconnect(conn_handle)) + wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS) + finally: + ble.active(0) + + +ble = bluetooth.BLE() +ble.active(1) +ble.irq(irq) diff --git a/tests/multi_bluetooth/ble_characteristic.py.exp b/tests/multi_bluetooth/ble_characteristic.py.exp new file mode 100644 index 000000000..d88949739 --- /dev/null +++ b/tests/multi_bluetooth/ble_characteristic.py.exp @@ -0,0 +1,24 @@ +--- instance0 --- +gap_advertise +_IRQ_CENTRAL_CONNECT +_IRQ_GATTS_WRITE b'central0' +_IRQ_GATTS_WRITE b'central1' +_IRQ_CENTRAL_DISCONNECT +--- instance1 --- +gap_connect +_IRQ_PERIPHERAL_CONNECT +_IRQ_GATTC_SERVICE_RESULT UUID128('00000000-1111-2222-3333-444444444444') +gattc_read +_IRQ_GATTC_READ_RESULT b'periph0' +gattc_write +_IRQ_GATTC_WRITE_STATUS 0 +_IRQ_GATTC_NOTIFY b'periph1' +gattc_read +_IRQ_GATTC_READ_RESULT b'periph1' +gattc_write +_IRQ_GATTC_WRITE_STATUS 0 +_IRQ_GATTC_NOTIFY b'periph2' +gattc_read +_IRQ_GATTC_READ_RESULT b'central1' +gap_disconnect: True +_IRQ_PERIPHERAL_DISCONNECT diff --git a/tests/multi_bluetooth/ble_gap_advertise.py b/tests/multi_bluetooth/ble_gap_advertise.py new file mode 100644 index 000000000..6b43f275f --- /dev/null +++ b/tests/multi_bluetooth/ble_gap_advertise.py @@ -0,0 +1,46 @@ +# Test BLE GAP advertising and scanning + +from micropython import const +import time, machine, bluetooth + +_IRQ_SCAN_RESULT = const(1 << 4) +_IRQ_SCAN_COMPLETE = const(1 << 5) + +ADV_TIME_S = 3 + + +def instance0(): + multitest.globals(BDADDR=ble.config("mac")) + print("gap_advertise(20_000)") + ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY") + multitest.next() + time.sleep(ADV_TIME_S) + print("gap_advertise(None)") + ble.gap_advertise(None) + ble.active(0) + + +def instance1(): + multitest.next() + finished = False + adv_data = None + + def irq(ev, data): + nonlocal finished, adv_data + if ev == _IRQ_SCAN_RESULT: + if data[1] == BDADDR: + adv_data = bytes(data[4]) + elif ev == _IRQ_SCAN_COMPLETE: + finished = True + + ble.config(rxbuf=2000) + ble.irq(irq) + ble.gap_scan(ADV_TIME_S * 1000, 10000, 10000) + while not finished: + machine.idle() + ble.active(0) + print("adv_data:", adv_data) + + +ble = bluetooth.BLE() +ble.active(1) diff --git a/tests/multi_bluetooth/ble_gap_advertise.py.exp b/tests/multi_bluetooth/ble_gap_advertise.py.exp new file mode 100644 index 000000000..b2ac61eb1 --- /dev/null +++ b/tests/multi_bluetooth/ble_gap_advertise.py.exp @@ -0,0 +1,5 @@ +--- instance0 --- +gap_advertise(20_000) +gap_advertise(None) +--- instance1 --- +adv_data: b'\x02\x01\x06\x04\xffMPY' diff --git a/tests/multi_bluetooth/ble_gap_connect.py b/tests/multi_bluetooth/ble_gap_connect.py new file mode 100644 index 000000000..def256f2c --- /dev/null +++ b/tests/multi_bluetooth/ble_gap_connect.py @@ -0,0 +1,94 @@ +# Test BLE GAP connect/disconnect + +from micropython import const +import time, machine, bluetooth + +TIMEOUT_MS = 4000 + +_IRQ_CENTRAL_CONNECT = const(1 << 0) +_IRQ_CENTRAL_DISCONNECT = const(1 << 1) +_IRQ_PERIPHERAL_CONNECT = const(1 << 6) +_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7) + +last_event = None +last_data = None + + +def irq(event, data): + global last_event, last_data + last_event = event + last_data = data + if event == _IRQ_CENTRAL_CONNECT: + print("_IRQ_CENTRAL_CONNECT") + elif event == _IRQ_CENTRAL_DISCONNECT: + print("_IRQ_CENTRAL_DISCONNECT") + elif event == _IRQ_PERIPHERAL_CONNECT: + print("_IRQ_PERIPHERAL_CONNECT") + elif event == _IRQ_PERIPHERAL_DISCONNECT: + print("_IRQ_PERIPHERAL_DISCONNECT") + + +def wait_for_event(event, timeout_ms): + t0 = time.ticks_ms() + while last_event != event and time.ticks_diff(time.ticks_ms(), t0) < timeout_ms: + machine.idle() + + +# Acting in peripheral role. +def instance0(): + multitest.globals(BDADDR=ble.config("mac")) + print("gap_advertise") + ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY") + multitest.next() + try: + # Wait for central to connect, then wait for it to disconnect. + wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS) + wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS) + if last_event != _IRQ_CENTRAL_DISCONNECT: + return + + # Start advertising again. + ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY") + + # Wait for central to connect, then disconnect it. + wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS) + if last_event != _IRQ_CENTRAL_CONNECT: + return + print("gap_disconnect:", ble.gap_disconnect(last_data[0])) + wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS) + finally: + ble.active(0) + + +# Acting in central role. +def instance1(): + multitest.next() + try: + # Connect to peripheral and then disconnect. + print("gap_connect") + ble.gap_connect(0, BDADDR) + wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS) + if last_event != _IRQ_PERIPHERAL_CONNECT: + return + print("gap_disconnect:", ble.gap_disconnect(last_data[0])) + wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS) + if last_event != _IRQ_PERIPHERAL_DISCONNECT: + return + + # Wait for peripheral to start advertising again. + time.sleep_ms(100) + + # Connect to peripheral and then let the peripheral disconnect us. + print("gap_connect") + ble.gap_connect(0, BDADDR) + wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS) + if last_event != _IRQ_PERIPHERAL_CONNECT: + return + wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS) + finally: + ble.active(0) + + +ble = bluetooth.BLE() +ble.active(1) +ble.irq(irq) diff --git a/tests/multi_bluetooth/ble_gap_connect.py.exp b/tests/multi_bluetooth/ble_gap_connect.py.exp new file mode 100644 index 000000000..1c96f1d48 --- /dev/null +++ b/tests/multi_bluetooth/ble_gap_connect.py.exp @@ -0,0 +1,15 @@ +--- instance0 --- +gap_advertise +_IRQ_CENTRAL_CONNECT +_IRQ_CENTRAL_DISCONNECT +_IRQ_CENTRAL_CONNECT +gap_disconnect: True +_IRQ_CENTRAL_DISCONNECT +--- instance1 --- +gap_connect +_IRQ_PERIPHERAL_CONNECT +gap_disconnect: True +_IRQ_PERIPHERAL_DISCONNECT +gap_connect +_IRQ_PERIPHERAL_CONNECT +_IRQ_PERIPHERAL_DISCONNECT diff --git a/tests/multi_bluetooth/ble_gatt_data_transfer.py b/tests/multi_bluetooth/ble_gatt_data_transfer.py new file mode 100644 index 000000000..c1e38a3d8 --- /dev/null +++ b/tests/multi_bluetooth/ble_gatt_data_transfer.py @@ -0,0 +1,168 @@ +# Test GATTC/S data transfer between peripheral and central, and use of gatts_set_buffer() + +from micropython import const +import time, machine, bluetooth + +TIMEOUT_MS = 5000 + +_IRQ_CENTRAL_CONNECT = const(1 << 0) +_IRQ_CENTRAL_DISCONNECT = const(1 << 1) +_IRQ_GATTS_WRITE = const(1 << 2) +_IRQ_PERIPHERAL_CONNECT = const(1 << 6) +_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7) +_IRQ_GATTC_SERVICE_RESULT = const(1 << 8) +_IRQ_GATTC_CHARACTERISTIC_RESULT = const(1 << 9) +_IRQ_GATTC_READ_RESULT = const(1 << 11) +_IRQ_GATTC_WRITE_STATUS = const(1 << 12) +_IRQ_GATTC_NOTIFY = const(1 << 13) + +FLAG_WRITE_NO_RESPONSE = const(1 << 2) + +SERVICE_UUID = bluetooth.UUID("00000001-1111-2222-3333-444444444444") +CHAR_CTRL_UUID = bluetooth.UUID("00000002-1111-2222-3333-444444444444") +CHAR_RX_UUID = bluetooth.UUID("00000003-1111-2222-3333-444444444444") +CHAR_TX_UUID = bluetooth.UUID("00000004-1111-2222-3333-444444444444") +CHAR_CTRL = (CHAR_CTRL_UUID, bluetooth.FLAG_WRITE | bluetooth.FLAG_NOTIFY) +CHAR_RX = (CHAR_RX_UUID, bluetooth.FLAG_WRITE | FLAG_WRITE_NO_RESPONSE) +CHAR_TX = (CHAR_TX_UUID, bluetooth.FLAG_NOTIFY) +SERVICE = (SERVICE_UUID, (CHAR_CTRL, CHAR_RX, CHAR_TX)) +SERVICES = (SERVICE,) + +last_event = None +last_data = None +ctrl_value_handle = 0 +rx_value_handle = 0 +tx_value_handle = 0 + + +def irq(event, data): + global last_event, last_data, ctrl_value_handle, rx_value_handle, tx_value_handle + last_event = event + last_data = data + if event == _IRQ_CENTRAL_CONNECT: + print("_IRQ_CENTRAL_CONNECT") + elif event == _IRQ_CENTRAL_DISCONNECT: + print("_IRQ_CENTRAL_DISCONNECT") + elif event == _IRQ_GATTS_WRITE: + print("_IRQ_GATTS_WRITE") + elif event == _IRQ_PERIPHERAL_CONNECT: + print("_IRQ_PERIPHERAL_CONNECT") + elif event == _IRQ_PERIPHERAL_DISCONNECT: + print("_IRQ_PERIPHERAL_DISCONNECT") + elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT: + if data[-1] == CHAR_CTRL_UUID: + print("_IRQ_GATTC_SERVICE_RESULT", data[-1]) + ctrl_value_handle = data[2] + elif data[-1] == CHAR_RX_UUID: + print("_IRQ_GATTC_SERVICE_RESULT", data[-1]) + rx_value_handle = data[2] + elif data[-1] == CHAR_TX_UUID: + print("_IRQ_GATTC_SERVICE_RESULT", data[-1]) + tx_value_handle = data[2] + elif event == _IRQ_GATTC_READ_RESULT: + print("_IRQ_GATTC_READ_RESULT", data[-1]) + elif event == _IRQ_GATTC_WRITE_STATUS: + print("_IRQ_GATTC_WRITE_STATUS", data[-1]) + elif event == _IRQ_GATTC_NOTIFY: + print("_IRQ_GATTC_NOTIFY", data[-1]) + + +def wait_for_event(event, timeout_ms): + t0 = time.ticks_ms() + while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms: + if isinstance(event, int): + if last_event == event: + break + elif event(): + break + machine.idle() + + +# Acting in peripheral role. +def instance0(): + multitest.globals(BDADDR=ble.config("mac")) + ((char_ctrl_handle, char_rx_handle, char_tx_handle),) = ble.gatts_register_services(SERVICES) + + # Increase the size of the rx buffer and enable append mode. + ble.gatts_set_buffer(char_rx_handle, 100, True) + + print("gap_advertise") + ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY") + multitest.next() + try: + # Wait for central to connect to us. + wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS) + if last_event != _IRQ_CENTRAL_CONNECT: + return + conn_handle, _, _ = last_data + + # Wait for the central to signal that it's done with its part of the test. + wait_for_event( + lambda: last_event == _IRQ_GATTS_WRITE and last_data[1] == char_ctrl_handle, + 2 * TIMEOUT_MS, + ) + + # Read all accumulated data from the central. + print("gatts_read:", ble.gatts_read(char_rx_handle)) + + # Notify the central a few times. + for i in range(4): + time.sleep_ms(300) + ble.gatts_notify(conn_handle, char_tx_handle, "message{}".format(i)) + + # Notify the central that we are done with our part of the test. + time.sleep_ms(300) + ble.gatts_notify(conn_handle, char_ctrl_handle, "OK") + + # Wait for the central to disconnect. + wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS) + finally: + ble.active(0) + + +# Acting in central role. +def instance1(): + multitest.next() + try: + # Connect to peripheral and then disconnect. + print("gap_connect") + ble.gap_connect(0, BDADDR) + wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS) + if last_event != _IRQ_PERIPHERAL_CONNECT: + return + conn_handle, _, _ = last_data + + # Discover characteristics. + ble.gattc_discover_characteristics(conn_handle, 1, 65535) + wait_for_event( + lambda: ctrl_value_handle and rx_value_handle and tx_value_handle, TIMEOUT_MS + ) + + # Write to the characteristic a few times, with and without response. + for i in range(4): + print("gattc_write") + ble.gattc_write(conn_handle, rx_value_handle, "central{}".format(i), i & 1) + if i & 1: + wait_for_event(_IRQ_GATTC_WRITE_STATUS, TIMEOUT_MS) + time.sleep_ms(400) + + # Write to say that we are done with our part of the test. + ble.gattc_write(conn_handle, ctrl_value_handle, "OK", 1) + wait_for_event(_IRQ_GATTC_WRITE_STATUS, TIMEOUT_MS) + + # Wait for notification that peripheral is done with its part of the test. + wait_for_event( + lambda: last_event == _IRQ_GATTC_NOTIFY and last_data[1] == ctrl_value_handle, + TIMEOUT_MS, + ) + + # Disconnect from peripheral. + print("gap_disconnect:", ble.gap_disconnect(conn_handle)) + wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS) + finally: + ble.active(0) + + +ble = bluetooth.BLE() +ble.active(1) +ble.irq(irq) diff --git a/tests/multi_bluetooth/ble_gatt_data_transfer.py.exp b/tests/multi_bluetooth/ble_gatt_data_transfer.py.exp new file mode 100644 index 000000000..6ecd48225 --- /dev/null +++ b/tests/multi_bluetooth/ble_gatt_data_transfer.py.exp @@ -0,0 +1,30 @@ +--- instance0 --- +gap_advertise +_IRQ_CENTRAL_CONNECT +_IRQ_GATTS_WRITE +_IRQ_GATTS_WRITE +_IRQ_GATTS_WRITE +_IRQ_GATTS_WRITE +_IRQ_GATTS_WRITE +gatts_read: b'central0central1central2central3' +_IRQ_CENTRAL_DISCONNECT +--- instance1 --- +gap_connect +_IRQ_PERIPHERAL_CONNECT +_IRQ_GATTC_SERVICE_RESULT UUID128('00000002-1111-2222-3333-444444444444') +_IRQ_GATTC_SERVICE_RESULT UUID128('00000003-1111-2222-3333-444444444444') +_IRQ_GATTC_SERVICE_RESULT UUID128('00000004-1111-2222-3333-444444444444') +gattc_write +gattc_write +_IRQ_GATTC_WRITE_STATUS 0 +gattc_write +gattc_write +_IRQ_GATTC_WRITE_STATUS 0 +_IRQ_GATTC_WRITE_STATUS 0 +_IRQ_GATTC_NOTIFY b'message0' +_IRQ_GATTC_NOTIFY b'message1' +_IRQ_GATTC_NOTIFY b'message2' +_IRQ_GATTC_NOTIFY b'message3' +_IRQ_GATTC_NOTIFY b'OK' +gap_disconnect: True +_IRQ_PERIPHERAL_DISCONNECT diff --git a/tests/multi_bluetooth/ble_gattc_discover_services.py b/tests/multi_bluetooth/ble_gattc_discover_services.py new file mode 100644 index 000000000..b01a890c7 --- /dev/null +++ b/tests/multi_bluetooth/ble_gattc_discover_services.py @@ -0,0 +1,99 @@ +# Test BLE GAP connect/disconnect + +from micropython import const +import time, machine, bluetooth + +TIMEOUT_MS = 5000 + +_IRQ_CENTRAL_CONNECT = const(1 << 0) +_IRQ_CENTRAL_DISCONNECT = const(1 << 1) +_IRQ_PERIPHERAL_CONNECT = const(1 << 6) +_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7) +_IRQ_GATTC_SERVICE_RESULT = const(1 << 8) + +UUID_A = bluetooth.UUID(0x180D) +UUID_B = bluetooth.UUID("A5A5A5A5-FFFF-9999-1111-5A5A5A5A5A5A") +SERVICE_A = ( + UUID_A, + (), +) +SERVICE_B = ( + UUID_B, + (), +) +SERVICES = (SERVICE_A, SERVICE_B) + +last_event = None +last_data = None +num_service_result = 0 + + +def irq(event, data): + global last_event, last_data, num_service_result + last_event = event + last_data = data + if event == _IRQ_CENTRAL_CONNECT: + print("_IRQ_CENTRAL_CONNECT") + elif event == _IRQ_CENTRAL_DISCONNECT: + print("_IRQ_CENTRAL_DISCONNECT") + elif event == _IRQ_PERIPHERAL_CONNECT: + print("_IRQ_PERIPHERAL_CONNECT") + elif event == _IRQ_PERIPHERAL_DISCONNECT: + print("_IRQ_PERIPHERAL_DISCONNECT") + elif event == _IRQ_GATTC_SERVICE_RESULT: + if data[3] == UUID_A or data[3] == UUID_B: + print("_IRQ_GATTC_SERVICE_RESULT", data[3]) + num_service_result += 1 + + +def wait_for_event(event, timeout_ms): + t0 = time.ticks_ms() + while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms: + if isinstance(event, int): + if last_event == event: + break + elif event(): + break + machine.idle() + + +# Acting in peripheral role. +def instance0(): + multitest.globals(BDADDR=ble.config("mac")) + ble.gatts_register_services(SERVICES) + print("gap_advertise") + ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY") + multitest.next() + try: + wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS) + wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS) + finally: + ble.active(0) + + +# Acting in central role. +def instance1(): + multitest.next() + try: + # Connect to peripheral and then disconnect. + print("gap_connect") + ble.gap_connect(0, BDADDR) + wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS) + if last_event != _IRQ_PERIPHERAL_CONNECT: + return + conn_handle, _, _ = last_data + + # Discover services. + ble.gattc_discover_services(conn_handle) + wait_for_event(lambda: num_service_result == 2, TIMEOUT_MS) + + # Disconnect from peripheral. + print("gap_disconnect:", ble.gap_disconnect(conn_handle)) + wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS) + finally: + ble.active(0) + + +ble = bluetooth.BLE() +ble.active(1) +ble.irq(irq) diff --git a/tests/multi_bluetooth/ble_gattc_discover_services.py.exp b/tests/multi_bluetooth/ble_gattc_discover_services.py.exp new file mode 100644 index 000000000..06c36b249 --- /dev/null +++ b/tests/multi_bluetooth/ble_gattc_discover_services.py.exp @@ -0,0 +1,11 @@ +--- instance0 --- +gap_advertise +_IRQ_CENTRAL_CONNECT +_IRQ_CENTRAL_DISCONNECT +--- instance1 --- +gap_connect +_IRQ_PERIPHERAL_CONNECT +_IRQ_GATTC_SERVICE_RESULT UUID16(0x180d) +_IRQ_GATTC_SERVICE_RESULT UUID128('a5a5a5a5-ffff-9999-1111-5a5a5a5a5a5a') +gap_disconnect: True +_IRQ_PERIPHERAL_DISCONNECT |