#!/usr/bin/env python3 # # Copyright (C) 2016 The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); you may not # use this file except in compliance with the License. You may obtain a copy of # the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations under # the License. """ This test script exercises different GATT connection tests. """ import pprint from queue import Empty import time from acts.test_decorators import test_tracker_info from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest from acts.test_utils.bt.bt_constants import ble_advertise_settings_modes from acts.test_utils.bt.bt_constants import ble_scan_settings_modes from acts.test_utils.bt.bt_constants import ble_scan_settings_match_nums from acts.test_utils.bt.bt_constants import bt_profile_constants from acts.test_utils.bt.bt_constants import gatt_characteristic from acts.test_utils.bt.bt_constants import gatt_descriptor from acts.test_utils.bt.bt_constants import gatt_service_types from acts.test_utils.bt.bt_constants import gatt_cb_err from acts.test_utils.bt.bt_constants import gatt_cb_strings from acts.test_utils.bt.bt_constants import gatt_connection_state from acts.test_utils.bt.bt_constants import gatt_mtu_size from acts.test_utils.bt.bt_constants import gatt_phy_mask from acts.test_utils.bt.bt_constants import gatt_transport from acts.test_utils.bt.bt_constants import scan_result from acts.test_utils.bt.bt_gatt_utils import GattTestUtilsError from acts.test_utils.bt.bt_gatt_utils import disconnect_gatt_connection from acts.test_utils.bt.bt_gatt_utils import wait_for_gatt_disconnect_event from acts.test_utils.bt.bt_gatt_utils import close_gatt_client from acts.test_utils.bt.bt_gatt_utils import log_gatt_server_uuids from acts.test_utils.bt.bt_gatt_utils import orchestrate_gatt_connection from acts.test_utils.bt.bt_gatt_utils import setup_gatt_connection from acts.test_utils.bt.bt_gatt_utils import setup_multiple_services from acts.test_utils.bt.bt_test_utils import get_mac_address_of_generic_advertisement from acts.test_utils.bt.bt_test_utils import clear_bonded_devices PHYSICAL_DISCONNECT_TIMEOUT = 5 class GattConnectTest(BluetoothBaseTest): adv_instances = [] bluetooth_gatt_list = [] gatt_server_list = [] default_timeout = 10 default_discovery_timeout = 3 def setup_class(self): super().setup_class() self.cen_ad = self.android_devices[0] self.per_ad = self.android_devices[1] def setup_test(self): super(BluetoothBaseTest, self).setup_test() bluetooth_gatt_list = [] self.gatt_server_list = [] self.adv_instances = [] # Ensure there is ample time for a physical disconnect in between # testcases. self.log.info( "Waiting for {} seconds for physical GATT disconnections".format( PHYSICAL_DISCONNECT_TIMEOUT)) time.sleep(PHYSICAL_DISCONNECT_TIMEOUT) def teardown_test(self): for bluetooth_gatt in self.bluetooth_gatt_list: self.cen_ad.droid.gattClientClose(bluetooth_gatt) for gatt_server in self.gatt_server_list: self.per_ad.droid.gattServerClose(gatt_server) for adv in self.adv_instances: self.per_ad.droid.bleStopBleAdvertising(adv) return True def _orchestrate_gatt_disconnection(self, bluetooth_gatt, gatt_callback): self.log.info("Disconnecting from peripheral device.") try: disconnect_gatt_connection(self.cen_ad, bluetooth_gatt, gatt_callback) close_gatt_client(self.cen_ad, bluetooth_gatt) if bluetooth_gatt in self.bluetooth_gatt_list: self.bluetooth_gatt_list.remove(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False return True def _find_service_added_event(self, gatt_server_cb, uuid): expected_event = gatt_cb_strings['serv_added'].format(gatt_server_cb) try: event = self.per_ad.ed.pop_event(expected_event, self.default_timeout) except Empty: self.log.error( gatt_cb_strings['serv_added_err'].format(expected_event)) return False if event['data']['serviceUuid'].lower() != uuid.lower(): self.log.error("Uuid mismatch. Found: {}, Expected {}.".format( event['data']['serviceUuid'], uuid)) return False return True def _verify_mtu_changed_on_client_and_server( self, expected_mtu, gatt_callback, gatt_server_callback): expected_event = gatt_cb_strings['mtu_changed'].format(gatt_callback) try: mtu_event = self.cen_ad.ed.pop_event(expected_event, self.default_timeout) mtu_size_found = mtu_event['data']['MTU'] if mtu_size_found != expected_mtu: self.log.error("MTU size found: {}, expected: {}".format( mtu_size_found, expected_mtu)) return False except Empty: self.log.error( gatt_cb_err['mtu_changed_err'].format(expected_event)) return False expected_event = gatt_cb_strings['mtu_serv_changed'].format( gatt_server_callback) try: mtu_event = self.per_ad.ed.pop_event(expected_event, self.default_timeout) mtu_size_found = mtu_event['data']['MTU'] if mtu_size_found != expected_mtu: self.log.error("MTU size found: {}, expected: {}".format( mtu_size_found, expected_mtu)) return False except Empty: self.log.error( gatt_cb_err['mtu_serv_changed_err'].format(expected_event)) return False return True @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='8a3530a3-c8bb-466b-9710-99e694c38618') def test_gatt_connect(self): """Test GATT connection over LE. Test establishing a gatt connection between a GATT server and GATT client. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. Disconnect the GATT connection. Expected Result: Verify that a connection was established and then disconnected successfully. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT Priority: 0 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.adv_instances.append(adv_callback) return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='a839b505-03ac-4783-be7e-1d43129a1948') def test_gatt_connect_stop_advertising(self): """Test GATT connection over LE then stop advertising A test case that verifies the GATT connection doesn't disconnect when LE advertisement is stopped. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. Stop the advertiser. 7. Verify no connection state changed happened. 8. Disconnect the GATT connection. Expected Result: Verify that a connection was established and not disconnected when advertisement stops. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT Priority: 0 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.per_ad.droid.bleStopBleAdvertising(adv_callback) try: event = self.cen_ad.ed.pop_event( gatt_cb_strings['gatt_conn_change'].format( gatt_callback, self.default_timeout)) self.log.error( "Connection event found when not expected: {}".format(event)) return False except Empty: self.log.info("No connection state change as expected") try: self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) except Exception as err: self.log.info("Failed to orchestrate disconnect: {}".format(err)) return False return True @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='b82f91a8-54bb-4779-a117-73dc7fdb28cc') def test_gatt_connect_autoconnect(self): """Test GATT connection over LE. Test re-establishing a gatt connection using autoconnect set to True in order to test connection whitelist. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. Disconnect the GATT connection. 7. Create a GATT connection with autoconnect set to True 8. Disconnect the GATT connection. Expected Result: Verify that a connection was re-established and then disconnected successfully. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT Priority: 0 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) autoconnect = False mac_address, adv_callback, scan_callback = ( get_mac_address_of_generic_advertisement(self.cen_ad, self.per_ad)) self.adv_instances.append(adv_callback) try: bluetooth_gatt, gatt_callback = setup_gatt_connection( self.cen_ad, mac_address, autoconnect) self.cen_ad.droid.bleStopBleScan(scan_callback) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False try: disconnect_gatt_connection(self.cen_ad, bluetooth_gatt, gatt_callback) close_gatt_client(self.cen_ad, bluetooth_gatt) if bluetooth_gatt in self.bluetooth_gatt_list: self.bluetooth_gatt_list.remove(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False autoconnect = True bluetooth_gatt = self.cen_ad.droid.gattClientConnectGatt( gatt_callback, mac_address, autoconnect, gatt_transport['auto'], False, gatt_phy_mask['1m_mask']) self.bluetooth_gatt_list.append(bluetooth_gatt) expected_event = gatt_cb_strings['gatt_conn_change'].format( gatt_callback) try: event = self.cen_ad.ed.pop_event(expected_event, self.default_timeout) except Empty: self.log.error( gatt_cb_err['gatt_conn_changed_err'].format(expected_event)) test_result = False return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='e506fa50-7cd9-4bd8-938a-6b85dcfe6bc6') def test_gatt_connect_opportunistic(self): """Test opportunistic GATT connection over LE. Test establishing a gatt connection between a GATT server and GATT client in opportunistic mode. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create GATT connection 1 between the scanner and advertiser normally 6. Create GATT connection 2 between the scanner and advertiser using opportunistic mode 7. Disconnect GATT connection 1 Expected Result: Verify GATT connection 2 automatically disconnects when GATT connection 1 disconnect Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT Priority: 0 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) mac_address, adv_callback, scan_callback = ( get_mac_address_of_generic_advertisement(self.cen_ad, self.per_ad)) # Make GATT connection 1 try: bluetooth_gatt_1, gatt_callback_1 = setup_gatt_connection( self.cen_ad, mac_address, False, transport=gatt_transport['auto'], opportunistic=False) self.cen_ad.droid.bleStopBleScan(scan_callback) self.adv_instances.append(adv_callback) self.bluetooth_gatt_list.append(bluetooth_gatt_1) except GattTestUtilsError as err: self.log.error(err) return False # Make GATT connection 2 try: bluetooth_gatt_2, gatt_callback_2 = setup_gatt_connection( self.cen_ad, mac_address, False, transport=gatt_transport['auto'], opportunistic=True) self.bluetooth_gatt_list.append(bluetooth_gatt_2) except GattTestUtilsError as err: self.log.error(err) return False # Disconnect GATT connection 1 try: disconnect_gatt_connection(self.cen_ad, bluetooth_gatt_1, gatt_callback_1) close_gatt_client(self.cen_ad, bluetooth_gatt_1) if bluetooth_gatt_1 in self.bluetooth_gatt_list: self.bluetooth_gatt_list.remove(bluetooth_gatt_1) except GattTestUtilsError as err: self.log.error(err) return False # Confirm that GATT connection 2 also disconnects wait_for_gatt_disconnect_event(self.cen_ad, gatt_callback_2) close_gatt_client(self.cen_ad, bluetooth_gatt_2) if bluetooth_gatt_2 in self.bluetooth_gatt_list: self.bluetooth_gatt_list.remove(bluetooth_gatt_2) return True @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='1e01838e-c4de-4720-9adf-9e0419378226') def test_gatt_request_min_mtu(self): """Test GATT connection over LE and exercise MTU sizes. Test establishing a gatt connection between a GATT server and GATT client. Request an MTU size that matches the correct minimum size. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. From the scanner (client) request MTU size change to the minimum value. 7. Find the MTU changed event on the client. 8. Disconnect the GATT connection. Expected Result: Verify that a connection was established and the MTU value found matches the expected MTU value. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, MTU Priority: 0 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.adv_instances.append(adv_callback) expected_mtu = gatt_mtu_size['min'] self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt, expected_mtu) if not self._verify_mtu_changed_on_client_and_server( expected_mtu, gatt_callback, gatt_server_cb): return False return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='c1fa3a2d-fb47-47db-bdd1-458928cd6a5f') def test_gatt_request_max_mtu(self): """Test GATT connection over LE and exercise MTU sizes. Test establishing a gatt connection between a GATT server and GATT client. Request an MTU size that matches the correct maximum size. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. From the scanner (client) request MTU size change to the maximum value. 7. Find the MTU changed event on the client. 8. Disconnect the GATT connection. Expected Result: Verify that a connection was established and the MTU value found matches the expected MTU value. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, MTU Priority: 0 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.adv_instances.append(adv_callback) expected_mtu = gatt_mtu_size['max'] self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt, expected_mtu) if not self._verify_mtu_changed_on_client_and_server( expected_mtu, gatt_callback, gatt_server_cb): return False return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='4416d483-dec3-46cb-8038-4d82620f873a') def test_gatt_request_out_of_bounds_mtu(self): """Test GATT connection over LE and exercise an out of bound MTU size. Test establishing a gatt connection between a GATT server and GATT client. Request an MTU size that is the MIN value minus 1. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. From the scanner (client) request MTU size change to the minimum value minus one. 7. Find the MTU changed event on the client. 8. Disconnect the GATT connection. Expected Result: Verify that an MTU changed event was not discovered and that it didn't cause an exception when requesting an out of bounds MTU. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, MTU Priority: 0 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.adv_instances.append(adv_callback) unexpected_mtu = gatt_mtu_size['min'] - 1 self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt, unexpected_mtu) if self._verify_mtu_changed_on_client_and_server( unexpected_mtu, gatt_callback, gatt_server_cb): return False return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='31ffb9ca-cc75-43fb-9802-c19f1c5856b6') def test_gatt_connect_trigger_on_read_rssi(self): """Test GATT connection over LE read RSSI. Test establishing a gatt connection between a GATT server and GATT client then read the RSSI. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. From the scanner, request to read the RSSI of the advertiser. 7. Disconnect the GATT connection. Expected Result: Verify that a connection was established and then disconnected successfully. Verify that the RSSI was ready correctly. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, RSSI Priority: 1 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.adv_instances.append(adv_callback) expected_event = gatt_cb_strings['rd_remote_rssi'].format( gatt_callback) if self.cen_ad.droid.gattClientReadRSSI(bluetooth_gatt): try: self.cen_ad.ed.pop_event(expected_event, self.default_timeout) except Empty: self.log.error( gatt_cb_err['rd_remote_rssi_err'].format(expected_event)) return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='dee9ef28-b872-428a-821b-cc62f27ba936') def test_gatt_connect_trigger_on_services_discovered(self): """Test GATT connection and discover services of peripheral. Test establishing a gatt connection between a GATT server and GATT client the discover all services from the connected device. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. From the scanner (central device), discover services. 7. Disconnect the GATT connection. Expected Result: Verify that a connection was established and then disconnected successfully. Verify that the service were discovered. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, Services Priority: 1 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.adv_instances.append(adv_callback) if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt): expected_event = gatt_cb_strings['gatt_serv_disc'].format( gatt_callback) try: event = self.cen_ad.ed.pop_event(expected_event, self.default_timeout) except Empty: self.log.error( gatt_cb_err['gatt_serv_disc'].format(expected_event)) return False return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='01883bdd-0cf8-48fb-bf15-467bbd4f065b') def test_gatt_connect_trigger_on_services_discovered_iterate_attributes( self): """Test GATT connection and iterate peripherals attributes. Test establishing a gatt connection between a GATT server and GATT client and iterate over all the characteristics and descriptors of the discovered services. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. From the scanner (central device), discover services. 7. Iterate over all the characteristics and descriptors of the discovered features. 8. Disconnect the GATT connection. Expected Result: Verify that a connection was established and then disconnected successfully. Verify that the services, characteristics, and descriptors were discovered. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, Services Characteristics, Descriptors Priority: 1 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.adv_instances.append(adv_callback) if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt): expected_event = gatt_cb_strings['gatt_serv_disc'].format( gatt_callback) try: event = self.cen_ad.ed.pop_event(expected_event, self.default_timeout) discovered_services_index = event['data']['ServicesIndex'] except Empty: self.log.error( gatt_cb_err['gatt_serv_disc'].format(expected_event)) return False log_gatt_server_uuids(self.cen_ad, discovered_services_index) return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='d4277bee-da99-4f48-8a4d-f81b5389da18') def test_gatt_connect_with_service_uuid_variations(self): """Test GATT connection with multiple service uuids. Test establishing a gatt connection between a GATT server and GATT client with multiple service uuid variations. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. From the scanner (central device), discover services. 7. Verify that all the service uuid variations are found. 8. Disconnect the GATT connection. Expected Result: Verify that a connection was established and then disconnected successfully. Verify that the service uuid variations are found. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, Services Priority: 2 """ try: gatt_server_cb, gatt_server = setup_multiple_services(self.per_ad) self.gatt_server_list.append(gatt_server) except GattTestUtilsError as err: self.log.error(err) return False try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False self.adv_instances.append(adv_callback) if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt): expected_event = gatt_cb_strings['gatt_serv_disc'].format( gatt_callback) try: event = self.cen_ad.ed.pop_event(expected_event, self.default_timeout) except Empty: self.log.error( gatt_cb_err['gatt_serv_disc'].format(expected_event)) return False discovered_services_index = event['data']['ServicesIndex'] log_gatt_server_uuids(self.cen_ad, discovered_services_index) return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='7d3442c5-f71f-44ae-bd35-f2569f01b3b8') def test_gatt_connect_in_quick_succession(self): """Test GATT connections multiple times. Test establishing a gatt connection between a GATT server and GATT client with multiple iterations. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 6. Disconnect the GATT connection. 7. Repeat steps 5 and 6 twenty times. Expected Result: Verify that a connection was established and then disconnected successfully twenty times. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, Stress Priority: 1 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) mac_address, adv_callback, scan_callback = get_mac_address_of_generic_advertisement( self.cen_ad, self.per_ad) autoconnect = False for i in range(1000): self.log.info("Starting connection iteration {}".format(i + 1)) try: bluetooth_gatt, gatt_callback = setup_gatt_connection( self.cen_ad, mac_address, autoconnect) self.cen_ad.droid.bleStopBleScan(scan_callback) except GattTestUtilsError as err: self.log.error(err) return False test_result = self._orchestrate_gatt_disconnection( bluetooth_gatt, gatt_callback) if not test_result: self.log.info("Failed to disconnect from peripheral device.") return False self.adv_instances.append(adv_callback) return True @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='148469d9-7ab0-4c08-b2e9-7e49e88da1fc') def test_gatt_connect_mitm_attack(self): """Test GATT connection with permission write encrypted mitm. Test establishing a gatt connection between a GATT server and GATT client while the GATT server's characteristic includes the property write value and the permission write encrypted mitm value. This will prompt LE pairing and then the devices will create a bond. Steps: 1. Create a GATT server and server callback on the peripheral device. 2. Create a unique service and characteristic uuid on the peripheral. 3. Create a characteristic on the peripheral with these properties: gatt_characteristic['property_write'], gatt_characteristic['permission_write_encrypted_mitm'] 4. Create a GATT service on the peripheral. 5. Add the characteristic to the GATT service. 6. Create a GATT connection between your central and peripheral device. 7. From the central device, discover the peripheral's services. 8. Iterate the services found until you find the unique characteristic created in step 3. 9. Once found, write a random but valid value to the characteristic. 10. Start pairing helpers on both devices immediately after attempting to write to the characteristic. 11. Within 10 seconds of writing the characteristic, there should be a prompt to bond the device from the peripheral. The helpers will handle the UI interaction automatically. (see BluetoothConnectionFacade.java bluetoothStartPairingHelper). 12. Verify that the two devices are bonded. Expected Result: Verify that a connection was established and the devices are bonded. Returns: Pass if True Fail if False TAGS: LE, Advertising, Filtering, Scanning, GATT, Characteristic, MITM Priority: 1 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) service_uuid = "3846D7A0-69C8-11E4-BA00-0002A5D5C51B" test_uuid = "aa7edd5a-4d1d-4f0e-883a-d145616a1630" bonded = False characteristic = self.per_ad.droid.gattServerCreateBluetoothGattCharacteristic( test_uuid, gatt_characteristic['property_write'], gatt_characteristic['permission_write_encrypted_mitm']) gatt_service = self.per_ad.droid.gattServerCreateService( service_uuid, gatt_service_types['primary']) self.per_ad.droid.gattServerAddCharacteristicToService( gatt_service, characteristic) self.per_ad.droid.gattServerAddService(gatt_server, gatt_service) result = self._find_service_added_event(gatt_server_cb, service_uuid) if not result: return False bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) self.adv_instances.append(adv_callback) if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt): expected_event = gatt_cb_strings['gatt_serv_disc'].format( gatt_callback) try: event = self.cen_ad.ed.pop_event(expected_event, self.default_timeout) except Empty: self.log.error( gatt_cb_err['gatt_serv_disc'].format(expected_event)) return False discovered_services_index = event['data']['ServicesIndex'] else: self.log.info("Failed to discover services.") return False test_value = [1, 2, 3, 4, 5, 6, 7] services_count = self.cen_ad.droid.gattClientGetDiscoveredServicesCount( discovered_services_index) for i in range(services_count): characteristic_uuids = ( self.cen_ad.droid.gattClientGetDiscoveredCharacteristicUuids( discovered_services_index, i)) for characteristic_uuid in characteristic_uuids: if characteristic_uuid == test_uuid: self.cen_ad.droid.bluetoothStartPairingHelper() self.per_ad.droid.bluetoothStartPairingHelper() self.cen_ad.droid.gattClientCharacteristicSetValue( bluetooth_gatt, discovered_services_index, i, characteristic_uuid, test_value) self.cen_ad.droid.gattClientWriteCharacteristic( bluetooth_gatt, discovered_services_index, i, characteristic_uuid) start_time = time.time() + self.default_timeout target_name = self.per_ad.droid.bluetoothGetLocalName() while time.time() < start_time and bonded == False: bonded_devices = \ self.cen_ad.droid.bluetoothGetBondedDevices() for device in bonded_devices: if ('name' in device.keys() and device['name'] == target_name): bonded = True break bonded = False target_name = self.cen_ad.droid.bluetoothGetLocalName() while time.time() < start_time and bonded == False: bonded_devices = \ self.per_ad.droid.bluetoothGetBondedDevices() for device in bonded_devices: if ('name' in device.keys() and device['name'] == target_name): bonded = True break # Dual mode devices will establish connection over the classic transport, # in order to establish bond over both transports, and do SDP. Starting # disconnection before all this is finished is not safe, might lead to # race conditions, i.e. bond over classic tranport shows up after LE # bond is already removed. time.sleep(4) for ad in [self.cen_ad, self.per_ad]: if not clear_bonded_devices(ad): return False # Necessary sleep time for entries to update unbonded state time.sleep(2) for ad in [self.cen_ad, self.per_ad]: bonded_devices = ad.droid.bluetoothGetBondedDevices() if len(bonded_devices) > 0: self.log.error( "Failed to unbond devices: {}".format(bonded_devices)) return False return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='cc3fc361-7bf1-4ee2-9e46-4a27c88ce6a8') def test_gatt_connect_get_connected_devices(self): """Test GATT connections show up in getConnectedDevices Test establishing a gatt connection between a GATT server and GATT client. Verify that active connections show up using BluetoothManager.getConnectedDevices API. Steps: 1. Start a generic advertisement. 2. Start a generic scanner. 3. Find the advertisement and extract the mac address. 4. Stop the first scanner. 5. Create a GATT connection between the scanner and advertiser. 7. Verify the GATT Client has an open connection to the GATT Server. 8. Verify the GATT Server has an open connection to the GATT Client. 9. Disconnect the GATT connection. Expected Result: Verify that a connection was established, connected devices are found on both the central and peripheral devices, and then disconnected successfully. Returns: Pass if True Fail if False TAGS: LE, Advertising, Scanning, GATT Priority: 2 """ gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) self.gatt_server_list.append(gatt_server) try: bluetooth_gatt, gatt_callback, adv_callback = ( orchestrate_gatt_connection(self.cen_ad, self.per_ad)) self.bluetooth_gatt_list.append(bluetooth_gatt) except GattTestUtilsError as err: self.log.error(err) return False conn_cen_devices = self.cen_ad.droid.bluetoothGetConnectedLeDevices( bt_profile_constants['gatt']) conn_per_devices = self.per_ad.droid.bluetoothGetConnectedLeDevices( bt_profile_constants['gatt_server']) target_name = self.per_ad.droid.bluetoothGetLocalName() error_message = ("Connected device {} not found in list of connected " "devices {}") if not any(d['name'] == target_name for d in conn_cen_devices): self.log.error(error_message.format(target_name, conn_cen_devices)) return False # For the GATT server only check the size of the list since # it may or may not include the device name. target_name = self.cen_ad.droid.bluetoothGetLocalName() if not conn_per_devices: self.log.error(error_message.format(target_name, conn_per_devices)) return False self.adv_instances.append(adv_callback) return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) @BluetoothBaseTest.bt_test_wrap @test_tracker_info(uuid='a0a37ca6-9fa8-4d35-9fdb-0e25b4b8a363') def test_gatt_connect_second_adv_after_canceling_first_adv(self): """Test GATT connection to peripherals second advertising address. The the ability of cancelling GATT connections and trying to reconnect to the same device via a different address. Steps: 1. A starts advertising 2. B starts scanning and finds A's mac address 3. Stop advertisement from step 1. Start a new advertisement on A and find the new new mac address, B knows of both old and new address. 4. B1 sends connect request to old address of A 5. B1 cancel connect attempt after 10 seconds 6. B1 sends connect request to new address of A 7. Verify B1 establish connection to A in less than 10 seconds Expected Result: Verify that a connection was established only on the second advertisement's mac address. Returns: Pass if True Fail if False TAGS: LE, Advertising, Scanning, GATT Priority: 3 """ autoconnect = False transport = gatt_transport['auto'] opportunistic = False # Setup a basic Gatt server on the peripheral gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() gatt_server = self.per_ad.droid.gattServerOpenGattServer( gatt_server_cb) # Set advertisement settings to include local name in advertisement # and set the advertising mode to low_latency. self.per_ad.droid.bleSetAdvertiseSettingsIsConnectable(True) self.per_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True) self.per_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( ble_advertise_settings_modes['low_latency']) # Setup necessary advertisement objects. advertise_data = self.per_ad.droid.bleBuildAdvertiseData() advertise_settings = self.per_ad.droid.bleBuildAdvertiseSettings() advertise_callback = self.per_ad.droid.bleGenBleAdvertiseCallback() # Step 1: Start advertisement self.per_ad.droid.bleStartBleAdvertising( advertise_callback, advertise_data, advertise_settings) # Setup scan settings for low_latency scanning and to include the local name # of the advertisement started in step 1. filter_list = self.cen_ad.droid.bleGenFilterList() self.cen_ad.droid.bleSetScanSettingsNumOfMatches( ble_scan_settings_match_nums['one']) self.cen_ad.droid.bleSetScanFilterDeviceName( self.per_ad.droid.bluetoothGetLocalName()) self.cen_ad.droid.bleBuildScanFilter(filter_list) self.cen_ad.droid.bleSetScanSettingsScanMode( ble_scan_settings_modes['low_latency']) # Setup necessary scan objects. scan_settings = self.cen_ad.droid.bleBuildScanSetting() scan_callback = self.cen_ad.droid.bleGenScanCallback() # Step 2: Start scanning on central Android device and find peripheral # address. self.cen_ad.droid.bleStartBleScan(filter_list, scan_settings, scan_callback) expected_event_name = scan_result.format(scan_callback) try: mac_address_pre_restart = self.cen_ad.ed.pop_event( expected_event_name, self.default_timeout)['data']['Result'][ 'deviceInfo']['address'] self.log.info( "Peripheral advertisement found with mac address: {}".format( mac_address_pre_restart)) except Empty: self.log.info("Peripheral advertisement not found") test_result = False # Step 3: Restart peripheral advertising such that a new mac address is # created. self.per_ad.droid.bleStopBleAdvertising(advertise_callback) self.per_ad.droid.bleStartBleAdvertising( advertise_callback, advertise_data, advertise_settings) mac_address_post_restart = mac_address_pre_restart while True: try: mac_address_post_restart = self.cen_ad.ed.pop_event( expected_event_name, self.default_timeout)['data']['Result'][ 'deviceInfo']['address'] self.log.info( "Peripheral advertisement found with mac address: {}".format( mac_address_post_restart)) except Empty: self.log.info("Peripheral advertisement not found") test_result = False if mac_address_pre_restart != mac_address_post_restart: break self.cen_ad.droid.bleStopBleScan(scan_callback) # Steps 4: Try to connect to the first mac address gatt_callback = self.cen_ad.droid.gattCreateGattCallback() self.log.info( "Gatt Connect to mac address {}.".format(mac_address_pre_restart)) bluetooth_gatt = self.cen_ad.droid.gattClientConnectGatt( gatt_callback, mac_address_pre_restart, autoconnect, transport, opportunistic, gatt_phy_mask['1m_mask']) self.bluetooth_gatt_list.append(bluetooth_gatt) expected_event = gatt_cb_strings['gatt_conn_change'].format( gatt_callback) try: event = self.cen_ad.ed.pop_event(expected_event, self.default_timeout) self.log.error( "Connection callback updated unexpectedly: {}".format(event)) return False except Empty: self.log.info("No connection update as expected.") # Step 5: Cancel connection request. self.cen_ad.droid.gattClientDisconnect(bluetooth_gatt) close_gatt_client(self.cen_ad, bluetooth_gatt) if bluetooth_gatt in self.bluetooth_gatt_list: self.bluetooth_gatt_list.remove(bluetooth_gatt) # Step 6: Connect to second mac address. gatt_callback = self.cen_ad.droid.gattCreateGattCallback() self.log.info( "Gatt Connect to mac address {}.".format(mac_address_post_restart)) bluetooth_gatt = self.cen_ad.droid.gattClientConnectGatt( gatt_callback, mac_address_post_restart, autoconnect, transport, opportunistic, gatt_phy_mask['1m_mask']) self.bluetooth_gatt_list.append(bluetooth_gatt) expected_event = gatt_cb_strings['gatt_conn_change'].format( gatt_callback) # Step 7: Verify connection was setup successfully. try: event = self.cen_ad.ed.pop_event(expected_event, self.default_timeout) self.log.info( "Connection callback updated successfully: {}".format(event)) if event['data']['State'] != gatt_connection_state['connected']: self.log.error( "Could not establish a connection to the peripheral.") return False except Empty: self.log.error("No connection update was found.") return False self.per_ad.droid.bleStopBleAdvertising(advertise_callback) return self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback)