• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2#
3#   Copyright 2022 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of 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,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import queue
18import logging
19import time
20from datetime import timedelta
21from grpc import RpcError
22
23from bluetooth_packets_python3 import hci_packets
24from blueberry.facade.hci import le_advertising_manager_facade_pb2 as le_advertising_facade
25from blueberry.facade.hci import le_initiator_address_facade_pb2 as le_initiator_address_facade
26from blueberry.facade import common_pb2 as common
27from blueberry.tests.gd.cert.closable import safeClose
28from blueberry.tests.gd.cert.truth import assertThat
29from blueberry.tests.gd.cert.py_le_acl_manager import PyLeAclManager
30from blueberry.tests.gd_sl4a.lib import gd_sl4a_base_test
31from blueberry.tests.gd_sl4a.lib.bt_constants import ble_scan_settings_modes, ble_address_types, scan_result, ble_scan_settings_phys, ble_scan_settings_callback_types
32from blueberry.tests.gd_sl4a.lib.ble_lib import generate_ble_scan_objects
33from blueberry.utils.bt_gatt_utils import setup_gatt_connection
34from blueberry.utils.bt_gatt_utils import GattTestUtilsError
35from blueberry.utils.bt_gatt_utils import disconnect_gatt_connection
36from blueberry.utils.bt_gatt_utils import wait_for_gatt_disconnect_event
37from blueberry.utils.bt_gatt_utils import wait_for_gatt_connection
38from blueberry.utils.bt_gatt_utils import close_gatt_client
39from mobly.controllers.android_device import AndroidDevice
40from mobly import asserts
41from mobly import test_runner
42from mobly.signals import TestFailure
43
44
45class GattConnectLowLayerTest(gd_sl4a_base_test.GdSl4aBaseTestClass):
46
47    def setup_class(self):
48        super().setup_class(cert_module='HCI_INTERFACES')
49        self.bluetooth_gatt_list = []
50        self.default_timeout = 30  # seconds
51
52    def setup_test(self):
53        super().setup_test()
54        self.cert_le_acl_manager = PyLeAclManager(self.cert)
55
56    def teardown_test(self):
57        try:
58            for bluetooth_gatt in self.bluetooth_gatt_list:
59                self.dut.sl4a.gattClientClose(bluetooth_gatt)
60        except Exception as err:
61            logging.error("Failed to close GATT client, error: {}".format(err))
62        try:
63            safeClose(self.cert_le_acl_manager)
64        except RpcError as err:
65            logging.error("Failed to close CERT acl manager, error: {}".format(err))
66        self.cert_le_acl_manager = None
67        super().teardown_test()
68
69    def _set_cert_privacy_policy_with_random_address(self, random_address):
70        private_policy = le_initiator_address_facade.PrivacyPolicy(
71            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
72            address_with_type=common.BluetoothAddressWithType(
73                address=common.BluetoothAddress(address=bytes(random_address, encoding='utf8')),
74                type=common.RANDOM_DEVICE_ADDRESS))
75        self.cert.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(private_policy)
76
77    def _start_cert_advertising_with_random_address(self, device_name, random_address):
78        gap_name = hci_packets.GapData()
79        gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
80        gap_name.data = list(bytes(device_name, encoding='utf8'))
81        gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))
82        config = le_advertising_facade.AdvertisingConfig(
83            advertisement=[gap_data],
84            interval_min=512,
85            interval_max=768,
86            advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
87            own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,
88            channel_map=7,
89            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)
90        request = le_advertising_facade.CreateAdvertiserRequest(config=config)
91        logging.info("Creating advertiser")
92        create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)
93        logging.info("Created advertiser")
94        return create_response
95
96    def _start_dut_scanning_for_address(self, address_type, address):
97        logging.info("Start scanning for address {} with address type {}".format(address, address_type))
98        self.dut.sl4a.bleSetScanSettingsScanMode(ble_scan_settings_modes['low_latency'])
99        filter_list, scan_settings, scan_callback = generate_ble_scan_objects(self.dut.sl4a)
100        # Start scanning on SL4A DUT side
101        self.dut.sl4a.bleSetScanFilterDeviceAddressAndType(address, int(address_type))
102        self.dut.sl4a.bleBuildScanFilter(filter_list)
103        self.dut.sl4a.bleStartBleScan(filter_list, scan_settings, scan_callback)
104        return scan_callback
105
106    def _wait_for_scan_result_event(self, expected_event_name):
107        try:
108            # Verify if there is scan result
109            event_info = self.dut.ed.pop_event(expected_event_name, self.default_timeout)
110            # Print out scan result
111            mac_address = event_info['data']['Result']['deviceInfo']['address']
112            logging.info("Filter advertisement with address {}".format(mac_address))
113            return mac_address, event_info
114        except queue.Empty as error:
115            logging.error("Could not find initial advertisement.")
116            return None, None
117
118    def _stop_advertising(self, advertiser_id):
119        logging.info("Stop advertising")
120        remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=advertiser_id)
121        self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)
122        logging.info("Stopped advertising")
123
124    def _stop_scanning(self, scan_callback):
125        logging.info("Stop scanning")
126        self.dut.sl4a.bleStopBleScan(scan_callback)
127        logging.info("Stopped scanning")
128
129    def _disconnect_gatt(self, device: AndroidDevice, bluetooth_gatt, gatt_callback):
130        try:
131            disconnect_gatt_connection(device, bluetooth_gatt, gatt_callback)
132        except GattTestUtilsError as err:
133            logging.error(err)
134            asserts.fail("Cannot disconnect GATT , error={}".format(err))
135        finally:
136            close_gatt_client(device, bluetooth_gatt)
137            if bluetooth_gatt in self.bluetooth_gatt_list:
138                self.bluetooth_gatt_list.remove(bluetooth_gatt)
139
140    def _wait_for_gatt_connection(self, device: AndroidDevice, gatt_callback, bluetooth_gatt):
141        try:
142            wait_for_gatt_connection(device, gatt_callback, bluetooth_gatt, timeout=self.default_timeout)
143        except GattTestUtilsError as err:
144            logging.error(err)
145            asserts.fail("Cannot observe GATT connection , error={}".format(err))
146
147    def _wait_for_gatt_disconnection(self, device: AndroidDevice, gatt_callback):
148        try:
149            wait_for_gatt_disconnect_event(device, gatt_callback)
150        except GattTestUtilsError as err:
151            logging.error(err)
152            asserts.fail("Cannot observe GATT disconnection, error={}".format(err))
153
154    def test_autoconnect_gatt_without_pairing_and_disconnect_quickly(self):
155        """
156        Steps:
157        1. CERT: advertises with Random Static address
158        2. DUT: connect without pairing within 30 seconds
159        3. CERT: verify GATT connection
160        4. Wait 5 seconds
161        5. DUT: Disconnect GATT
162        6. CERT: Verify that GATT is disconnected within 5 seconds
163        """
164        # Use random address on cert side
165        logging.info("Setting random address")
166        RANDOM_ADDRESS = 'D0:05:04:03:02:01'
167        DEVICE_NAME = 'Im_The_CERT!'
168        self._set_cert_privacy_policy_with_random_address(RANDOM_ADDRESS)
169        logging.info("Set random address")
170
171        self.cert_le_acl_manager.listen_for_incoming_connections()
172
173        # Setup cert side to advertise
174        create_response = self._start_cert_advertising_with_random_address(DEVICE_NAME, RANDOM_ADDRESS)
175        logging.info("Started advertising")
176
177        # Setup SL4A DUT side to scan
178        addr_type = ble_address_types["random"]
179        scan_callback_token = self._start_dut_scanning_for_address(addr_type, RANDOM_ADDRESS)
180        logging.info("Started scanning")
181
182        # Wait for results
183        expected_event_name = scan_result.format(scan_callback_token)
184        scanned_mac_address, event_info = self._wait_for_scan_result_event(expected_event_name)
185
186        self._stop_scanning(scan_callback_token)
187        assertThat(scanned_mac_address).isNotNone()
188        assertThat(scanned_mac_address).isEqualTo(RANDOM_ADDRESS)
189
190        autoconnect = True
191        try:
192            bluetooth_gatt, gatt_callback = setup_gatt_connection(
193                self.dut, RANDOM_ADDRESS, autoconnect, timeout_seconds=self.default_timeout)
194        except GattTestUtilsError as err:
195            logging.error(err)
196            asserts.fail("Cannot make the first connection , error={}".format(err))
197            return
198        cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
199        self.dut.log.info("Device {} connected first time".format(RANDOM_ADDRESS))
200        self.dut.log.info("Sleeping 5 seconds to simulate real life connection")
201        time.sleep(5)
202        self.dut.log.info("Disconnecting GATT")
203        self._disconnect_gatt(self.dut, bluetooth_gatt, gatt_callback)
204        self.dut.log.info("Device {} disconnected first time from DUT".format(RANDOM_ADDRESS))
205        logging.info("Waiting 5 seconds to disconnect from CERT")
206        cert_acl_connection.wait_for_disconnection_complete(timeout=timedelta(seconds=5))
207        cert_acl_connection.close()
208        self._stop_advertising(create_response.advertiser_id)
209
210    def test_autoconnect_gatt_twice_with_random_address_without_pairing(self):
211        """
212        Steps:
213        1. CERT: advertises with Random Static address
214        2. DUT: connect without pairing
215        3. CERT: verify GATT connection
216        4. Wait 5 seconds
217        5. DUT: Disconnect GATT
218        6. CERT: Verify that GATT is disconnected within 30 seconds
219        7. DUT: Try to connect to Cert again, and verify it can be connected
220                within 30 seconds
221        """
222        # Use random address on cert side
223        logging.info("Setting random address")
224        RANDOM_ADDRESS = 'D0:05:04:03:02:01'
225        DEVICE_NAME = 'Im_The_CERT!'
226        self._set_cert_privacy_policy_with_random_address(RANDOM_ADDRESS)
227        logging.info("Set random address")
228
229        self.cert_le_acl_manager.listen_for_incoming_connections()
230
231        # Setup cert side to advertise
232        create_response = self._start_cert_advertising_with_random_address(DEVICE_NAME, RANDOM_ADDRESS)
233        logging.info("Started advertising")
234
235        # Setup SL4A DUT side to scan
236        addr_type = ble_address_types["random"]
237        scan_callback_token = self._start_dut_scanning_for_address(addr_type, RANDOM_ADDRESS)
238        logging.info("Started scanning")
239
240        # Wait for results
241        expected_event_name = scan_result.format(scan_callback_token)
242        scanned_mac_address, event_info = self._wait_for_scan_result_event(expected_event_name)
243
244        self._stop_scanning(scan_callback_token)
245        assertThat(scanned_mac_address).isNotNone()
246        assertThat(scanned_mac_address).isEqualTo(RANDOM_ADDRESS)
247
248        logging.info("Setting up first GATT connection to CERT")
249        autoconnect = True
250        try:
251            bluetooth_gatt, gatt_callback = setup_gatt_connection(
252                self.dut, RANDOM_ADDRESS, autoconnect, timeout_seconds=self.default_timeout)
253        except GattTestUtilsError as err:
254            logging.error(err)
255            asserts.fail("Cannot make the first connection , error={}".format(err))
256            return
257        cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
258        # listen early as GATT might be reconnected on error
259        self.cert_le_acl_manager.listen_for_incoming_connections()
260        self.dut.log.info("Device {} connected first time".format(RANDOM_ADDRESS))
261        self.dut.log.info("Sleeping 5 seconds to simulate real life connection")
262        time.sleep(5)
263        self.dut.log.info("Disconnecting first GATT connection")
264        self._disconnect_gatt(self.dut, bluetooth_gatt, gatt_callback)
265        self.dut.log.info("Device {} disconnected first time from DUT".format(RANDOM_ADDRESS))
266        logging.info("Waiting 30 seconds to disconnect from CERT")
267        cert_acl_connection.wait_for_disconnection_complete(timeout=timedelta(seconds=30))
268        logging.info("Setting up second GATT connection to CERT")
269        try:
270            bluetooth_gatt, gatt_callback = setup_gatt_connection(
271                self.dut, RANDOM_ADDRESS, autoconnect, timeout_seconds=self.default_timeout)
272        except GattTestUtilsError as err:
273            close_gatt_client(self.dut, bluetooth_gatt)
274            logging.error(err)
275            asserts.fail("Cannot make the second connection , error={}".format(err))
276        cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
277        self.dut.log.info("Device {} connected second time".format(RANDOM_ADDRESS))
278        self.dut.log.info("Disconnect second GATT connection")
279        self._disconnect_gatt(self.dut, bluetooth_gatt, gatt_callback)
280        logging.info("Wait for CERT to disconnect")
281        cert_acl_connection.wait_for_disconnection_complete(timeout=timedelta(seconds=30))
282        cert_acl_connection.close()
283        self.dut.log.info("Device {} disconnected second time".format(RANDOM_ADDRESS))
284        self._stop_advertising(create_response.advertiser_id)
285
286    def test_disconnect_autoconnect_without_close(self):
287        """
288        Steps:
289        1. CERT: advertises with Random Static address
290        2. DUT: connect without pairing within 30 seconds
291        3. CERT: verify GATT connection
292        4. Wait 5 seconds
293        5. DUT: Disconnect GATT, but do not close it, keep CERT advertising ON
294        6. CERT: Verify that GATT is disconnected within 5 seconds
295        7. CERT: Verify that no further GATT connection is made
296        8. CERT: Stop advertising
297        """
298        # Use random address on cert side
299        logging.info("Setting random address")
300        RANDOM_ADDRESS = 'D0:05:04:03:02:01'
301        DEVICE_NAME = 'Im_The_CERT!'
302        self._set_cert_privacy_policy_with_random_address(RANDOM_ADDRESS)
303        logging.info("Set random address")
304
305        self.cert_le_acl_manager.listen_for_incoming_connections()
306
307        # Setup cert side to advertise
308        create_response = self._start_cert_advertising_with_random_address(DEVICE_NAME, RANDOM_ADDRESS)
309        logging.info("Started advertising")
310
311        # Setup SL4A DUT side to scan
312        addr_type = ble_address_types["random"]
313        scan_callback_token = self._start_dut_scanning_for_address(addr_type, RANDOM_ADDRESS)
314        logging.info("Started scanning")
315
316        # Wait for results
317        expected_event_name = scan_result.format(scan_callback_token)
318        scanned_mac_address, event_info = self._wait_for_scan_result_event(expected_event_name)
319
320        self._stop_scanning(scan_callback_token)
321        assertThat(scanned_mac_address).isNotNone()
322        assertThat(scanned_mac_address).isEqualTo(RANDOM_ADDRESS)
323
324        autoconnect = True
325        try:
326            bluetooth_gatt, gatt_callback = setup_gatt_connection(
327                self.dut, RANDOM_ADDRESS, autoconnect, timeout_seconds=self.default_timeout)
328        except GattTestUtilsError as err:
329            logging.error(err)
330            asserts.fail("Cannot make the first connection , error={}".format(err))
331            return
332        cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
333        self.cert_le_acl_manager.listen_for_incoming_connections()
334        self.dut.log.info("Device {} connected first time".format(RANDOM_ADDRESS))
335        self.dut.log.info("Sleeping 5 seconds to simulate real life connection")
336        time.sleep(5)
337        self.dut.log.info("Disconnect first GATT connection")
338        self._disconnect_gatt(self.dut, bluetooth_gatt, gatt_callback)
339        self.dut.log.info("Device {} disconnected first time from DUT".format(RANDOM_ADDRESS))
340        logging.info("Waiting 5 seconds to disconnect from CERT")
341        cert_acl_connection.wait_for_disconnection_complete(timeout=timedelta(seconds=5))
342        cert_acl_connection.close()
343        logging.info("Verifying that no further GATT connection is made")
344        try:
345            cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
346            asserts.fail("Should not have a GATT connection")
347        except TestFailure:
348            pass
349        logging.info("Stop advertising")
350        self._stop_advertising(create_response.advertiser_id)
351
352    def test_autoconnect_without_proactive_disconnect(self):
353        """
354        Steps:
355        1. CERT: advertises with Random Static address
356        2. DUT: connect without pairing within 30 seconds
357        3. CERT: verify GATT connection
358        4. Wait 5 seconds
359        5. CERT: Turn off advertising
360        6. CERT: Disconnect existing GATT connection
361        7. DUT: Verify that GATT is disconnected within 5 seconds
362        8. CERT: Start advertising
363        9. DUT: Verify GATT connects within 5 seconds
364        10. CERT: Stop advertising and disconnect DUT
365        11. DUT: Verify that GATT disconnects within 5 seconds
366        """
367        # Use random address on cert side
368        logging.info("Setting random address")
369        RANDOM_ADDRESS = 'D0:05:04:03:02:01'
370        DEVICE_NAME = 'Im_The_CERT!'
371        self._set_cert_privacy_policy_with_random_address(RANDOM_ADDRESS)
372        logging.info("Set random address")
373
374        self.cert_le_acl_manager.listen_for_incoming_connections()
375
376        # Setup cert side to advertise
377        create_response = self._start_cert_advertising_with_random_address(DEVICE_NAME, RANDOM_ADDRESS)
378        logging.info("Started advertising")
379
380        # Setup SL4A DUT side to scan
381        addr_type = ble_address_types["random"]
382        scan_callback_token = self._start_dut_scanning_for_address(addr_type, RANDOM_ADDRESS)
383        logging.info("Started scanning")
384
385        # Wait for results
386        expected_event_name = scan_result.format(scan_callback_token)
387        scanned_mac_address, event_info = self._wait_for_scan_result_event(expected_event_name)
388
389        self._stop_scanning(scan_callback_token)
390        assertThat(scanned_mac_address).isNotNone()
391        assertThat(scanned_mac_address).isEqualTo(RANDOM_ADDRESS)
392
393        autoconnect = True
394        try:
395            bluetooth_gatt, gatt_callback = setup_gatt_connection(
396                self.dut, RANDOM_ADDRESS, autoconnect, timeout_seconds=self.default_timeout)
397        except GattTestUtilsError as err:
398            logging.error(err)
399            asserts.fail("Cannot make the first connection , error={}".format(err))
400            return
401        cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
402        self.dut.log.info("Device {} connected first time".format(RANDOM_ADDRESS))
403        self.dut.log.info("Sleeping 5 seconds to simulate real life connection")
404        time.sleep(5)
405        logging.info("Stopping cert advertising")
406        self._stop_advertising(create_response.advertiser_id)
407        logging.info("Disconnecting cert")
408        cert_acl_connection.disconnect()
409        logging.info("Waiting for cert to disconnect")
410        cert_acl_connection.wait_for_disconnection_complete(timeout=timedelta(seconds=10))
411        cert_acl_connection.close()
412        logging.info("Waiting for DUT to see disconnection")
413        self._wait_for_gatt_disconnection(self.dut, gatt_callback)
414        self.dut.log.info("Device {} disconnected first time from DUT".format(RANDOM_ADDRESS))
415        logging.info("Waiting 5 seconds to disconnect from CERT")
416        logging.info("Start CERT advertising")
417        self.cert_le_acl_manager.listen_for_incoming_connections()
418        create_response = self._start_cert_advertising_with_random_address(DEVICE_NAME, RANDOM_ADDRESS)
419        logging.info("Waiting for GATT to connect")
420        self._wait_for_gatt_connection(self.dut, gatt_callback, bluetooth_gatt)
421        logging.info("Waiting on CERT as well for background connection")
422        cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
423        logging.info("GATT is connected via background connection")
424        self.dut.log.info("Sleeping 5 seconds to simulate real life connection")
425        time.sleep(5)
426        logging.info("Stopping cert advertising")
427        self._stop_advertising(create_response.advertiser_id)
428        logging.info("Disconnecting cert")
429        cert_acl_connection.disconnect()
430        logging.info("Waiting for cert to disconnect")
431        cert_acl_connection.wait_for_disconnection_complete(timeout=timedelta(seconds=10))
432        cert_acl_connection.close()
433        logging.info("Waiting for DUT to see disconnection")
434        self._wait_for_gatt_disconnection(self.dut, gatt_callback)
435        logging.info("Verifying that no further GATT connection is made")
436        try:
437            self.cert_le_acl_manager.complete_incoming_connection()
438            asserts.fail("Should not have a GATT connection")
439        except TestFailure:
440            pass
441
442    def test_autoconnect_without_proactive_disconnect_repeatedly(self):
443        """
444        Steps:
445        1. CERT: advertises with Random Static address
446        2. DUT: connect without pairing within 30 seconds
447        3. CERT: verify GATT connection
448        4. Wait 5 seconds
449        5. CERT: Turn off advertising
450        6. CERT: Disconnect existing GATT connection
451        7. DUT: Verify that GATT is disconnected within 5 seconds
452        8. CERT: Start advertising
453        9. DUT: Verify GATT connects within 5 seconds
454        10. CERT: Stop advertising and disconnect DUT
455        11. DUT: Verify that GATT disconnects within 5 seconds
456        12. Repeat step 8 to 11 for 20 times
457        """
458        # Use random address on cert side
459        logging.info("Setting random address")
460        RANDOM_ADDRESS = 'D0:05:04:03:02:01'
461        DEVICE_NAME = 'Im_The_CERT!'
462        self._set_cert_privacy_policy_with_random_address(RANDOM_ADDRESS)
463        logging.info("Set random address")
464
465        self.cert_le_acl_manager.listen_for_incoming_connections()
466
467        # Setup cert side to advertise
468        create_response = self._start_cert_advertising_with_random_address(DEVICE_NAME, RANDOM_ADDRESS)
469        logging.info("Started advertising")
470
471        # Setup SL4A DUT side to scan
472        addr_type = ble_address_types["random"]
473        scan_callback_token = self._start_dut_scanning_for_address(addr_type, RANDOM_ADDRESS)
474        logging.info("Started scanning")
475
476        # Wait for results
477        expected_event_name = scan_result.format(scan_callback_token)
478        scanned_mac_address, event_info = self._wait_for_scan_result_event(expected_event_name)
479
480        self._stop_scanning(scan_callback_token)
481        assertThat(scanned_mac_address).isNotNone()
482        assertThat(scanned_mac_address).isEqualTo(RANDOM_ADDRESS)
483
484        autoconnect = True
485        try:
486            bluetooth_gatt, gatt_callback = setup_gatt_connection(
487                self.dut, RANDOM_ADDRESS, autoconnect, timeout_seconds=self.default_timeout)
488        except GattTestUtilsError as err:
489            logging.error(err)
490            asserts.fail("Cannot make the first connection , error={}".format(err))
491            return
492        cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
493        self.dut.log.info("Device {} connected first time".format(RANDOM_ADDRESS))
494        self.dut.log.info("Sleeping 5 seconds to simulate real life connection")
495        time.sleep(5)
496        logging.info("Stopping CERT advertising")
497        self._stop_advertising(create_response.advertiser_id)
498        logging.info("Stopped CERT advertising, now disconnect cert")
499        cert_acl_connection.disconnect()
500        logging.info("Waiting for cert to disconnect")
501        cert_acl_connection.wait_for_disconnection_complete(timeout=timedelta(seconds=10))
502        cert_acl_connection.close()
503        logging.info("Cert disconnected, waiting for DUT to see disconnection event")
504        self._wait_for_gatt_disconnection(self.dut, gatt_callback)
505        self.dut.log.info("Device {} disconnected first time from DUT".format(RANDOM_ADDRESS))
506        logging.info("Waiting 5 seconds to disconnect from CERT")
507        for i in range(20):
508            logging.info("Start advertising on CERT")
509            self.cert_le_acl_manager.listen_for_incoming_connections()
510            create_response = self._start_cert_advertising_with_random_address(DEVICE_NAME, RANDOM_ADDRESS)
511            self.dut.log.info("Wait on DUT for background GATT connection")
512            self._wait_for_gatt_connection(self.dut, gatt_callback, bluetooth_gatt)
513            logging.info("Waiting on CERT as well for background connection")
514            cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
515            logging.info("GATT is connected via background connection")
516            self.dut.log.info("Sleeping 5 seconds to simulate real life connection")
517            time.sleep(5)
518            logging.info("Stop advertising from CERT")
519            self._stop_advertising(create_response.advertiser_id)
520            logging.info("Disconnect from CERT")
521            cert_acl_connection.disconnect()
522            logging.info("Waiting on CERT end for disconnection to happen")
523            cert_acl_connection.wait_for_disconnection_complete(timeout=timedelta(seconds=10))
524            cert_acl_connection.close()
525            self.dut.log.info("Waiting on DUT end for disconnection to happen")
526            self._wait_for_gatt_disconnection(self.dut, gatt_callback)
527        logging.info("Verifying that no further GATT connection is made")
528        try:
529            cert_acl_connection = self.cert_le_acl_manager.complete_incoming_connection()
530            asserts.fail("Should not have a GATT connection")
531        except TestFailure:
532            pass
533
534
535if __name__ == '__main__':
536    test_runner.main()
537