1#  Copyright (C) 2023 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"""Esim transfer stress test, only use Bluetooth connection."""
16
17import dataclasses
18import datetime
19import logging
20import os
21import sys
22import time
23
24# check the python version
25if sys.version_info < (3,10):
26  logging.error('The test only can run on python 3.10 and above')
27  exit()
28
29from mobly import asserts
30from mobly import base_test
31from mobly import test_runner
32
33# Allows local imports to be resolved via relative path, so the test can be run
34# without building.
35_performance_test_dir = os.path.dirname(os.path.dirname(__file__))
36if _performance_test_dir not in sys.path:
37  sys.path.append(_performance_test_dir)
38
39from performance_test import nc_base_test
40from performance_test import nc_constants
41from performance_test import nearby_connection_wrapper
42from performance_test import setup_utils
43
44_TEST_SCRIPT_VERSION = '1.6'
45
46_DELAY_BETWEEN_EACH_TEST_CYCLE = datetime.timedelta(seconds=5)
47_TRANSFER_FILE_SIZE_1MB = 1024
48
49_PERFORMANCE_TEST_REPEAT_COUNT = 100
50_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR = 5
51
52
53class EsimTransferStressTest(nc_base_test.NCBaseTestClass):
54  """Esim transfer stress test."""
55
56  @dataclasses.dataclass(frozen=False)
57  class EsimTransferTestMetrics:
58    """Metrics data for esim transfer test."""
59    discovery_latencies: list[datetime.timedelta] = dataclasses.field(
60        default_factory=list[datetime.timedelta])
61    connection_latencies: list[datetime.timedelta] = dataclasses.field(
62        default_factory=list[datetime.timedelta])
63    bt_transfer_throughputs_kbps: list[float] = dataclasses.field(
64        default_factory=list[float])
65
66  # @typing.override
67  def __init__(self, configs):
68    super().__init__(configs)
69    self._test_result = nc_constants.SingleTestResult()
70    self._esim_transfer_test_metrics = self.EsimTransferTestMetrics()
71    self._test_script_version = _TEST_SCRIPT_VERSION
72
73  # @typing.override
74  def setup_class(self):
75    super().setup_class()
76    self.performance_test_iterations = getattr(
77        self.test_esim_transfer_performance, base_test.ATTR_REPEAT_CNT)
78    logging.info('performance test iterations: %s',
79                 self.performance_test_iterations)
80
81  @base_test.repeat(
82      count=_PERFORMANCE_TEST_REPEAT_COUNT,
83      max_consecutive_error=_PERFORMANCE_TEST_MAX_CONSECUTIVE_ERROR)
84  def test_esim_transfer_performance(self):
85    """Esim transfer stress test, which only transfer data through BT."""
86    try:
87      self._mimic_esim_transfer()
88    finally:
89      self._write_current_test_report()
90      self._collect_current_test_metrics()
91      time.sleep(_DELAY_BETWEEN_EACH_TEST_CYCLE.total_seconds())
92
93  def _mimic_esim_transfer(self):
94    if self.test_parameters.toggle_airplane_mode_target_side:
95      setup_utils.toggle_airplane_mode(self.advertiser)
96    if self.test_parameters.reset_wifi_connection:
97      self._reset_wifi_connection()
98    self._test_result = nc_constants.SingleTestResult()
99    # 1. connect to wifi
100    wifi_ssid = self.test_parameters.wifi_ssid
101    wifi_password = self.test_parameters.wifi_password
102    if wifi_ssid:
103      discoverer_wifi_latency = setup_utils.connect_to_wifi_wlan_till_success(
104          self.discoverer, wifi_ssid, wifi_password
105      )
106      self.discoverer.log.info(
107          'connecting to wifi in '
108          f'{round(discoverer_wifi_latency.total_seconds())} s'
109      )
110      advertiser_wlan_latency = setup_utils.connect_to_wifi_wlan_till_success(
111          self.advertiser, wifi_ssid, wifi_password)
112      self.advertiser.log.info(
113          'connecting to wifi in '
114          f'{round(advertiser_wlan_latency.total_seconds())} s')
115      self.advertiser.log.info(
116          self.advertiser.nearby.wifiGetConnectionInfo().get('mFrequency')
117      )
118
119    # 2. set up BT connection
120    advertising_discovery_medium = nc_constants.NearbyMedium(
121        self.test_parameters.advertising_discovery_medium
122    )
123    nearby_snippet_1 = nearby_connection_wrapper.NearbyConnectionWrapper(
124        self.advertiser,
125        self.discoverer,
126        self.advertiser.nearby,
127        self.discoverer.nearby,
128        advertising_discovery_medium=advertising_discovery_medium,
129        connection_medium=nc_constants.NearbyMedium.BT_ONLY,
130        upgrade_medium=nc_constants.NearbyMedium.BT_ONLY,
131    )
132    connection_setup_timeouts = nc_constants.ConnectionSetupTimeouts(
133        nc_constants.FIRST_DISCOVERY_TIMEOUT,
134        nc_constants.FIRST_CONNECTION_INIT_TIMEOUT,
135        nc_constants.FIRST_CONNECTION_RESULT_TIMEOUT)
136
137    try:
138      nearby_snippet_1.start_nearby_connection(
139          timeouts=connection_setup_timeouts,
140          medium_upgrade_type=nc_constants.MediumUpgradeType.NON_DISRUPTIVE)
141    finally:
142      self._test_result.connection_setup_quality_info = (
143          nearby_snippet_1.connection_quality_info
144      )
145
146    # 3. transfer file through bluetooth
147    file_1_mb = _TRANSFER_FILE_SIZE_1MB
148    self._test_result.bt_transfer_throughput_kbps = (
149        nearby_snippet_1.transfer_file(
150            file_1_mb, nc_constants.FILE_1M_PAYLOAD_TRANSFER_TIMEOUT,
151            nc_constants.PayloadType.FILE))
152    # 4. disconnect
153    nearby_snippet_1.disconnect_endpoint()
154
155  def _write_current_test_report(self):
156    """Writes test report for each iteration."""
157
158    quality_info = {
159        'bt connection': (
160            self._test_result.connection_setup_quality_info.get_dict()),
161        'bt_kBps': self._test_result.bt_transfer_throughput_kbps,
162    }
163    test_report = {'quality_info': quality_info}
164
165    self.discoverer.log.info(test_report)
166    self.record_data({
167        'Test Class': self.TAG,
168        'Test Name': self.current_test_info.name,
169        'properties': test_report,
170    })
171
172  def _collect_current_test_metrics(self):
173    """Collects test result metrics for each iteration."""
174    self._esim_transfer_test_metrics.discovery_latencies.append(
175        self._test_result.connection_setup_quality_info.discovery_latency
176    )
177    self._esim_transfer_test_metrics.connection_latencies.append(
178        self._test_result.connection_setup_quality_info.connection_latency
179    )
180    self._esim_transfer_test_metrics.bt_transfer_throughputs_kbps.append(
181        self._test_result.bt_transfer_throughput_kbps
182    )
183
184  # @typing.override
185  def _summary_test_results(self):
186    """Summarizes test results of all iterations."""
187    bt_transfer_stats = self._stats_throughput_result(
188        'BT',
189        self._esim_transfer_test_metrics.bt_transfer_throughputs_kbps,
190        nc_constants.BT_TRANSFER_SUCCESS_RATE_TARGET_PERCENTAGE,
191        self.test_parameters.bt_transfer_throughput_median_benchmark_kbps)
192
193    discovery_stats = self._stats_latency_result(
194        self._esim_transfer_test_metrics.discovery_latencies)
195    connection_stats = self._stats_latency_result(
196        self._esim_transfer_test_metrics.connection_latencies)
197
198    passed = True
199    result_message = 'Passed'
200    fail_message = ''
201    if bt_transfer_stats.fail_targets:
202      fail_message += self._generate_target_fail_message(
203          bt_transfer_stats.fail_targets)
204    if fail_message:
205      passed = False
206      result_message = 'Test Failed due to:\n' + fail_message
207
208    detailed_stats = {
209        '0 test iterations': self.performance_test_iterations,
210        '1 Completed BT transfer': f'{bt_transfer_stats.success_count}',
211        '2 BT transfer failures': {
212            '1 discovery': discovery_stats.failure_count,
213            '2 connection': connection_stats.failure_count,
214            '3 transfer': self.performance_test_iterations - (
215                bt_transfer_stats.success_count),
216        },
217        '3 50% and 95% of BT transfer speed (KBps)': (
218            f'{bt_transfer_stats.percentile_50_kbps}'
219            f' / {bt_transfer_stats.percentile_95_kbps}'),
220        '4 50% and 95% of discovery latency(sec)': (
221            f'{discovery_stats.percentile_50}'
222            f' / {discovery_stats.percentile_95}'),
223
224        '5 50% and 95% of connection latency(sec)': (
225            f'{connection_stats.percentile_50}'
226            f' / {connection_stats.percentile_95}'),
227    }
228
229    self.record_data({
230        'Test Class': self.TAG,
231        'properties': {
232            'test_script_version': self._test_script_version,
233            '00_test_report_alias_name': (
234                self.test_parameters.test_report_alias_name),
235            '01_test_result': result_message,
236            '02_source_device_serial': self.discoverer.serial,
237            '03_target_device_serial': self.advertiser.serial,
238            '04_source_GMS_version': setup_utils.dump_gms_version(
239                self.discoverer),
240            '05_target_GMS_version': setup_utils.dump_gms_version(
241                self.advertiser),
242            '06_detailed_stats': detailed_stats
243            }
244        })
245
246    asserts.assert_true(passed, result_message)
247
248
249if __name__ == '__main__':
250  test_runner.main()
251