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