1#!/usr/bin/env python3 2# 3# Copyright (C) 2016 The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); you may not 6# use this file except in compliance with the License. You may obtain a copy of 7# the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 14# License for the specific language governing permissions and limitations under 15# the License. 16""" 17This test script exercises different GATT connection tests. 18""" 19 20import pprint 21from queue import Empty 22import time 23 24from acts.test_decorators import test_tracker_info 25from acts_contrib.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest 26from acts_contrib.test_utils.bt.bt_constants import ble_advertise_settings_modes 27from acts_contrib.test_utils.bt.bt_constants import ble_scan_settings_modes 28from acts_contrib.test_utils.bt.bt_constants import ble_scan_settings_match_nums 29from acts_contrib.test_utils.bt.bt_constants import bt_profile_constants 30from acts_contrib.test_utils.bt.bt_constants import gatt_characteristic 31from acts_contrib.test_utils.bt.bt_constants import gatt_descriptor 32from acts_contrib.test_utils.bt.bt_constants import gatt_service_types 33from acts_contrib.test_utils.bt.bt_constants import gatt_cb_err 34from acts_contrib.test_utils.bt.bt_constants import gatt_cb_strings 35from acts_contrib.test_utils.bt.bt_constants import gatt_connection_state 36from acts_contrib.test_utils.bt.bt_constants import gatt_mtu_size 37from acts_contrib.test_utils.bt.bt_constants import gatt_phy_mask 38from acts_contrib.test_utils.bt.bt_constants import gatt_transport 39from acts_contrib.test_utils.bt.bt_constants import scan_result 40from acts_contrib.test_utils.bt.bt_gatt_utils import GattTestUtilsError 41from acts_contrib.test_utils.bt.bt_gatt_utils import disconnect_gatt_connection 42from acts_contrib.test_utils.bt.bt_gatt_utils import wait_for_gatt_disconnect_event 43from acts_contrib.test_utils.bt.bt_gatt_utils import close_gatt_client 44from acts_contrib.test_utils.bt.bt_gatt_utils import log_gatt_server_uuids 45from acts_contrib.test_utils.bt.bt_gatt_utils import orchestrate_gatt_connection 46from acts_contrib.test_utils.bt.bt_gatt_utils import setup_gatt_connection 47from acts_contrib.test_utils.bt.bt_gatt_utils import setup_multiple_services 48from acts_contrib.test_utils.bt.bt_test_utils import get_mac_address_of_generic_advertisement 49from acts_contrib.test_utils.bt.bt_test_utils import clear_bonded_devices 50 51PHYSICAL_DISCONNECT_TIMEOUT = 5 52 53 54class GattConnectTest(BluetoothBaseTest): 55 adv_instances = [] 56 bluetooth_gatt_list = [] 57 gatt_server_list = [] 58 default_timeout = 10 59 default_discovery_timeout = 3 60 61 def setup_class(self): 62 super().setup_class() 63 self.cen_ad = self.android_devices[0] 64 self.per_ad = self.android_devices[1] 65 66 def setup_test(self): 67 super(BluetoothBaseTest, self).setup_test() 68 bluetooth_gatt_list = [] 69 self.gatt_server_list = [] 70 self.adv_instances = [] 71 # Ensure there is ample time for a physical disconnect in between 72 # testcases. 73 self.log.info( 74 "Waiting for {} seconds for physical GATT disconnections".format( 75 PHYSICAL_DISCONNECT_TIMEOUT)) 76 time.sleep(PHYSICAL_DISCONNECT_TIMEOUT) 77 78 def teardown_test(self): 79 for bluetooth_gatt in self.bluetooth_gatt_list: 80 self.cen_ad.droid.gattClientClose(bluetooth_gatt) 81 for gatt_server in self.gatt_server_list: 82 self.per_ad.droid.gattServerClose(gatt_server) 83 for adv in self.adv_instances: 84 self.per_ad.droid.bleStopBleAdvertising(adv) 85 return True 86 87 def _orchestrate_gatt_disconnection(self, bluetooth_gatt, gatt_callback): 88 self.log.info("Disconnecting from peripheral device.") 89 try: 90 disconnect_gatt_connection(self.cen_ad, bluetooth_gatt, 91 gatt_callback) 92 close_gatt_client(self.cen_ad, bluetooth_gatt) 93 if bluetooth_gatt in self.bluetooth_gatt_list: 94 self.bluetooth_gatt_list.remove(bluetooth_gatt) 95 except GattTestUtilsError as err: 96 self.log.error(err) 97 return False 98 return True 99 100 def _find_service_added_event(self, gatt_server_cb, uuid): 101 expected_event = gatt_cb_strings['serv_added'].format(gatt_server_cb) 102 try: 103 event = self.per_ad.ed.pop_event(expected_event, 104 self.default_timeout) 105 except Empty: 106 self.log.error( 107 gatt_cb_strings['serv_added_err'].format(expected_event)) 108 return False 109 if event['data']['serviceUuid'].lower() != uuid.lower(): 110 self.log.error("Uuid mismatch. Found: {}, Expected {}.".format( 111 event['data']['serviceUuid'], uuid)) 112 return False 113 return True 114 115 def _verify_mtu_changed_on_client_and_server( 116 self, expected_mtu, gatt_callback, gatt_server_callback): 117 expected_event = gatt_cb_strings['mtu_changed'].format(gatt_callback) 118 try: 119 mtu_event = self.cen_ad.ed.pop_event(expected_event, 120 self.default_timeout) 121 mtu_size_found = mtu_event['data']['MTU'] 122 if mtu_size_found != expected_mtu: 123 self.log.error("MTU size found: {}, expected: {}".format( 124 mtu_size_found, expected_mtu)) 125 return False 126 except Empty: 127 self.log.error( 128 gatt_cb_err['mtu_changed_err'].format(expected_event)) 129 return False 130 131 expected_event = gatt_cb_strings['mtu_serv_changed'].format( 132 gatt_server_callback) 133 try: 134 mtu_event = self.per_ad.ed.pop_event(expected_event, 135 self.default_timeout) 136 mtu_size_found = mtu_event['data']['MTU'] 137 if mtu_size_found != expected_mtu: 138 self.log.error("MTU size found: {}, expected: {}".format( 139 mtu_size_found, expected_mtu)) 140 return False 141 except Empty: 142 self.log.error( 143 gatt_cb_err['mtu_serv_changed_err'].format(expected_event)) 144 return False 145 return True 146 147 @BluetoothBaseTest.bt_test_wrap 148 @test_tracker_info(uuid='8a3530a3-c8bb-466b-9710-99e694c38618') 149 def test_gatt_connect(self): 150 """Test GATT connection over LE. 151 152 Test establishing a gatt connection between a GATT server and GATT 153 client. 154 155 Steps: 156 1. Start a generic advertisement. 157 2. Start a generic scanner. 158 3. Find the advertisement and extract the mac address. 159 4. Stop the first scanner. 160 5. Create a GATT connection between the scanner and advertiser. 161 6. Disconnect the GATT connection. 162 163 Expected Result: 164 Verify that a connection was established and then disconnected 165 successfully. 166 167 Returns: 168 Pass if True 169 Fail if False 170 171 TAGS: LE, Advertising, Filtering, Scanning, GATT 172 Priority: 0 173 """ 174 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 175 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 176 gatt_server_cb) 177 self.gatt_server_list.append(gatt_server) 178 try: 179 bluetooth_gatt, gatt_callback, adv_callback = ( 180 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 181 self.bluetooth_gatt_list.append(bluetooth_gatt) 182 except GattTestUtilsError as err: 183 self.log.error(err) 184 return False 185 self.adv_instances.append(adv_callback) 186 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 187 gatt_callback) 188 189 @BluetoothBaseTest.bt_test_wrap 190 @test_tracker_info(uuid='a839b505-03ac-4783-be7e-1d43129a1948') 191 def test_gatt_connect_stop_advertising(self): 192 """Test GATT connection over LE then stop advertising 193 194 A test case that verifies the GATT connection doesn't 195 disconnect when LE advertisement is stopped. 196 197 Steps: 198 1. Start a generic advertisement. 199 2. Start a generic scanner. 200 3. Find the advertisement and extract the mac address. 201 4. Stop the first scanner. 202 5. Create a GATT connection between the scanner and advertiser. 203 6. Stop the advertiser. 204 7. Verify no connection state changed happened. 205 8. Disconnect the GATT connection. 206 207 Expected Result: 208 Verify that a connection was established and not disconnected 209 when advertisement stops. 210 211 Returns: 212 Pass if True 213 Fail if False 214 215 TAGS: LE, Advertising, Filtering, Scanning, GATT 216 Priority: 0 217 """ 218 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 219 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 220 gatt_server_cb) 221 self.gatt_server_list.append(gatt_server) 222 try: 223 bluetooth_gatt, gatt_callback, adv_callback = ( 224 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 225 self.bluetooth_gatt_list.append(bluetooth_gatt) 226 except GattTestUtilsError as err: 227 self.log.error(err) 228 return False 229 self.per_ad.droid.bleStopBleAdvertising(adv_callback) 230 try: 231 event = self.cen_ad.ed.pop_event( 232 gatt_cb_strings['gatt_conn_change'].format( 233 gatt_callback, self.default_timeout)) 234 self.log.error( 235 "Connection event found when not expected: {}".format(event)) 236 return False 237 except Empty: 238 self.log.info("No connection state change as expected") 239 try: 240 self._orchestrate_gatt_disconnection(bluetooth_gatt, gatt_callback) 241 except Exception as err: 242 self.log.info("Failed to orchestrate disconnect: {}".format(err)) 243 return False 244 return True 245 246 @BluetoothBaseTest.bt_test_wrap 247 @test_tracker_info(uuid='b82f91a8-54bb-4779-a117-73dc7fdb28cc') 248 def test_gatt_connect_autoconnect(self): 249 """Test GATT connection over LE. 250 251 Test re-establishing a gatt connection using autoconnect 252 set to True in order to test connection whitelist. 253 254 Steps: 255 1. Start a generic advertisement. 256 2. Start a generic scanner. 257 3. Find the advertisement and extract the mac address. 258 4. Stop the first scanner. 259 5. Create a GATT connection between the scanner and advertiser. 260 6. Disconnect the GATT connection. 261 7. Create a GATT connection with autoconnect set to True 262 8. Disconnect the GATT connection. 263 264 Expected Result: 265 Verify that a connection was re-established and then disconnected 266 successfully. 267 268 Returns: 269 Pass if True 270 Fail if False 271 272 TAGS: LE, Advertising, Filtering, Scanning, GATT 273 Priority: 0 274 """ 275 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 276 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 277 gatt_server_cb) 278 self.gatt_server_list.append(gatt_server) 279 autoconnect = False 280 mac_address, adv_callback, scan_callback = ( 281 get_mac_address_of_generic_advertisement(self.cen_ad, self.per_ad)) 282 self.adv_instances.append(adv_callback) 283 try: 284 bluetooth_gatt, gatt_callback = setup_gatt_connection( 285 self.cen_ad, mac_address, autoconnect) 286 self.cen_ad.droid.bleStopBleScan(scan_callback) 287 self.bluetooth_gatt_list.append(bluetooth_gatt) 288 except GattTestUtilsError as err: 289 self.log.error(err) 290 return False 291 try: 292 disconnect_gatt_connection(self.cen_ad, bluetooth_gatt, 293 gatt_callback) 294 close_gatt_client(self.cen_ad, bluetooth_gatt) 295 if bluetooth_gatt in self.bluetooth_gatt_list: 296 self.bluetooth_gatt_list.remove(bluetooth_gatt) 297 except GattTestUtilsError as err: 298 self.log.error(err) 299 return False 300 autoconnect = True 301 bluetooth_gatt = self.cen_ad.droid.gattClientConnectGatt( 302 gatt_callback, mac_address, autoconnect, gatt_transport['auto'], 303 False, gatt_phy_mask['1m_mask']) 304 self.bluetooth_gatt_list.append(bluetooth_gatt) 305 expected_event = gatt_cb_strings['gatt_conn_change'].format( 306 gatt_callback) 307 try: 308 event = self.cen_ad.ed.pop_event(expected_event, 309 self.default_timeout) 310 except Empty: 311 self.log.error( 312 gatt_cb_err['gatt_conn_changed_err'].format(expected_event)) 313 test_result = False 314 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 315 gatt_callback) 316 317 @BluetoothBaseTest.bt_test_wrap 318 @test_tracker_info(uuid='e506fa50-7cd9-4bd8-938a-6b85dcfe6bc6') 319 def test_gatt_connect_opportunistic(self): 320 """Test opportunistic GATT connection over LE. 321 322 Test establishing a gatt connection between a GATT server and GATT 323 client in opportunistic mode. 324 325 Steps: 326 1. Start a generic advertisement. 327 2. Start a generic scanner. 328 3. Find the advertisement and extract the mac address. 329 4. Stop the first scanner. 330 5. Create GATT connection 1 between the scanner and advertiser normally 331 6. Create GATT connection 2 between the scanner and advertiser using 332 opportunistic mode 333 7. Disconnect GATT connection 1 334 335 Expected Result: 336 Verify GATT connection 2 automatically disconnects when GATT connection 337 1 disconnect 338 339 Returns: 340 Pass if True 341 Fail if False 342 343 TAGS: LE, Advertising, Filtering, Scanning, GATT 344 Priority: 0 345 """ 346 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 347 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 348 gatt_server_cb) 349 self.gatt_server_list.append(gatt_server) 350 mac_address, adv_callback, scan_callback = ( 351 get_mac_address_of_generic_advertisement(self.cen_ad, self.per_ad)) 352 # Make GATT connection 1 353 try: 354 bluetooth_gatt_1, gatt_callback_1 = setup_gatt_connection( 355 self.cen_ad, 356 mac_address, 357 False, 358 transport=gatt_transport['auto'], 359 opportunistic=False) 360 self.cen_ad.droid.bleStopBleScan(scan_callback) 361 self.adv_instances.append(adv_callback) 362 self.bluetooth_gatt_list.append(bluetooth_gatt_1) 363 except GattTestUtilsError as err: 364 self.log.error(err) 365 return False 366 # Make GATT connection 2 367 try: 368 bluetooth_gatt_2, gatt_callback_2 = setup_gatt_connection( 369 self.cen_ad, 370 mac_address, 371 False, 372 transport=gatt_transport['auto'], 373 opportunistic=True) 374 self.bluetooth_gatt_list.append(bluetooth_gatt_2) 375 except GattTestUtilsError as err: 376 self.log.error(err) 377 return False 378 # Disconnect GATT connection 1 379 try: 380 disconnect_gatt_connection(self.cen_ad, bluetooth_gatt_1, 381 gatt_callback_1) 382 close_gatt_client(self.cen_ad, bluetooth_gatt_1) 383 if bluetooth_gatt_1 in self.bluetooth_gatt_list: 384 self.bluetooth_gatt_list.remove(bluetooth_gatt_1) 385 except GattTestUtilsError as err: 386 self.log.error(err) 387 return False 388 # Confirm that GATT connection 2 also disconnects 389 wait_for_gatt_disconnect_event(self.cen_ad, gatt_callback_2) 390 close_gatt_client(self.cen_ad, bluetooth_gatt_2) 391 if bluetooth_gatt_2 in self.bluetooth_gatt_list: 392 self.bluetooth_gatt_list.remove(bluetooth_gatt_2) 393 return True 394 395 @BluetoothBaseTest.bt_test_wrap 396 @test_tracker_info(uuid='1e01838e-c4de-4720-9adf-9e0419378226') 397 def test_gatt_request_min_mtu(self): 398 """Test GATT connection over LE and exercise MTU sizes. 399 400 Test establishing a gatt connection between a GATT server and GATT 401 client. Request an MTU size that matches the correct minimum size. 402 403 Steps: 404 1. Start a generic advertisement. 405 2. Start a generic scanner. 406 3. Find the advertisement and extract the mac address. 407 4. Stop the first scanner. 408 5. Create a GATT connection between the scanner and advertiser. 409 6. From the scanner (client) request MTU size change to the 410 minimum value. 411 7. Find the MTU changed event on the client. 412 8. Disconnect the GATT connection. 413 414 Expected Result: 415 Verify that a connection was established and the MTU value found 416 matches the expected MTU value. 417 418 Returns: 419 Pass if True 420 Fail if False 421 422 TAGS: LE, Advertising, Filtering, Scanning, GATT, MTU 423 Priority: 0 424 """ 425 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 426 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 427 gatt_server_cb) 428 self.gatt_server_list.append(gatt_server) 429 try: 430 bluetooth_gatt, gatt_callback, adv_callback = ( 431 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 432 self.bluetooth_gatt_list.append(bluetooth_gatt) 433 except GattTestUtilsError as err: 434 self.log.error(err) 435 return False 436 self.adv_instances.append(adv_callback) 437 expected_mtu = gatt_mtu_size['min'] 438 self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt, expected_mtu) 439 if not self._verify_mtu_changed_on_client_and_server( 440 expected_mtu, gatt_callback, gatt_server_cb): 441 return False 442 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 443 gatt_callback) 444 445 @BluetoothBaseTest.bt_test_wrap 446 @test_tracker_info(uuid='c1fa3a2d-fb47-47db-bdd1-458928cd6a5f') 447 def test_gatt_request_max_mtu(self): 448 """Test GATT connection over LE and exercise MTU sizes. 449 450 Test establishing a gatt connection between a GATT server and GATT 451 client. Request an MTU size that matches the correct maximum size. 452 453 Steps: 454 1. Start a generic advertisement. 455 2. Start a generic scanner. 456 3. Find the advertisement and extract the mac address. 457 4. Stop the first scanner. 458 5. Create a GATT connection between the scanner and advertiser. 459 6. From the scanner (client) request MTU size change to the 460 maximum value. 461 7. Find the MTU changed event on the client. 462 8. Disconnect the GATT connection. 463 464 Expected Result: 465 Verify that a connection was established and the MTU value found 466 matches the expected MTU value. 467 468 Returns: 469 Pass if True 470 Fail if False 471 472 TAGS: LE, Advertising, Filtering, Scanning, GATT, MTU 473 Priority: 0 474 """ 475 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 476 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 477 gatt_server_cb) 478 self.gatt_server_list.append(gatt_server) 479 try: 480 bluetooth_gatt, gatt_callback, adv_callback = ( 481 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 482 self.bluetooth_gatt_list.append(bluetooth_gatt) 483 except GattTestUtilsError as err: 484 self.log.error(err) 485 return False 486 self.adv_instances.append(adv_callback) 487 expected_mtu = gatt_mtu_size['max'] 488 self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt, expected_mtu) 489 if not self._verify_mtu_changed_on_client_and_server( 490 expected_mtu, gatt_callback, gatt_server_cb): 491 return False 492 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 493 gatt_callback) 494 495 @BluetoothBaseTest.bt_test_wrap 496 @test_tracker_info(uuid='4416d483-dec3-46cb-8038-4d82620f873a') 497 def test_gatt_request_out_of_bounds_mtu(self): 498 """Test GATT connection over LE and exercise an out of bound MTU size. 499 500 Test establishing a gatt connection between a GATT server and GATT 501 client. Request an MTU size that is the MIN value minus 1. 502 503 Steps: 504 1. Start a generic advertisement. 505 2. Start a generic scanner. 506 3. Find the advertisement and extract the mac address. 507 4. Stop the first scanner. 508 5. Create a GATT connection between the scanner and advertiser. 509 6. From the scanner (client) request MTU size change to the 510 minimum value minus one. 511 7. Find the MTU changed event on the client. 512 8. Disconnect the GATT connection. 513 514 Expected Result: 515 Verify that an MTU changed event was not discovered and that 516 it didn't cause an exception when requesting an out of bounds 517 MTU. 518 519 Returns: 520 Pass if True 521 Fail if False 522 523 TAGS: LE, Advertising, Filtering, Scanning, GATT, MTU 524 Priority: 0 525 """ 526 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 527 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 528 gatt_server_cb) 529 self.gatt_server_list.append(gatt_server) 530 try: 531 bluetooth_gatt, gatt_callback, adv_callback = ( 532 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 533 self.bluetooth_gatt_list.append(bluetooth_gatt) 534 except GattTestUtilsError as err: 535 self.log.error(err) 536 return False 537 self.adv_instances.append(adv_callback) 538 unexpected_mtu = gatt_mtu_size['min'] - 1 539 self.cen_ad.droid.gattClientRequestMtu(bluetooth_gatt, unexpected_mtu) 540 if self._verify_mtu_changed_on_client_and_server( 541 unexpected_mtu, gatt_callback, gatt_server_cb): 542 return False 543 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 544 gatt_callback) 545 546 @BluetoothBaseTest.bt_test_wrap 547 @test_tracker_info(uuid='31ffb9ca-cc75-43fb-9802-c19f1c5856b6') 548 def test_gatt_connect_trigger_on_read_rssi(self): 549 """Test GATT connection over LE read RSSI. 550 551 Test establishing a gatt connection between a GATT server and GATT 552 client then read the RSSI. 553 554 Steps: 555 1. Start a generic advertisement. 556 2. Start a generic scanner. 557 3. Find the advertisement and extract the mac address. 558 4. Stop the first scanner. 559 5. Create a GATT connection between the scanner and advertiser. 560 6. From the scanner, request to read the RSSI of the advertiser. 561 7. Disconnect the GATT connection. 562 563 Expected Result: 564 Verify that a connection was established and then disconnected 565 successfully. Verify that the RSSI was ready correctly. 566 567 Returns: 568 Pass if True 569 Fail if False 570 571 TAGS: LE, Advertising, Filtering, Scanning, GATT, RSSI 572 Priority: 1 573 """ 574 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 575 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 576 gatt_server_cb) 577 self.gatt_server_list.append(gatt_server) 578 try: 579 bluetooth_gatt, gatt_callback, adv_callback = ( 580 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 581 self.bluetooth_gatt_list.append(bluetooth_gatt) 582 except GattTestUtilsError as err: 583 self.log.error(err) 584 return False 585 self.adv_instances.append(adv_callback) 586 expected_event = gatt_cb_strings['rd_remote_rssi'].format( 587 gatt_callback) 588 if self.cen_ad.droid.gattClientReadRSSI(bluetooth_gatt): 589 try: 590 self.cen_ad.ed.pop_event(expected_event, self.default_timeout) 591 except Empty: 592 self.log.error( 593 gatt_cb_err['rd_remote_rssi_err'].format(expected_event)) 594 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 595 gatt_callback) 596 597 @BluetoothBaseTest.bt_test_wrap 598 @test_tracker_info(uuid='dee9ef28-b872-428a-821b-cc62f27ba936') 599 def test_gatt_connect_trigger_on_services_discovered(self): 600 """Test GATT connection and discover services of peripheral. 601 602 Test establishing a gatt connection between a GATT server and GATT 603 client the discover all services from the connected device. 604 605 Steps: 606 1. Start a generic advertisement. 607 2. Start a generic scanner. 608 3. Find the advertisement and extract the mac address. 609 4. Stop the first scanner. 610 5. Create a GATT connection between the scanner and advertiser. 611 6. From the scanner (central device), discover services. 612 7. Disconnect the GATT connection. 613 614 Expected Result: 615 Verify that a connection was established and then disconnected 616 successfully. Verify that the service were discovered. 617 618 Returns: 619 Pass if True 620 Fail if False 621 622 TAGS: LE, Advertising, Filtering, Scanning, GATT, Services 623 Priority: 1 624 """ 625 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 626 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 627 gatt_server_cb) 628 self.gatt_server_list.append(gatt_server) 629 try: 630 bluetooth_gatt, gatt_callback, adv_callback = ( 631 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 632 self.bluetooth_gatt_list.append(bluetooth_gatt) 633 except GattTestUtilsError as err: 634 self.log.error(err) 635 return False 636 self.adv_instances.append(adv_callback) 637 if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt): 638 expected_event = gatt_cb_strings['gatt_serv_disc'].format( 639 gatt_callback) 640 try: 641 event = self.cen_ad.ed.pop_event(expected_event, 642 self.default_timeout) 643 except Empty: 644 self.log.error( 645 gatt_cb_err['gatt_serv_disc'].format(expected_event)) 646 return False 647 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 648 gatt_callback) 649 650 @BluetoothBaseTest.bt_test_wrap 651 @test_tracker_info(uuid='01883bdd-0cf8-48fb-bf15-467bbd4f065b') 652 def test_gatt_connect_trigger_on_services_discovered_iterate_attributes( 653 self): 654 """Test GATT connection and iterate peripherals attributes. 655 656 Test establishing a gatt connection between a GATT server and GATT 657 client and iterate over all the characteristics and descriptors of the 658 discovered services. 659 660 Steps: 661 1. Start a generic advertisement. 662 2. Start a generic scanner. 663 3. Find the advertisement and extract the mac address. 664 4. Stop the first scanner. 665 5. Create a GATT connection between the scanner and advertiser. 666 6. From the scanner (central device), discover services. 667 7. Iterate over all the characteristics and descriptors of the 668 discovered features. 669 8. Disconnect the GATT connection. 670 671 Expected Result: 672 Verify that a connection was established and then disconnected 673 successfully. Verify that the services, characteristics, and descriptors 674 were discovered. 675 676 Returns: 677 Pass if True 678 Fail if False 679 680 TAGS: LE, Advertising, Filtering, Scanning, GATT, Services 681 Characteristics, Descriptors 682 Priority: 1 683 """ 684 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 685 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 686 gatt_server_cb) 687 self.gatt_server_list.append(gatt_server) 688 try: 689 bluetooth_gatt, gatt_callback, adv_callback = ( 690 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 691 self.bluetooth_gatt_list.append(bluetooth_gatt) 692 except GattTestUtilsError as err: 693 self.log.error(err) 694 return False 695 self.adv_instances.append(adv_callback) 696 if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt): 697 expected_event = gatt_cb_strings['gatt_serv_disc'].format( 698 gatt_callback) 699 try: 700 event = self.cen_ad.ed.pop_event(expected_event, 701 self.default_timeout) 702 discovered_services_index = event['data']['ServicesIndex'] 703 except Empty: 704 self.log.error( 705 gatt_cb_err['gatt_serv_disc'].format(expected_event)) 706 return False 707 log_gatt_server_uuids(self.cen_ad, discovered_services_index) 708 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 709 gatt_callback) 710 711 @BluetoothBaseTest.bt_test_wrap 712 @test_tracker_info(uuid='d4277bee-da99-4f48-8a4d-f81b5389da18') 713 def test_gatt_connect_with_service_uuid_variations(self): 714 """Test GATT connection with multiple service uuids. 715 716 Test establishing a gatt connection between a GATT server and GATT 717 client with multiple service uuid variations. 718 719 Steps: 720 1. Start a generic advertisement. 721 2. Start a generic scanner. 722 3. Find the advertisement and extract the mac address. 723 4. Stop the first scanner. 724 5. Create a GATT connection between the scanner and advertiser. 725 6. From the scanner (central device), discover services. 726 7. Verify that all the service uuid variations are found. 727 8. Disconnect the GATT connection. 728 729 Expected Result: 730 Verify that a connection was established and then disconnected 731 successfully. Verify that the service uuid variations are found. 732 733 Returns: 734 Pass if True 735 Fail if False 736 737 TAGS: LE, Advertising, Filtering, Scanning, GATT, Services 738 Priority: 2 739 """ 740 try: 741 gatt_server_cb, gatt_server = setup_multiple_services(self.per_ad) 742 self.gatt_server_list.append(gatt_server) 743 except GattTestUtilsError as err: 744 self.log.error(err) 745 return False 746 try: 747 bluetooth_gatt, gatt_callback, adv_callback = ( 748 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 749 self.bluetooth_gatt_list.append(bluetooth_gatt) 750 except GattTestUtilsError as err: 751 self.log.error(err) 752 return False 753 self.adv_instances.append(adv_callback) 754 if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt): 755 expected_event = gatt_cb_strings['gatt_serv_disc'].format( 756 gatt_callback) 757 try: 758 event = self.cen_ad.ed.pop_event(expected_event, 759 self.default_timeout) 760 except Empty: 761 self.log.error( 762 gatt_cb_err['gatt_serv_disc'].format(expected_event)) 763 return False 764 discovered_services_index = event['data']['ServicesIndex'] 765 log_gatt_server_uuids(self.cen_ad, discovered_services_index) 766 767 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 768 gatt_callback) 769 770 @BluetoothBaseTest.bt_test_wrap 771 @test_tracker_info(uuid='7d3442c5-f71f-44ae-bd35-f2569f01b3b8') 772 def test_gatt_connect_in_quick_succession(self): 773 """Test GATT connections multiple times. 774 775 Test establishing a gatt connection between a GATT server and GATT 776 client with multiple iterations. 777 778 Steps: 779 1. Start a generic advertisement. 780 2. Start a generic scanner. 781 3. Find the advertisement and extract the mac address. 782 4. Stop the first scanner. 783 5. Create a GATT connection between the scanner and advertiser. 784 6. Disconnect the GATT connection. 785 7. Repeat steps 5 and 6 twenty times. 786 787 Expected Result: 788 Verify that a connection was established and then disconnected 789 successfully twenty times. 790 791 Returns: 792 Pass if True 793 Fail if False 794 795 TAGS: LE, Advertising, Filtering, Scanning, GATT, Stress 796 Priority: 1 797 """ 798 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 799 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 800 gatt_server_cb) 801 self.gatt_server_list.append(gatt_server) 802 mac_address, adv_callback, scan_callback = get_mac_address_of_generic_advertisement( 803 self.cen_ad, self.per_ad) 804 autoconnect = False 805 for i in range(1000): 806 self.log.info("Starting connection iteration {}".format(i + 1)) 807 try: 808 bluetooth_gatt, gatt_callback = setup_gatt_connection( 809 self.cen_ad, mac_address, autoconnect) 810 self.cen_ad.droid.bleStopBleScan(scan_callback) 811 except GattTestUtilsError as err: 812 self.log.error(err) 813 return False 814 test_result = self._orchestrate_gatt_disconnection( 815 bluetooth_gatt, gatt_callback) 816 if not test_result: 817 self.log.info("Failed to disconnect from peripheral device.") 818 return False 819 self.adv_instances.append(adv_callback) 820 return True 821 822 @BluetoothBaseTest.bt_test_wrap 823 @test_tracker_info(uuid='148469d9-7ab0-4c08-b2e9-7e49e88da1fc') 824 def test_gatt_connect_mitm_attack(self): 825 """Test GATT connection with permission write encrypted mitm. 826 827 Test establishing a gatt connection between a GATT server and GATT 828 client while the GATT server's characteristic includes the property 829 write value and the permission write encrypted mitm value. This will 830 prompt LE pairing and then the devices will create a bond. 831 832 Steps: 833 1. Create a GATT server and server callback on the peripheral device. 834 2. Create a unique service and characteristic uuid on the peripheral. 835 3. Create a characteristic on the peripheral with these properties: 836 gatt_characteristic['property_write'], 837 gatt_characteristic['permission_write_encrypted_mitm'] 838 4. Create a GATT service on the peripheral. 839 5. Add the characteristic to the GATT service. 840 6. Create a GATT connection between your central and peripheral device. 841 7. From the central device, discover the peripheral's services. 842 8. Iterate the services found until you find the unique characteristic 843 created in step 3. 844 9. Once found, write a random but valid value to the characteristic. 845 10. Start pairing helpers on both devices immediately after attempting 846 to write to the characteristic. 847 11. Within 10 seconds of writing the characteristic, there should be 848 a prompt to bond the device from the peripheral. The helpers will 849 handle the UI interaction automatically. (see 850 BluetoothConnectionFacade.java bluetoothStartPairingHelper). 851 12. Verify that the two devices are bonded. 852 853 Expected Result: 854 Verify that a connection was established and the devices are bonded. 855 856 Returns: 857 Pass if True 858 Fail if False 859 860 TAGS: LE, Advertising, Filtering, Scanning, GATT, Characteristic, MITM 861 Priority: 1 862 """ 863 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 864 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 865 gatt_server_cb) 866 self.gatt_server_list.append(gatt_server) 867 service_uuid = "3846D7A0-69C8-11E4-BA00-0002A5D5C51B" 868 test_uuid = "aa7edd5a-4d1d-4f0e-883a-d145616a1630" 869 bonded = False 870 characteristic = self.per_ad.droid.gattServerCreateBluetoothGattCharacteristic( 871 test_uuid, gatt_characteristic['property_write'], 872 gatt_characteristic['permission_write_encrypted_mitm']) 873 gatt_service = self.per_ad.droid.gattServerCreateService( 874 service_uuid, gatt_service_types['primary']) 875 self.per_ad.droid.gattServerAddCharacteristicToService( 876 gatt_service, characteristic) 877 self.per_ad.droid.gattServerAddService(gatt_server, gatt_service) 878 result = self._find_service_added_event(gatt_server_cb, service_uuid) 879 if not result: 880 return False 881 bluetooth_gatt, gatt_callback, adv_callback = ( 882 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 883 self.bluetooth_gatt_list.append(bluetooth_gatt) 884 self.adv_instances.append(adv_callback) 885 if self.cen_ad.droid.gattClientDiscoverServices(bluetooth_gatt): 886 expected_event = gatt_cb_strings['gatt_serv_disc'].format( 887 gatt_callback) 888 try: 889 event = self.cen_ad.ed.pop_event(expected_event, 890 self.default_timeout) 891 except Empty: 892 self.log.error( 893 gatt_cb_err['gatt_serv_disc'].format(expected_event)) 894 return False 895 discovered_services_index = event['data']['ServicesIndex'] 896 else: 897 self.log.info("Failed to discover services.") 898 return False 899 test_value = [1, 2, 3, 4, 5, 6, 7] 900 services_count = self.cen_ad.droid.gattClientGetDiscoveredServicesCount( 901 discovered_services_index) 902 for i in range(services_count): 903 characteristic_uuids = ( 904 self.cen_ad.droid.gattClientGetDiscoveredCharacteristicUuids( 905 discovered_services_index, i)) 906 for characteristic_uuid in characteristic_uuids: 907 if characteristic_uuid == test_uuid: 908 self.cen_ad.droid.bluetoothStartPairingHelper() 909 self.per_ad.droid.bluetoothStartPairingHelper() 910 self.cen_ad.droid.gattClientCharacteristicSetValue( 911 bluetooth_gatt, discovered_services_index, i, 912 characteristic_uuid, test_value) 913 self.cen_ad.droid.gattClientWriteCharacteristic( 914 bluetooth_gatt, discovered_services_index, i, 915 characteristic_uuid) 916 start_time = time.time() + self.default_timeout 917 target_name = self.per_ad.droid.bluetoothGetLocalName() 918 while time.time() < start_time and bonded == False: 919 bonded_devices = \ 920 self.cen_ad.droid.bluetoothGetBondedDevices() 921 for device in bonded_devices: 922 if ('name' in device.keys() 923 and device['name'] == target_name): 924 bonded = True 925 break 926 bonded = False 927 target_name = self.cen_ad.droid.bluetoothGetLocalName() 928 while time.time() < start_time and bonded == False: 929 bonded_devices = \ 930 self.per_ad.droid.bluetoothGetBondedDevices() 931 for device in bonded_devices: 932 if ('name' in device.keys() 933 and device['name'] == target_name): 934 bonded = True 935 break 936 937 # Dual mode devices will establish connection over the classic transport, 938 # in order to establish bond over both transports, and do SDP. Starting 939 # disconnection before all this is finished is not safe, might lead to 940 # race conditions, i.e. bond over classic tranport shows up after LE 941 # bond is already removed. 942 time.sleep(4) 943 944 for ad in [self.cen_ad, self.per_ad]: 945 if not clear_bonded_devices(ad): 946 return False 947 948 # Necessary sleep time for entries to update unbonded state 949 time.sleep(2) 950 951 for ad in [self.cen_ad, self.per_ad]: 952 bonded_devices = ad.droid.bluetoothGetBondedDevices() 953 if len(bonded_devices) > 0: 954 self.log.error( 955 "Failed to unbond devices: {}".format(bonded_devices)) 956 return False 957 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 958 gatt_callback) 959 960 @BluetoothBaseTest.bt_test_wrap 961 @test_tracker_info(uuid='cc3fc361-7bf1-4ee2-9e46-4a27c88ce6a8') 962 def test_gatt_connect_get_connected_devices(self): 963 """Test GATT connections show up in getConnectedDevices 964 965 Test establishing a gatt connection between a GATT server and GATT 966 client. Verify that active connections show up using 967 BluetoothManager.getConnectedDevices API. 968 969 Steps: 970 1. Start a generic advertisement. 971 2. Start a generic scanner. 972 3. Find the advertisement and extract the mac address. 973 4. Stop the first scanner. 974 5. Create a GATT connection between the scanner and advertiser. 975 7. Verify the GATT Client has an open connection to the GATT Server. 976 8. Verify the GATT Server has an open connection to the GATT Client. 977 9. Disconnect the GATT connection. 978 979 Expected Result: 980 Verify that a connection was established, connected devices are found 981 on both the central and peripheral devices, and then disconnected 982 successfully. 983 984 Returns: 985 Pass if True 986 Fail if False 987 988 TAGS: LE, Advertising, Scanning, GATT 989 Priority: 2 990 """ 991 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 992 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 993 gatt_server_cb) 994 self.gatt_server_list.append(gatt_server) 995 try: 996 bluetooth_gatt, gatt_callback, adv_callback = ( 997 orchestrate_gatt_connection(self.cen_ad, self.per_ad)) 998 self.bluetooth_gatt_list.append(bluetooth_gatt) 999 except GattTestUtilsError as err: 1000 self.log.error(err) 1001 return False 1002 conn_cen_devices = self.cen_ad.droid.bluetoothGetConnectedLeDevices( 1003 bt_profile_constants['gatt']) 1004 conn_per_devices = self.per_ad.droid.bluetoothGetConnectedLeDevices( 1005 bt_profile_constants['gatt_server']) 1006 target_name = self.per_ad.droid.bluetoothGetLocalName() 1007 error_message = ("Connected device {} not found in list of connected " 1008 "devices {}") 1009 if not any(d['name'] == target_name for d in conn_cen_devices): 1010 self.log.error(error_message.format(target_name, conn_cen_devices)) 1011 return False 1012 # For the GATT server only check the size of the list since 1013 # it may or may not include the device name. 1014 target_name = self.cen_ad.droid.bluetoothGetLocalName() 1015 if not conn_per_devices: 1016 self.log.error(error_message.format(target_name, conn_per_devices)) 1017 return False 1018 self.adv_instances.append(adv_callback) 1019 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 1020 gatt_callback) 1021 1022 @BluetoothBaseTest.bt_test_wrap 1023 @test_tracker_info(uuid='a0a37ca6-9fa8-4d35-9fdb-0e25b4b8a363') 1024 def test_gatt_connect_second_adv_after_canceling_first_adv(self): 1025 """Test GATT connection to peripherals second advertising address. 1026 1027 The the ability of cancelling GATT connections and trying to reconnect 1028 to the same device via a different address. 1029 1030 Steps: 1031 1. A starts advertising 1032 2. B starts scanning and finds A's mac address 1033 3. Stop advertisement from step 1. Start a new advertisement on A and 1034 find the new new mac address, B knows of both old and new address. 1035 4. B1 sends connect request to old address of A 1036 5. B1 cancel connect attempt after 10 seconds 1037 6. B1 sends connect request to new address of A 1038 7. Verify B1 establish connection to A in less than 10 seconds 1039 1040 Expected Result: 1041 Verify that a connection was established only on the second 1042 advertisement's mac address. 1043 1044 Returns: 1045 Pass if True 1046 Fail if False 1047 1048 TAGS: LE, Advertising, Scanning, GATT 1049 Priority: 3 1050 """ 1051 autoconnect = False 1052 transport = gatt_transport['auto'] 1053 opportunistic = False 1054 # Setup a basic Gatt server on the peripheral 1055 gatt_server_cb = self.per_ad.droid.gattServerCreateGattServerCallback() 1056 gatt_server = self.per_ad.droid.gattServerOpenGattServer( 1057 gatt_server_cb) 1058 1059 # Set advertisement settings to include local name in advertisement 1060 # and set the advertising mode to low_latency. 1061 self.per_ad.droid.bleSetAdvertiseSettingsIsConnectable(True) 1062 self.per_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True) 1063 self.per_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( 1064 ble_advertise_settings_modes['low_latency']) 1065 1066 # Setup necessary advertisement objects. 1067 advertise_data = self.per_ad.droid.bleBuildAdvertiseData() 1068 advertise_settings = self.per_ad.droid.bleBuildAdvertiseSettings() 1069 advertise_callback = self.per_ad.droid.bleGenBleAdvertiseCallback() 1070 1071 # Step 1: Start advertisement 1072 self.per_ad.droid.bleStartBleAdvertising( 1073 advertise_callback, advertise_data, advertise_settings) 1074 1075 # Setup scan settings for low_latency scanning and to include the local name 1076 # of the advertisement started in step 1. 1077 filter_list = self.cen_ad.droid.bleGenFilterList() 1078 self.cen_ad.droid.bleSetScanSettingsNumOfMatches( 1079 ble_scan_settings_match_nums['one']) 1080 self.cen_ad.droid.bleSetScanFilterDeviceName( 1081 self.per_ad.droid.bluetoothGetLocalName()) 1082 self.cen_ad.droid.bleBuildScanFilter(filter_list) 1083 self.cen_ad.droid.bleSetScanSettingsScanMode( 1084 ble_scan_settings_modes['low_latency']) 1085 1086 # Setup necessary scan objects. 1087 scan_settings = self.cen_ad.droid.bleBuildScanSetting() 1088 scan_callback = self.cen_ad.droid.bleGenScanCallback() 1089 1090 # Step 2: Start scanning on central Android device and find peripheral 1091 # address. 1092 self.cen_ad.droid.bleStartBleScan(filter_list, scan_settings, 1093 scan_callback) 1094 expected_event_name = scan_result.format(scan_callback) 1095 try: 1096 mac_address_pre_restart = self.cen_ad.ed.pop_event( 1097 expected_event_name, self.default_timeout)['data']['Result'][ 1098 'deviceInfo']['address'] 1099 self.log.info( 1100 "Peripheral advertisement found with mac address: {}".format( 1101 mac_address_pre_restart)) 1102 except Empty: 1103 self.log.info("Peripheral advertisement not found") 1104 test_result = False 1105 1106 # Step 3: Restart peripheral advertising such that a new mac address is 1107 # created. 1108 self.per_ad.droid.bleStopBleAdvertising(advertise_callback) 1109 self.per_ad.droid.bleStartBleAdvertising( 1110 advertise_callback, advertise_data, advertise_settings) 1111 1112 mac_address_post_restart = mac_address_pre_restart 1113 1114 while True: 1115 try: 1116 mac_address_post_restart = self.cen_ad.ed.pop_event( 1117 expected_event_name, self.default_timeout)['data']['Result'][ 1118 'deviceInfo']['address'] 1119 self.log.info( 1120 "Peripheral advertisement found with mac address: {}".format( 1121 mac_address_post_restart)) 1122 except Empty: 1123 self.log.info("Peripheral advertisement not found") 1124 test_result = False 1125 1126 if mac_address_pre_restart != mac_address_post_restart: 1127 break 1128 1129 self.cen_ad.droid.bleStopBleScan(scan_callback) 1130 1131 # Steps 4: Try to connect to the first mac address 1132 gatt_callback = self.cen_ad.droid.gattCreateGattCallback() 1133 self.log.info( 1134 "Gatt Connect to mac address {}.".format(mac_address_pre_restart)) 1135 bluetooth_gatt = self.cen_ad.droid.gattClientConnectGatt( 1136 gatt_callback, mac_address_pre_restart, autoconnect, transport, 1137 opportunistic, gatt_phy_mask['1m_mask']) 1138 self.bluetooth_gatt_list.append(bluetooth_gatt) 1139 expected_event = gatt_cb_strings['gatt_conn_change'].format( 1140 gatt_callback) 1141 try: 1142 event = self.cen_ad.ed.pop_event(expected_event, 1143 self.default_timeout) 1144 self.log.error( 1145 "Connection callback updated unexpectedly: {}".format(event)) 1146 return False 1147 except Empty: 1148 self.log.info("No connection update as expected.") 1149 1150 # Step 5: Cancel connection request. 1151 self.cen_ad.droid.gattClientDisconnect(bluetooth_gatt) 1152 close_gatt_client(self.cen_ad, bluetooth_gatt) 1153 if bluetooth_gatt in self.bluetooth_gatt_list: 1154 self.bluetooth_gatt_list.remove(bluetooth_gatt) 1155 1156 # Step 6: Connect to second mac address. 1157 gatt_callback = self.cen_ad.droid.gattCreateGattCallback() 1158 self.log.info( 1159 "Gatt Connect to mac address {}.".format(mac_address_post_restart)) 1160 bluetooth_gatt = self.cen_ad.droid.gattClientConnectGatt( 1161 gatt_callback, mac_address_post_restart, autoconnect, transport, 1162 opportunistic, gatt_phy_mask['1m_mask']) 1163 self.bluetooth_gatt_list.append(bluetooth_gatt) 1164 expected_event = gatt_cb_strings['gatt_conn_change'].format( 1165 gatt_callback) 1166 1167 # Step 7: Verify connection was setup successfully. 1168 try: 1169 event = self.cen_ad.ed.pop_event(expected_event, 1170 self.default_timeout) 1171 self.log.info( 1172 "Connection callback updated successfully: {}".format(event)) 1173 if event['data']['State'] != gatt_connection_state['connected']: 1174 self.log.error( 1175 "Could not establish a connection to the peripheral.") 1176 return False 1177 except Empty: 1178 self.log.error("No connection update was found.") 1179 return False 1180 1181 self.per_ad.droid.bleStopBleAdvertising(advertise_callback) 1182 1183 return self._orchestrate_gatt_disconnection(bluetooth_gatt, 1184 gatt_callback) 1185