1#  Copyright (C) 2024 The Android Open Source Project
2#
3#  Licensed under the Apache License, Version 2.0 (the "License");
4#  you may not use this file except in compliance with the License.
5#  You may obtain a copy of the License at
6#
7#       http://www.apache.org/licenses/LICENSE-2.0
8#
9#  Unless required by applicable law or agreed to in writing, software
10#  distributed under the License is distributed on an "AS IS" BASIS,
11#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12#  See the License for the specific language governing permissions and
13#  limitations under the License.
14
15"""Bluetooth nearby connection functional test actor."""
16
17from mobly import asserts
18
19from betocq import nc_constants
20from betocq import nearby_connection_wrapper
21from betocq.function_tests import function_test_actor_base
22
23
24class BtMultiplexFunctionTestActor(
25    function_test_actor_base.FunctionTestActorBase):
26  """The actor for running BT/BlE multiplex function test.
27
28  It allows to setup multiple BT connections.
29  """
30
31  def test_bt_multiplex_connections(self):
32    """Test the capability of setting up two BT connections (multiplex).
33
34    This is only required for specific CUJ: quick start
35    step 1: set up 2 BT connections
36    step 2: transfer a small file with the 2nd BT connection
37    """
38    if (
39        self.test_parameters.target_cuj_name
40        != nc_constants.TARGET_CUJ_QUICK_START
41        and not self.test_parameters.requires_bt_multiplex
42    ):
43      self._skipped = True
44      asserts.skip(
45          'BT multiplex is not required for this CUJ -'
46          f' {self.test_parameters.target_cuj_name}'
47      )
48
49    # 1. set up 1st BT connection
50    advertising_discovery_medium = nc_constants.NearbyMedium.BLE_ONLY
51    nearby_snippet_2 = nearby_connection_wrapper.NearbyConnectionWrapper(
52        self.advertiser,
53        self.discoverer,
54        self.advertiser.nearby2,
55        self.discoverer.nearby2,
56        advertising_discovery_medium=advertising_discovery_medium,
57        connection_medium=nc_constants.NearbyMedium.BT_ONLY,
58        upgrade_medium=nc_constants.NearbyMedium.BT_ONLY,
59    )
60    connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
61        nc_constants.FIRST_DISCOVERY_TIMEOUT,
62        nc_constants.FIRST_CONNECTION_INIT_TIMEOUT,
63        nc_constants.FIRST_CONNECTION_RESULT_TIMEOUT)
64
65    try:
66      nearby_snippet_2.start_nearby_connection(
67          timeouts=connection_setup_timeouts,
68          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE)
69    finally:
70      self._test_failure_reason = nearby_snippet_2.test_failure_reason
71      self._test_result.quality_info = (
72          nearby_snippet_2.connection_quality_info
73      )
74    # 2nd bt connection
75    nearby_snippet = nearby_connection_wrapper.NearbyConnectionWrapper(
76        self.advertiser,
77        self.discoverer,
78        self.advertiser.nearby,
79        self.discoverer.nearby,
80        advertising_discovery_medium=advertising_discovery_medium,
81        connection_medium=nc_constants.NearbyMedium.BT_ONLY,
82        upgrade_medium=nc_constants.NearbyMedium.BT_ONLY,
83    )
84
85    second_connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
86        nc_constants.SECOND_DISCOVERY_TIMEOUT,
87        nc_constants.SECOND_CONNECTION_INIT_TIMEOUT,
88        nc_constants.SECOND_CONNECTION_RESULT_TIMEOUT)
89    try:
90      nearby_snippet.start_nearby_connection(
91          timeouts=second_connection_setup_timeouts,
92          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE)
93    finally:
94      self._test_failure_reason = nearby_snippet.test_failure_reason
95      self._test_result.prior_nc_quality_info = (
96          nearby_snippet.connection_quality_info
97      )
98
99    # 2. transfer file through bluetooth
100    try:
101      self._test_result.file_transfer_throughput_kbps = (
102          nearby_snippet.transfer_file(
103              nc_constants.TRANSFER_FILE_SIZE_1KB,
104              nc_constants.BT_1K_PAYLOAD_TRANSFER_TIMEOUT,
105              nc_constants.PayloadType.FILE))
106    finally:
107      self._test_failure_reason = nearby_snippet.test_failure_reason
108
109    # 3. disconnect
110    nearby_snippet_2.disconnect_endpoint()
111    nearby_snippet.disconnect_endpoint()
112
113  def get_test_result_message(self) -> str:
114    if self._skipped:
115      return (
116          'SKIPPED - not required for the target CUJ: '
117          f'{self.test_parameters.target_cuj_name}'
118      )
119    if (
120        self._test_failure_reason
121        == nc_constants.SingleTestFailureReason.SUCCESS
122    ):
123      return 'PASS'
124    if (
125        self._test_failure_reason
126        is nc_constants.SingleTestFailureReason.FILE_TRANSFER_FAIL
127    ):
128      return 'The Bluetooth performance is really bad.'
129    else:
130      return ''.join([
131          f'FAIL: due to {self._test_failure_reason.name} - ',
132          f'{nc_constants.COMMON_TRIAGE_TIP.get(self._test_failure_reason)}'
133          ])
134