• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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.test_utils.bt.bt_test_utils import generate_id_by_size
34from acts.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 = 10
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 _orchestrate_single_connect_disconnect(self):
53        adv_name = generate_id_by_size(10)
54        adv_data = {"name": adv_name}
55        self.fuchsia_server_dut.ble_lib.bleStartBleAdvertising(
56            adv_data, self.ble_advertise_interval)
57        device = le_scan_for_device_by_name(self.fuchsia_client_dut, self.log,
58                                            adv_name,
59                                            self.scan_timeout_seconds)
60        if device is None:
61            raise signals.TestFailure("Scanner unable to find advertisement.")
62        connect_result = self.fuchsia_client_dut.gattc_lib.bleConnectToPeripheral(
63            device["id"])
64        if connect_result.get("error") is not None:
65            raise signals.TestFailure(
66                self.gatt_connect_err_message.format(
67                    connect_result.get("error")))
68        self.log.info("Connection Successful...")
69        disconnect_result = self.fuchsia_client_dut.gattc_lib.bleDisconnectPeripheral(
70            device["id"])
71        if disconnect_result.get("error") is not None:
72            raise signals.TestFailure(
73                self.gatt_disconnect_err_message.format(
74                    connect_result.get("error")))
75        self.log.info("Disconnection Successful...")
76        self.fuchsia_server_dut.ble_lib.bleStopBleAdvertising()
77
78    # TODO: add @test_tracker_info(uuid='')
79    def test_connect_reconnect_n_iterations_over_le(self):
80        """Test GATT reconnection n times.
81
82        Verify that the GATT client device can discover and connect to
83        a perpheral n times. Default value is 1000.
84
85        Steps:
86        1. Setup Ble advertisement on peripheral with unique advertisement
87            name.
88        2. GATT client scans for peripheral advertisement.
89        3. Upon find the advertisement, send a connection request to
90            peripheral.
91
92        Expected Result:
93        Verify that there are no errors after each GATT connection.
94
95        Returns:
96          signals.TestPass if no errors
97          signals.TestFailure if there are any errors during the test.
98
99        TAGS: GATT
100        Priority: 1
101        """
102        for i in range(self.default_iterations):
103            self.log.info("Starting iteration {}".format(i + 1))
104            self._orchestrate_single_connect_disconnect()
105            self.log.info("Iteration {} successful".format(i + 1))
106        raise signals.TestPass("Success")
107