1#!/usr/bin/env python3 2# 3# Copyright (C) 2019 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 is a stress test for Fuchsia GATT connections. 18 19Custom Params: 20gatt_connect_stress_test_iterations 21 22 Example: 23 "gatt_connect_stress_test_iterations": 10 24 25Setup: 26This test only requires two fuchsia devices as the purpose is to test 27the robusntess of GATT connections. 28""" 29 30from acts import signals 31from acts.base_test import BaseTestClass 32from acts.test_decorators import test_tracker_info 33from acts_contrib.test_utils.bt.bt_test_utils import generate_id_by_size 34from acts_contrib.test_utils.fuchsia.bt_test_utils import le_scan_for_device_by_name 35 36 37class GattConnectionStressTest(BaseTestClass): 38 gatt_connect_err_message = "Gatt connection failed with: {}" 39 gatt_disconnect_err_message = "Gatt disconnection failed with: {}" 40 ble_advertise_interval = 50 41 scan_timeout_seconds = 60 42 default_iterations = 1000 43 44 def setup_class(self): 45 super().setup_class() 46 self.fuchsia_client_dut = self.fuchsia_devices[0] 47 self.fuchsia_server_dut = self.fuchsia_devices[1] 48 self.default_iterations = self.user_params.get( 49 "gatt_connect_stress_test_iterations", self.default_iterations) 50 51 def on_fail(self, test_name, begin_time): 52 for fd in self.fuchsia_devices: 53 fd.take_bug_report(test_name, begin_time) 54 55 def _orchestrate_single_connect_disconnect(self): 56 adv_name = generate_id_by_size(10) 57 adv_data = { 58 "name": adv_name, 59 "appearance": None, 60 "service_data": None, 61 "tx_power_level": None, 62 "service_uuids": None, 63 "manufacturer_data": None, 64 "uris": None, 65 } 66 scan_response = None 67 connectable = True 68 self.fuchsia_server_dut.sl4f.ble_lib.bleStartBleAdvertising( 69 adv_data, scan_response, self.ble_advertise_interval, connectable) 70 device = le_scan_for_device_by_name(self.fuchsia_client_dut, self.log, 71 adv_name, 72 self.scan_timeout_seconds) 73 if device is None: 74 raise signals.TestFailure("Scanner unable to find advertisement.") 75 connect_result = self.fuchsia_client_dut.sl4f.gattc_lib.bleConnectToPeripheral( 76 device["id"]) 77 if connect_result.get("error") is not None: 78 raise signals.TestFailure( 79 self.gatt_connect_err_message.format( 80 connect_result.get("error"))) 81 self.log.info("Connection Successful...") 82 disconnect_result = self.fuchsia_client_dut.sl4f.gattc_lib.bleDisconnectPeripheral( 83 device["id"]) 84 if disconnect_result.get("error") is not None: 85 raise signals.TestFailure( 86 self.gatt_disconnect_err_message.format( 87 connect_result.get("error"))) 88 self.log.info("Disconnection Successful...") 89 self.fuchsia_server_dut.sl4f.ble_lib.bleStopBleAdvertising() 90 91 # TODO: add @test_tracker_info(uuid='') 92 def test_connect_reconnect_n_iterations_over_le(self): 93 """Test GATT reconnection n times. 94 95 Verify that the GATT client device can discover and connect to 96 a perpheral n times. Default value is 1000. 97 98 Steps: 99 1. Setup Ble advertisement on peripheral with unique advertisement 100 name. 101 2. GATT client scans for peripheral advertisement. 102 3. Upon find the advertisement, send a connection request to 103 peripheral. 104 105 Expected Result: 106 Verify that there are no errors after each GATT connection. 107 108 Returns: 109 signals.TestPass if no errors 110 signals.TestFailure if there are any errors during the test. 111 112 TAGS: GATT 113 Priority: 1 114 """ 115 for i in range(self.default_iterations): 116 self.log.info("Starting iteration {}".format(i + 1)) 117 self._orchestrate_single_connect_disconnect() 118 self.log.info("Iteration {} successful".format(i + 1)) 119 raise signals.TestPass("Success") 120