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"""Tests for Neaby Connection between two Android devices."""
16
17from mobly import asserts
18from mobly import test_runner
19from mobly.snippet import errors
20
21from betocq import nc_base_test
22from betocq import nc_constants
23from betocq import nearby_connection_wrapper
24from betocq import setup_utils
25from betocq import version
26
27
28class NearbyConnectionsFunctionTest(nc_base_test.NCBaseTestClass):
29  """Nearby Connection E2E tests."""
30
31  def __init__(self, configs):
32    super().__init__(configs)
33    self._skipped: bool = False
34    self._test_failure_reason: nc_constants.SingleTestFailureReason = (
35        nc_constants.SingleTestFailureReason.UNINITIALIZED
36    )
37
38  def test_nearby_connections_3p_apis(self):
39    """Test the capability of 3P APIs for Nearby Connections."""
40
41    # Verify that from the 3P snippet, it fails to call 1P APIs
42    with asserts.assert_raises(errors.ApiError):
43      self.advertiser.nearby3p.getLocalEndpointId()
44    with asserts.assert_raises(errors.ApiError):
45      self.discoverer.nearby3p.getLocalEndpointId()
46
47    self._test_result = nc_constants.SingleTestResult()
48    # 1. discovery/advertising
49    nearby_snippet_3p = nearby_connection_wrapper.NearbyConnectionWrapper(
50        self.advertiser,
51        self.discoverer,
52        self.advertiser.nearby3p,
53        self.discoverer.nearby3p,
54        advertising_discovery_medium=nc_constants.NearbyMedium.AUTO,
55        connection_medium=nc_constants.NearbyMedium.AUTO,
56        upgrade_medium=nc_constants.NearbyMedium.AUTO,
57    )
58
59    # 2. create connection
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_3p.start_nearby_connection(
67          timeouts=connection_setup_timeouts,
68          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE,
69      )
70    finally:
71      self._test_failure_reason = nearby_snippet_3p.test_failure_reason
72      self._test_result.file_transfer_nc_setup_quality_info = (
73          nearby_snippet_3p.connection_quality_info
74      )
75
76    # 3. transfer file
77    try:
78      self._test_result.file_transfer_throughput_kbps = (
79          nearby_snippet_3p.transfer_file(
80              nc_constants.TRANSFER_FILE_SIZE_20MB,
81              nc_constants.WIFI_2G_20M_PAYLOAD_TRANSFER_TIMEOUT,
82              nc_constants.PayloadType.FILE,
83          )
84      )
85    finally:
86      self._test_failure_reason = nearby_snippet_3p.test_failure_reason
87
88    # 4. disconnect
89    nearby_snippet_3p.disconnect_endpoint()
90    self._summary_test_results()
91
92  def teardown_test(self) -> None:
93    self._test_result_messages[self.current_test_info.name] = (
94        self._get_test_result_message()
95    )
96    self.record_data({
97        'Test Name': self.current_test_info.name,
98        'sponge_properties': {
99            'result': self._get_test_result_message(),
100        },
101    })
102    super().teardown_test()
103
104  def _summary_test_results(self):
105    """Summarizes test results of all function tests."""
106
107    self.record_data({
108        'Test Class': self.TAG,
109        'sponge_properties': {
110            '00_test_script_verion': version.TEST_SCRIPT_VERSION,
111            '01_source_device_serial': self.discoverer.serial,
112            '02_target_device_serial': self.advertiser.serial,
113            '03_source_GMS_version': setup_utils.dump_gms_version(
114                self.discoverer
115            ),
116            '04_target_GMS_version': setup_utils.dump_gms_version(
117                self.advertiser
118            ),
119            '05_test_result': self._test_result_messages,
120        },
121    })
122
123  def _get_test_result_message(self) -> str:
124    if self._skipped:
125      return (
126          'SKIPPED - not required for the target CUJ: '
127          f'{self.test_parameters.target_cuj_name}'
128      )
129    if (
130        self._test_failure_reason
131        == nc_constants.SingleTestFailureReason.SUCCESS
132    ):
133      return 'PASS'
134    if (
135        self._test_failure_reason
136        is nc_constants.SingleTestFailureReason.FILE_TRANSFER_FAIL
137    ):
138      return 'The Bluetooth performance is really bad.'
139    else:
140      return ''.join([
141          f'FAIL: due to {self._test_failure_reason.name} - ',
142          f'{nc_constants.COMMON_TRIAGE_TIP.get(self._test_failure_reason)}'
143          ])
144
145
146if __name__ == '__main__':
147  test_runner.main()