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"""Base class for all fixed wifi medium function test actors."""
16import logging
17from mobly import asserts
18
19from betocq import nc_constants
20from betocq import nearby_connection_wrapper
21from betocq import setup_utils
22from betocq.function_tests import function_test_actor_base
23
24
25class FixedWifiMediumFunctionTestActor(
26    function_test_actor_base.FunctionTestActorBase):
27  """The base of actors for running the specified fixed wifi D2D medium."""
28
29  def run_fixed_wifi_medium_test(
30      self, wifi_medium: nc_constants.NearbyMedium) -> None:
31    """upgrade the medium from BT to the specified medium and transfer a sample data."""
32    self._wifi_medium_under_test = wifi_medium
33    self._test_result = nc_constants.SingleTestResult()
34
35    # 1. set up BT and WiFi connection
36    advertising_discovery_medium = nc_constants.NearbyMedium(
37        self.test_parameters.advertising_discovery_medium
38    )
39    nearby_snippet = nearby_connection_wrapper.NearbyConnectionWrapper(
40        self.advertiser,
41        self.discoverer,
42        self.advertiser.nearby,
43        self.discoverer.nearby,
44        advertising_discovery_medium=advertising_discovery_medium,
45        connection_medium=nc_constants.NearbyMedium.BT_ONLY,
46        upgrade_medium=self._wifi_medium_under_test,
47    )
48
49    connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
50        nc_constants.FIRST_DISCOVERY_TIMEOUT,
51        nc_constants.FIRST_CONNECTION_INIT_TIMEOUT,
52        nc_constants.FIRST_CONNECTION_RESULT_TIMEOUT)
53
54    try:
55      nearby_snippet.start_nearby_connection(
56          timeouts=connection_setup_timeouts,
57          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE)
58    finally:
59      self._test_failure_reason = nearby_snippet.test_failure_reason
60      self._test_result.quality_info = (
61          nearby_snippet.connection_quality_info
62      )
63
64    # 2. transfer file through WiFi
65    try:
66      self._test_result.file_transfer_throughput_kbps = (
67          nearby_snippet.transfer_file(
68              nc_constants.TRANSFER_FILE_SIZE_1KB,
69              nc_constants.WIFI_1K_PAYLOAD_TRANSFER_TIMEOUT,
70              nc_constants.PayloadType.FILE))
71    finally:
72      self._test_failure_reason = nearby_snippet.test_failure_reason
73
74    # 3. disconnect
75    nearby_snippet.disconnect_endpoint()
76
77  def connect_to_wifi(self):
78    if self.test_parameters.toggle_airplane_mode_target_side:
79      setup_utils.toggle_airplane_mode(self.advertiser)
80
81    wifi_ssid, wifi_password = self._get_wifi_ssid_password()
82
83    if not wifi_ssid:
84      self._test_failure_reason = (
85          nc_constants.SingleTestFailureReason.AP_IS_NOT_CONFIGURED
86      )
87      asserts.fail('Wifi AP must be specified.')
88
89    logging.info('connect to wifi: %s', wifi_ssid)
90
91    # source device
92    self._test_failure_reason = (
93        nc_constants.SingleTestFailureReason.SOURCE_WIFI_CONNECTION
94    )
95    discoverer_wifi_latency = setup_utils.connect_to_wifi_sta_till_success(
96        self.discoverer, wifi_ssid, wifi_password
97    )
98    self.discoverer.log.info(
99        'connecting to wifi in '
100        f'{round(discoverer_wifi_latency.total_seconds())} s'
101    )
102    # target device
103    self._test_failure_reason = (
104        nc_constants.SingleTestFailureReason.TARGET_WIFI_CONNECTION
105    )
106    advertiser_wlan_latency = setup_utils.connect_to_wifi_sta_till_success(
107        self.advertiser, wifi_ssid, wifi_password)
108    self.advertiser.log.info(
109        'connecting to wifi in '
110        f'{round(advertiser_wlan_latency.total_seconds())} s')
111    self.advertiser.log.info(
112        self.advertiser.nearby.wifiGetConnectionInfo().get('mFrequency')
113    )
114    self._test_failure_reason = (
115        nc_constants.SingleTestFailureReason.SUCCESS
116    )
117
118  def get_test_result_message(self) -> str:
119    if (
120        self._test_failure_reason
121        == nc_constants.SingleTestFailureReason.SUCCESS
122    ):
123      return 'PASS'
124    if (
125        self._test_failure_reason
126        == nc_constants.SingleTestFailureReason.WIFI_MEDIUM_UPGRADE
127    ):
128      return ''.join([
129          f'{self._test_failure_reason.name} - ',
130          self._get_medium_upgrade_failure_tip()
131      ])
132    if (
133        self._test_failure_reason
134        == nc_constants.SingleTestFailureReason.FILE_TRANSFER_FAIL
135    ):
136      return ''.join([
137          f'{self._test_failure_reason.name} - ',
138          self._get_file_transfer_failure_tip()
139      ])
140    return ''.join([
141        f'{self._test_failure_reason.name} - ',
142        nc_constants.COMMON_TRIAGE_TIP.get(self._test_failure_reason),
143    ])
144
145  def _get_medium_upgrade_failure_tip(self) -> str:
146    return nc_constants.MEDIUM_UPGRADE_FAIL_TRIAGE_TIPS.get(
147        self._wifi_medium_under_test, ' unsupported test medium')
148
149  def _get_file_transfer_failure_tip(self) -> str:
150    if self._wifi_medium_under_test is not None:
151      return f'{self._wifi_medium_under_test.name}: the performance is too bad.'
152    asserts.fail('unexpected calling of _get_file_transfer_failure_tip')
153