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