1#!/usr/bin/env python3
2#
3# Copyright (C) 2016 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"""
17Test script to execute Bluetooth basic functionality test cases.
18This test was designed to be run in a shield box.
19"""
20
21import time
22from random import randint
23
24from queue import Empty
25from acts.test_decorators import test_tracker_info
26from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
27from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
28from acts.test_utils.bt.bt_test_utils import orchestrate_rfcomm_connection
29from acts.test_utils.bt.bt_test_utils import write_read_verify_data
30
31
32class RfcommLongevityTest(BluetoothBaseTest):
33    default_timeout = 10
34    longev_iterations = 200
35    write_iterations = 5000
36    generic_message = (
37        "Space: the final frontier. These are the voyages of "
38        "the starship Enterprise. Its continuing mission: to explore "
39        "strange new worlds, to seek out new life and new civilizations,"
40        " to boldly go where no man has gone before.")
41
42    def setup_class(self):
43        super().setup_class()
44        self.client_ad = self.android_devices[0]
45        self.server_ad = self.android_devices[1]
46
47    @test_tracker_info(uuid='2790acad-1f6e-4216-aadf-83561dddcfa3')
48    def test_rfcomm_longev_read_write_message(self):
49        """Longevity test an RFCOMM connection's I/O with a generic message
50
51        Test the longevity of RFCOMM with a basic read/write
52        connect/disconnect sequence.
53
54        Steps:
55        1. Establish a bonding between two Android devices.
56        2. Write data to RFCOMM from the client droid.
57        3. Read data from RFCOMM from the server droid.
58        4. Verify data written matches data read.
59        5. Repeat steps 2-4 5000 times.
60        6. Disconnect RFCOMM connection.
61        7. Repeat steps 1-6 1000 times.
62
63        Expected Result:
64        Each iteration should read and write to the RFCOMM connection
65        successfully. Each connect and disconnect should be successful.
66
67        Returns:
68          Pass if True
69          Fail if False
70
71        TAGS: Classic, Longevity, RFCOMM
72        Priority: 2
73        """
74
75        for i in range(self.longev_iterations):
76            self.log.info("iteration {} connection".format(i + 1))
77            if not orchestrate_rfcomm_connection(self.client_ad,
78                                                 self.server_ad):
79                return False
80            for n in range(self.write_iterations):
81                self.log.info("iteration {} data".format(((n + 1) + (
82                    i * self.write_iterations))))
83                if not write_read_verify_data(self.client_ad, self.server_ad,
84                                              self.generic_message, False):
85                    return False
86                self.log.info("Iteration {} completed".format(n))
87            self.client_ad.droid.bluetoothRfcommStop()
88            self.server_ad.droid.bluetoothRfcommStop()
89        return True
90
91    @test_tracker_info(uuid='01c420b8-845a-4539-90bf-8d7747f471ca')
92    def test_rfcomm_longev_read_write_small_message(self):
93        """Longevity test an RFCOMM connection's I/O with a small message
94
95        Test the longevity of RFCOMM with a basic read/write
96        connect/disconnect sequence. The data being transfered is only
97        one character in size.
98
99        Steps:
100        1. Establish a bonding between two Android devices.
101        2. Write data to RFCOMM from the client droid.
102        3. Read data from RFCOMM from the server droid.
103        4. Verify data written matches data read.
104        5. Repeat steps 2-4 5000 times.
105        6. Disconnect RFCOMM connection.
106        7. Repeat steps 1-6 1000 times.
107
108        Expected Result:
109        Each iteration should read and write to the RFCOMM connection
110        successfully. Each connect and disconnect should be successful.
111
112        Returns:
113          Pass if True
114          Fail if False
115
116        TAGS: Classic, Longevity, RFCOMM
117        Priority: 2
118        """
119        message = "x"
120        for i in range(self.longev_iterations):
121            self.log.info("iteration {} connection".format(i + 1))
122            if not orchestrate_rfcomm_connection(self.client_ad,
123                                                 self.server_ad):
124                return False
125            for n in range(self.write_iterations):
126                self.log.info("iteration {} data".format(((n + 1) + (
127                    i * self.write_iterations))))
128                if not write_read_verify_data(self.client_ad, self.server_ad,
129                                              message, False):
130                    return False
131                self.log.info("Iteration {} completed".format(n))
132            self.client_ad.droid.bluetoothRfcommStop()
133            self.server_ad.droid.bluetoothRfcommStop()
134        return True
135
136    @test_tracker_info(uuid='8a92772a-511e-4f15-8e8b-194e499a46eb')
137    def test_rfcomm_longev_read_write_binary_message(self):
138        """Longevity test an RFCOMM connection's I/O with a binary message
139
140        Test the longevity of RFCOMM with a basic read/write
141        connect/disconnect sequence. The data being transfered is in a
142        binary format.
143
144        Steps:
145        1. Establish a bonding between two Android devices.
146        2. Write data to RFCOMM from the client droid.
147        3. Read data from RFCOMM from the server droid.
148        4. Verify data written matches data read.
149        5. Repeat steps 2-4 5000 times.
150        6. Disconnect RFCOMM connection.
151        7. Repeat steps 1-6 1000 times.
152
153        Expected Result:
154        Each iteration should read and write to the RFCOMM connection
155        successfully. Each connect and disconnect should be successful.
156
157        Returns:
158          Pass if True
159          Fail if False
160
161        TAGS: Classic, Longevity, RFCOMM
162        Priority: 2
163        """
164        binary_message = "11010101"
165        for i in range(self.longev_iterations):
166            self.log.info("iteration {} connection".format(i + 1))
167            if not orchestrate_rfcomm_connection(self.client_ad,
168                                                 self.server_ad):
169                return False
170            for n in range(self.write_iterations):
171                self.log.info("iteration {} data".format(((n + 1) + (
172                    i * self.write_iterations))))
173                if not write_read_verify_data(self.client_ad, self.server_ad,
174                                              binary_message, True):
175                    return False
176                self.log.info("Iteration {} completed".format(n))
177            self.client_ad.droid.bluetoothRfcommStop()
178            self.server_ad.droid.bluetoothRfcommStop()
179        return True
180
181    @test_tracker_info(uuid='ff0ab2e4-2a7d-45b9-b034-4cd32c0fa139')
182    def test_rfcomm_longev_read_write_large_message(self):
183        """Longevity test an RFCOMM connection's I/O with a large message
184
185        Test the longevity of RFCOMM with a basic read/write
186        connect/disconnect sequence. The data being transfered is 990 chars
187        in size.
188
189        Steps:
190        1. Establish a bonding between two Android devices.
191        2. Write data to RFCOMM from the client droid.
192        3. Read data from RFCOMM from the server droid.
193        4. Verify data written matches data read.
194        5. Repeat steps 2-4 5000 times.
195        6. Disconnect RFCOMM connection.
196        7. Repeat steps 1-6 1000 times.
197
198        Expected Result:
199        Each iteration should read and write to the RFCOMM connection
200        successfully. Each connect and disconnect should be successful.
201
202        Returns:
203          Pass if True
204          Fail if False
205
206        TAGS: Classic, Longevity, RFCOMM
207        Priority: 2
208        """
209        message = "x" * 990  # largest message size till sl4a fixed
210        for i in range(self.longev_iterations):
211            self.log.info("iteration {} connection".format(i + 1))
212            if not orchestrate_rfcomm_connection(self.client_ad,
213                                                 self.server_ad):
214                return False
215            for n in range(self.write_iterations):
216                self.log.info("iteration {} data".format(((n + 1) + (
217                    i * self.write_iterations))))
218                if not write_read_verify_data(self.client_ad, self.server_ad,
219                                              message, False):
220                    return False
221                self.log.info("Iteration {} completed".format(n))
222            self.client_ad.droid.bluetoothRfcommStop()
223            self.server_ad.droid.bluetoothRfcommStop()
224        return True
225
226    @test_tracker_info(uuid='950924b7-d893-4a33-ba09-d80d53dc7d13')
227    def test_rfcomm_longev_connection_interuption(self):
228        """Longevity test an RFCOMM connection's with socket interuptions
229
230        Test the longevity of RFCOMM with a basic read/write
231        connect/disconnect sequence. Randomly in the sequence of reads and
232        writes the socket on the client side will close. There should be
233        an exception thrown for writing the next set of data and the
234        test should start up a new connection and continue.
235
236        Steps:
237        1. Establish a bonding between two Android devices.
238        2. Write data to RFCOMM from the client droid.
239        3. Read data from RFCOMM from the server droid.
240        4. Verify data written matches data read.
241        5. Repeat steps 2-4 5000 times or until the random interupt occurs.
242        6. Re-establish an RFCOMM connection.
243        7. Repeat steps 1-6 1000 times.
244
245        Expected Result:
246        Each iteration should read and write to the RFCOMM connection
247        successfully. Each connect and disconnect should be successful.
248        Devices should recover a new connection after each interruption.
249
250        Returns:
251          Pass if True
252          Fail if False
253
254        TAGS: Classic, Longevity, RFCOMM
255        Priority: 2
256        """
257        for i in range(self.longev_iterations):
258            try:
259                self.log.info("iteration {} connection".format(i + 1))
260                if not orchestrate_rfcomm_connection(self.client_ad,
261                                                     self.server_ad):
262                    return False
263                random_interup_iteration = randint(0, self.write_iterations)
264                for n in range(self.write_iterations):
265                    self.log.info("iteration {} data".format(((n + 1) + (
266                        i * self.write_iterations))))
267                    if not write_read_verify_data(self.client_ad,
268                                                  self.server_ad,
269                                                  self.generic_message, False):
270                        return False
271                    self.log.info("Iteration {} completed".format(n))
272                    if n > random_interup_iteration:
273                        self.client_ad.droid.bluetoothRfcommCloseSocket()
274                self.client_ad.droid.bluetoothRfcommStop()
275                self.server_ad.droid.bluetoothRfcommStop()
276            except Exception:
277                self.log.info("Exception found as expected. Continuing...")
278                try:
279                    self.client_ad.droid.bluetoothRfcommStop()
280                except Exception as err:
281                    self.log.error(
282                        "Error closing client connection: {}".format(err))
283                    return False
284                try:
285                    self.server_ad.droid.bluetoothRfcommStop()
286                except Exception as err:
287                    self.log.error(
288                        "Error closing server connection: {}".format(err))
289                    return False
290        return True
291
292    @test_tracker_info(uuid='155a25be-3e6c-4462-a78f-f6a161b90953')
293    def test_rfcomm_longev_data_elasticity(self):
294        """Longevity test an RFCOMM connection's I/O with changing data size
295
296        Test the longevity of RFCOMM with a basic read/write
297        connect/disconnect sequence. The data being transfered changes
298        in size after each write/read sequence to increase up to 990
299        chars in size and decrease down to 1 in size. This repeats through
300        the entire test in order to exercise different size values being
301        written.
302
303        Steps:
304        1. Establish a bonding between two Android devices.
305        2. Write data to RFCOMM from the client droid.
306        3. Read data from RFCOMM from the server droid.
307        4. Verify data written matches data read.
308        5. Change data size according to above description.
309        6. Repeat steps 2-5 5000 times.
310        7. Disconnect RFCOMM connection.
311        8. Repeat steps 1-6 1000 times.
312
313        Expected Result:
314        Each iteration should read and write to the RFCOMM connection
315        successfully. Each connect and disconnect should be successful.
316
317        Returns:
318          Pass if True
319          Fail if False
320
321        TAGS: Classic, Longevity, RFCOMM
322        Priority: 2
323        """
324        message = "x"
325        resize_toggle = 1
326        for i in range(self.longev_iterations):
327            try:
328                self.log.info("iteration {} connection".format(i + 1))
329                if not orchestrate_rfcomm_connection(self.client_ad,
330                                                     self.server_ad):
331                    return False
332                for n in range(self.write_iterations):
333                    self.log.info("iteration {} data".format(((n + 1) + (
334                        i * self.write_iterations))))
335                    if not write_read_verify_data(
336                            self.client_ad, self.server_ad, message, False):
337                        return False
338                    self.log.info("Iteration {} completed".format(n))
339                    size_of_message = len(message)
340                    #max size is 990 due to a bug in sl4a.
341                    if size_of_message >= 990:
342                        resize_toggle = 0
343                    elif size_of_message <= 1:
344                        resize_toggle = 1
345                    if resize_toggle == 0:
346                        message = "x" * (size_of_message - 1)
347                    else:
348                        message = "x" * (size_of_message + 1)
349                self.client_ad.droid.bluetoothRfcommStop()
350                self.server_ad.droid.bluetoothRfcommStop()
351            except Exception as err:
352                self.log.info("Error in longevity test: {}".format(err))
353                return False
354        return True
355