1#!/usr/bin/env python3.4 2# 3# Copyright 2022 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the 'License'); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of 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, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import collections 18import csv 19import itertools 20import json 21import numpy 22import os 23import time 24from acts import asserts 25from acts import context 26from acts import base_test 27from acts import utils 28from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 29from acts.controllers.utils_lib import ssh 30from acts.controllers import iperf_server as ipf 31from acts_contrib.test_utils.cellular.keysight_5g_testapp import Keysight5GTestApp 32from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils 33from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 34 35from functools import partial 36 37LONG_SLEEP = 10 38MEDIUM_SLEEP = 2 39IPERF_TIMEOUT = 10 40SHORT_SLEEP = 1 41SUBFRAME_LENGTH = 0.001 42STOP_COUNTER_LIMIT = 3 43 44 45class Cellular5GFR2ThroughputTest(base_test.BaseTestClass): 46 """Class to test cellular throughput 47 48 This class implements cellular throughput tests on a lab/callbox setup. 49 The class setups up the callbox in the desired configurations, configures 50 and connects the phone, and runs traffic/iperf throughput. 51 """ 52 53 def __init__(self, controllers): 54 base_test.BaseTestClass.__init__(self, controllers) 55 self.testcase_metric_logger = ( 56 BlackboxMappedMetricLogger.for_test_case()) 57 self.testclass_metric_logger = ( 58 BlackboxMappedMetricLogger.for_test_class()) 59 self.publish_testcase_metrics = True 60 61 def setup_class(self): 62 """Initializes common test hardware and parameters. 63 64 This function initializes hardwares and compiles parameters that are 65 common to all tests in this class. 66 """ 67 self.dut = self.android_devices[-1] 68 self.testclass_params = self.user_params['throughput_test_params'] 69 self.keysight_test_app = Keysight5GTestApp( 70 self.user_params['Keysight5GTestApp']) 71 self.testclass_results = collections.OrderedDict() 72 self.iperf_server = self.iperf_servers[0] 73 self.iperf_client = self.iperf_clients[0] 74 self.remote_server = ssh.connection.SshConnection( 75 ssh.settings.from_config( 76 self.user_params['RemoteServer']['ssh_config'])) 77 if self.testclass_params.get('reload_scpi', 1): 78 self.keysight_test_app.import_scpi_file( 79 self.testclass_params['scpi_file']) 80 # Configure test retries 81 self.user_params['retry_tests'] = [self.__class__.__name__] 82 83 # Turn Airplane mode on 84 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 85 'Can not turn on airplane mode.') 86 87 def teardown_class(self): 88 self.log.info('Turning airplane mode on') 89 try: 90 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 91 'Can not turn on airplane mode.') 92 except: 93 self.log.warning('Cannot perform teardown operations on DUT.') 94 try: 95 self.keysight_test_app.set_cell_state('LTE', 1, 0) 96 self.keysight_test_app.destroy() 97 except: 98 self.log.warning('Cannot perform teardown operations on tester.') 99 self.process_testclass_results() 100 101 def setup_test(self): 102 if self.testclass_params['enable_pixel_logs']: 103 cputils.start_pixel_logger(self.dut) 104 105 def on_retry(self): 106 """Function to control test logic on retried tests. 107 108 This function is automatically executed on tests that are being 109 retried. In this case the function resets wifi, toggles it off and on 110 and sets a retry_flag to enable further tweaking the test logic on 111 second attempts. 112 """ 113 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 114 'Can not turn on airplane mode.') 115 if self.keysight_test_app.get_cell_state('LTE', 'CELL1'): 116 self.log.info('Turning LTE off.') 117 self.keysight_test_app.set_cell_state('LTE', 'CELL1', 0) 118 119 def teardown_test(self): 120 self.log.info('Turing airplane mode on') 121 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 122 'Can not turn on airplane mode.') 123 log_path = os.path.join( 124 context.get_current_context().get_full_output_path(), 'pixel_logs') 125 os.makedirs(self.log_path, exist_ok=True) 126 if self.testclass_params['enable_pixel_logs']: 127 cputils.stop_pixel_logger(self.dut, log_path) 128 self.process_testcase_results() 129 self.pass_fail_check() 130 131 def process_testcase_results(self): 132 if self.current_test_name not in self.testclass_results: 133 return 134 testcase_data = self.testclass_results[self.current_test_name] 135 results_file_path = os.path.join( 136 context.get_current_context().get_full_output_path(), 137 '{}.json'.format(self.current_test_name)) 138 with open(results_file_path, 'w') as results_file: 139 json.dump(wputils.serialize_dict(testcase_data), 140 results_file, 141 indent=4) 142 testcase_result = testcase_data['results'][0] 143 metric_map = { 144 'min_dl_tput': 145 testcase_result['tput_result']['total']['DL']['min_tput'], 146 'max_dl_tput': 147 testcase_result['tput_result']['total']['DL']['max_tput'], 148 'avg_dl_tput': 149 testcase_result['tput_result']['total']['DL']['average_tput'], 150 'theoretical_dl_tput': 151 testcase_result['tput_result']['total']['DL']['theoretical_tput'], 152 'dl_bler': 153 testcase_result['bler_result']['total']['DL']['nack_ratio'] * 100, 154 'min_dl_tput': 155 testcase_result['tput_result']['total']['UL']['min_tput'], 156 'max_dl_tput': 157 testcase_result['tput_result']['total']['UL']['max_tput'], 158 'avg_dl_tput': 159 testcase_result['tput_result']['total']['UL']['average_tput'], 160 'theoretical_dl_tput': 161 testcase_result['tput_result']['total']['UL']['theoretical_tput'], 162 'ul_bler': 163 testcase_result['bler_result']['total']['UL']['nack_ratio'] * 100, 164 'tcp_udp_tput': 165 testcase_result.get('iperf_throughput', float('nan')) 166 } 167 if self.publish_testcase_metrics: 168 for metric_name, metric_value in metric_map.items(): 169 self.testcase_metric_logger.add_metric(metric_name, 170 metric_value) 171 172 def pass_fail_check(self): 173 pass 174 175 def process_testclass_results(self): 176 """Saves CSV with all test results to enable comparison.""" 177 results_file_path = os.path.join( 178 context.get_current_context().get_full_output_path(), 179 'results.csv') 180 with open(results_file_path, 'w', newline='') as csvfile: 181 field_names = [ 182 'Band', 'Channel', 'DL Carriers', 'UL Carriers', 'DL MCS', 183 'DL MIMO', 'UL MCS', 'UL MIMO', 'Cell Power', 184 'DL Min. Throughput', 'DL Max. Throughput', 185 'DL Avg. Throughput', 'DL Theoretical Throughput', 186 'UL Min. Throughput', 'UL Max. Throughput', 187 'UL Avg. Throughput', 'UL Theoretical Throughput', 188 'DL BLER (%)', 'UL BLER (%)', 'TCP/UDP Throughput' 189 ] 190 writer = csv.DictWriter(csvfile, fieldnames=field_names) 191 writer.writeheader() 192 193 for testcase_name, testcase_results in self.testclass_results.items( 194 ): 195 for result in testcase_results['results']: 196 writer.writerow({ 197 'Band': 198 testcase_results['testcase_params']['band'], 199 'Channel': 200 testcase_results['testcase_params']['channel'], 201 'DL Carriers': 202 testcase_results['testcase_params']['num_dl_cells'], 203 'UL Carriers': 204 testcase_results['testcase_params']['num_ul_cells'], 205 'DL MCS': 206 testcase_results['testcase_params']['dl_mcs'], 207 'DL MIMO': 208 testcase_results['testcase_params']['dl_mimo_config'], 209 'UL MCS': 210 testcase_results['testcase_params']['ul_mcs'], 211 'UL MIMO': 212 testcase_results['testcase_params']['ul_mimo_config'], 213 'Cell Power': 214 result['cell_power'], 215 'DL Min. Throughput': 216 result['tput_result']['total']['DL']['min_tput'], 217 'DL Max. Throughput': 218 result['tput_result']['total']['DL']['max_tput'], 219 'DL Avg. Throughput': 220 result['tput_result']['total']['DL']['average_tput'], 221 'DL Theoretical Throughput': 222 result['tput_result']['total']['DL'] 223 ['theoretical_tput'], 224 'UL Min. Throughput': 225 result['tput_result']['total']['UL']['min_tput'], 226 'UL Max. Throughput': 227 result['tput_result']['total']['UL']['max_tput'], 228 'UL Avg. Throughput': 229 result['tput_result']['total']['UL']['average_tput'], 230 'UL Theoretical Throughput': 231 result['tput_result']['total']['UL'] 232 ['theoretical_tput'], 233 'DL BLER (%)': 234 result['bler_result']['total']['DL']['nack_ratio'] * 235 100, 236 'UL BLER (%)': 237 result['bler_result']['total']['UL']['nack_ratio'] * 238 100, 239 'TCP/UDP Throughput': 240 result.get('iperf_throughput', 0) 241 }) 242 243 def setup_tester(self, testcase_params): 244 if not self.keysight_test_app.get_cell_state('LTE', 'CELL1'): 245 self.log.info('Turning LTE on.') 246 self.keysight_test_app.set_cell_state('LTE', 'CELL1', 1) 247 self.log.info('Turning off airplane mode') 248 asserts.assert_true(utils.force_airplane_mode(self.dut, False), 249 'Can not turn on airplane mode.') 250 for cell in testcase_params['dl_cell_list']: 251 self.keysight_test_app.set_cell_band('NR5G', cell, 252 testcase_params['band']) 253 self.keysight_test_app.set_cell_mimo_config( 254 'NR5G', cell, 'DL', testcase_params['dl_mimo_config']) 255 self.keysight_test_app.set_cell_dl_power( 256 'NR5G', cell, testcase_params['cell_power_list'][0], 1) 257 for cell in testcase_params['ul_cell_list']: 258 self.keysight_test_app.set_cell_mimo_config( 259 'NR5G', cell, 'UL', testcase_params['ul_mimo_config']) 260 self.keysight_test_app.configure_contiguous_nr_channels( 261 testcase_params['dl_cell_list'][0], testcase_params['band'], 262 testcase_params['channel']) 263 # Consider configuring schedule quick config 264 self.keysight_test_app.set_nr_cell_schedule_scenario( 265 testcase_params['dl_cell_list'][0], 266 testcase_params['schedule_scenario']) 267 self.keysight_test_app.set_nr_ul_dft_precoding( 268 testcase_params['dl_cell_list'][0], 269 testcase_params['transform_precoding']) 270 self.keysight_test_app.set_nr_cell_mcs( 271 testcase_params['dl_cell_list'][0], testcase_params['dl_mcs'], 272 testcase_params['ul_mcs']) 273 self.keysight_test_app.set_dl_carriers(testcase_params['dl_cell_list']) 274 self.keysight_test_app.set_ul_carriers(testcase_params['ul_cell_list']) 275 self.log.info('Waiting for LTE and applying aggregation') 276 if not self.keysight_test_app.wait_for_cell_status( 277 'LTE', 'CELL1', 'CONN', 60): 278 asserts.fail('DUT did not connect to LTE.') 279 self.keysight_test_app.apply_carrier_agg() 280 self.log.info('Waiting for 5G connection') 281 connected = self.keysight_test_app.wait_for_cell_status( 282 'NR5G', testcase_params['dl_cell_list'][-1], ['ACT', 'CONN'], 60) 283 if not connected: 284 asserts.fail('DUT did not connect to NR.') 285 time.sleep(SHORT_SLEEP) 286 287 def run_iperf_traffic(self, testcase_params): 288 self.iperf_server.start(tag=0) 289 dut_ip = self.dut.droid.connectivityGetIPv4Addresses('rmnet0')[0] 290 if 'iperf_server_address' in self.testclass_params: 291 iperf_server_address = self.testclass_params[ 292 'iperf_server_address'] 293 elif isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 294 iperf_server_address = dut_ip 295 else: 296 iperf_server_address = wputils.get_server_address( 297 self.remote_server, dut_ip, '255.255.255.0') 298 client_output_path = self.iperf_client.start( 299 iperf_server_address, testcase_params['iperf_args'], 0, 300 self.testclass_params['traffic_duration'] + IPERF_TIMEOUT) 301 server_output_path = self.iperf_server.stop() 302 # Parse and log result 303 if testcase_params['use_client_output']: 304 iperf_file = client_output_path 305 else: 306 iperf_file = server_output_path 307 try: 308 iperf_result = ipf.IPerfResult(iperf_file) 309 current_throughput = numpy.mean(iperf_result.instantaneous_rates[ 310 self.testclass_params['iperf_ignored_interval']:-1]) * 8 * ( 311 1.024**2) 312 except: 313 self.log.warning( 314 'ValueError: Cannot get iperf result. Setting to 0') 315 current_throughput = 0 316 return current_throughput 317 318 def _test_nr_throughput_bler(self, testcase_params): 319 """Test function to run cellular throughput and BLER measurements. 320 321 The function runs BLER/throughput measurement after configuring the 322 callbox and DUT. The test supports running PHY or TCP/UDP layer traffic 323 in a variety of band/carrier/mcs/etc configurations. 324 325 Args: 326 testcase_params: dict containing test-specific parameters 327 Returns: 328 result: dict containing throughput results and meta data 329 """ 330 testcase_params = self.compile_test_params(testcase_params) 331 testcase_results = collections.OrderedDict() 332 testcase_results['testcase_params'] = testcase_params 333 testcase_results['results'] = [] 334 # Setup tester and wait for DUT to connect 335 self.setup_tester(testcase_params) 336 # Run test 337 stop_counter = 0 338 for cell_power in testcase_params['cell_power_list']: 339 result = collections.OrderedDict() 340 result['cell_power'] = cell_power 341 # Set DL cell power 342 for cell in testcase_params['dl_cell_list']: 343 self.keysight_test_app.set_cell_dl_power( 344 'NR5G', cell, result['cell_power'], 1) 345 self.keysight_test_app.select_display_tab( 346 'NR5G', testcase_params['dl_cell_list'][0], 'BTHR', 'OTAGRAPH') 347 time.sleep(SHORT_SLEEP) 348 # Start BLER and throughput measurements 349 self.keysight_test_app.start_bler_measurement( 350 'NR5G', testcase_params['dl_cell_list'], 351 testcase_params['bler_measurement_length']) 352 if self.testclass_params['traffic_type'] != 'PHY': 353 result['iperf_throughput'] = self.run_iperf_traffic( 354 testcase_params) 355 if self.testclass_params['log_power_metrics']: 356 if testcase_params[ 357 'bler_measurement_length'] >= 5000 and self.testclass_params[ 358 'traffic_type'] == 'PHY': 359 time.sleep(testcase_params['bler_measurement_length'] / 360 1000 - 5) 361 cputils.log_system_power_metrics(self.dut, verbose=0) 362 else: 363 self.log.warning('Test too short to log metrics') 364 365 result['bler_result'] = self.keysight_test_app.get_bler_result( 366 'NR5G', testcase_params['dl_cell_list'], 367 testcase_params['bler_measurement_length']) 368 result['tput_result'] = self.keysight_test_app.get_throughput( 369 'NR5G', testcase_params['dl_cell_list']) 370 371 # Print Test Summary 372 self.log.info("Cell Power: {}dBm".format(cell_power)) 373 self.log.info( 374 "DL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}" 375 .format( 376 result['tput_result']['total']['DL']['min_tput'], 377 result['tput_result']['total']['DL']['average_tput'], 378 result['tput_result']['total']['DL']['max_tput'], 379 result['tput_result']['total']['DL']['theoretical_tput'])) 380 self.log.info( 381 "UL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}" 382 .format( 383 result['tput_result']['total']['UL']['min_tput'], 384 result['tput_result']['total']['UL']['average_tput'], 385 result['tput_result']['total']['UL']['max_tput'], 386 result['tput_result']['total']['UL']['theoretical_tput'])) 387 self.log.info("DL BLER: {:.2f}%\tUL BLER: {:.2f}%".format( 388 result['bler_result']['total']['DL']['nack_ratio'] * 100, 389 result['bler_result']['total']['UL']['nack_ratio'] * 100)) 390 testcase_results['results'].append(result) 391 if self.testclass_params['traffic_type'] != 'PHY': 392 self.log.info("{} {} Tput: {:.2f} Mbps".format( 393 self.testclass_params['traffic_type'], 394 testcase_params['traffic_direction'], 395 result['iperf_throughput'])) 396 397 if result['bler_result']['total']['DL']['nack_ratio'] * 100 > 99: 398 stop_counter = stop_counter + 1 399 else: 400 stop_counter = 0 401 if stop_counter == STOP_COUNTER_LIMIT: 402 break 403 # Turn off NR cells 404 for cell in testcase_params['dl_cell_list'][::-1]: 405 self.keysight_test_app.set_cell_state('NR5G', cell, 0) 406 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 407 'Can not turn on airplane mode.') 408 409 # Save results 410 self.testclass_results[self.current_test_name] = testcase_results 411 412 def compile_test_params(self, testcase_params): 413 """Function that completes all test params based on the test name. 414 415 Args: 416 testcase_params: dict containing test-specific parameters 417 """ 418 testcase_params['bler_measurement_length'] = int( 419 self.testclass_params['traffic_duration'] / SUBFRAME_LENGTH) 420 testcase_params['cell_power_list'] = numpy.arange( 421 self.testclass_params['cell_power_start'], 422 self.testclass_params['cell_power_stop'], 423 self.testclass_params['cell_power_step']) 424 if self.testclass_params['traffic_type'] == 'PHY': 425 return testcase_params 426 if self.testclass_params['traffic_type'] == 'TCP': 427 testcase_params['iperf_socket_size'] = self.testclass_params.get( 428 'tcp_socket_size', None) 429 testcase_params['iperf_processes'] = self.testclass_params.get( 430 'tcp_processes', 1) 431 elif self.testclass_params['traffic_type'] == 'UDP': 432 testcase_params['iperf_socket_size'] = self.testclass_params.get( 433 'udp_socket_size', None) 434 testcase_params['iperf_processes'] = self.testclass_params.get( 435 'udp_processes', 1) 436 if (testcase_params['traffic_direction'] == 'DL' 437 and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb) 438 ) or (testcase_params['traffic_direction'] == 'UL' 439 and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)): 440 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 441 duration=self.testclass_params['traffic_duration'], 442 reverse_direction=1, 443 traffic_type=self.testclass_params['traffic_type'], 444 socket_size=testcase_params['iperf_socket_size'], 445 num_processes=testcase_params['iperf_processes'], 446 udp_throughput=self.testclass_params['UDP_rates'].get( 447 testcase_params['num_dl_cells'], 448 self.testclass_params['UDP_rates']["default"]), 449 udp_length=1440) 450 testcase_params['use_client_output'] = True 451 elif (testcase_params['traffic_direction'] == 'UL' 452 and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb) 453 ) or (testcase_params['traffic_direction'] == 'DL' 454 and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)): 455 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 456 duration=self.testclass_params['traffic_duration'], 457 reverse_direction=0, 458 traffic_type=self.testclass_params['traffic_type'], 459 socket_size=testcase_params['iperf_socket_size'], 460 num_processes=testcase_params['iperf_processes'], 461 udp_throughput=self.testclass_params['UDP_rates'].get( 462 testcase_params['num_dl_cells'], 463 self.testclass_params['UDP_rates']["default"]), 464 udp_length=1440) 465 testcase_params['use_client_output'] = False 466 return testcase_params 467 468 def generate_test_cases(self, bands, channels, mcs_pair_list, 469 num_dl_cells_list, num_ul_cells_list, 470 dl_mimo_config, ul_mimo_config, **kwargs): 471 """Function that auto-generates test cases for a test class.""" 472 test_cases = ['test_load_scpi'] 473 474 for band, channel, num_ul_cells, num_dl_cells, mcs_pair in itertools.product( 475 bands, channels, num_ul_cells_list, num_dl_cells_list, 476 mcs_pair_list): 477 if num_ul_cells > num_dl_cells: 478 continue 479 if channel not in cputils.PCC_PRESET_MAPPING[band]: 480 continue 481 test_name = 'test_nr_throughput_bler_{}_{}_DL_{}CC_mcs{}_{}_UL_{}CC_mcs{}_{}'.format( 482 band, channel, num_dl_cells, mcs_pair[0], dl_mimo_config, 483 num_ul_cells, mcs_pair[1], ul_mimo_config) 484 test_params = collections.OrderedDict( 485 band=band, 486 channel=channel, 487 dl_mcs=mcs_pair[0], 488 ul_mcs=mcs_pair[1], 489 num_dl_cells=num_dl_cells, 490 num_ul_cells=num_ul_cells, 491 dl_mimo_config=dl_mimo_config, 492 ul_mimo_config=ul_mimo_config, 493 dl_cell_list=list(range(1, num_dl_cells + 1)), 494 ul_cell_list=list(range(1, num_ul_cells + 1)), 495 **kwargs) 496 setattr(self, test_name, 497 partial(self._test_nr_throughput_bler, test_params)) 498 test_cases.append(test_name) 499 return test_cases 500 501 502class Cellular5GFR2_DL_ThroughputTest(Cellular5GFR2ThroughputTest): 503 504 def __init__(self, controllers): 505 super().__init__(controllers) 506 self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 507 ['low', 'mid', 'high'], 508 [(16, 4), (27, 4)], 509 list(range(1, 9)), 510 list(range(1, 3)), 511 dl_mimo_config='N2X2', 512 ul_mimo_config='N1X1', 513 schedule_scenario="FULL_TPUT", 514 traffic_direction='DL', 515 transform_precoding=0) 516 517 518class Cellular5GFR2_CP_UL_ThroughputTest(Cellular5GFR2ThroughputTest): 519 520 def __init__(self, controllers): 521 super().__init__(controllers) 522 self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 523 ['low', 'mid', 'high'], 524 [(4, 16), (4, 27)], [1], [1], 525 dl_mimo_config='N2X2', 526 ul_mimo_config='N1X1', 527 schedule_scenario="FULL_TPUT", 528 traffic_direction='UL', 529 transform_precoding=0) 530 self.tests.extend( 531 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 532 ['low', 'mid', 'high'], 533 [(4, 16), (4, 27)], [1], [1], 534 dl_mimo_config='N2X2', 535 ul_mimo_config='N2X2', 536 schedule_scenario="FULL_TPUT", 537 traffic_direction='UL', 538 transform_precoding=0)) 539 self.tests.extend( 540 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 541 ['low', 'mid', 'high'], 542 [(4, 16), (4, 27)], [2], [2], 543 dl_mimo_config='N2X2', 544 ul_mimo_config='N2X2', 545 schedule_scenario="FULL_TPUT", 546 traffic_direction='UL', 547 transform_precoding=0)) 548 self.tests.extend( 549 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 550 ['low', 'mid', 'high'], 551 [(4, 16), (4, 27)], [3], [3], 552 dl_mimo_config='N2X2', 553 ul_mimo_config='N2X2', 554 schedule_scenario="UL_RMC", 555 traffic_direction='UL', 556 transform_precoding=0)) 557 self.tests.extend( 558 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 559 ['low', 'mid', 'high'], 560 [(4, 16), (4, 27)], [4], [4], 561 dl_mimo_config='N2X2', 562 ul_mimo_config='N2X2', 563 schedule_scenario="FULL_TPUT", 564 traffic_direction='UL', 565 transform_precoding=0)) 566 567 568class Cellular5GFR2_DFTS_UL_ThroughputTest(Cellular5GFR2ThroughputTest): 569 570 def __init__(self, controllers): 571 super().__init__(controllers) 572 self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 573 ['low', 'mid', 'high'], 574 [(4, 16), (4, 27)], [1], [1], 575 dl_mimo_config='N2X2', 576 ul_mimo_config='N1X1', 577 schedule_scenario="FULL_TPUT", 578 traffic_direction='UL', 579 transform_precoding=1) 580 self.tests.extend( 581 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 582 ['low', 'mid', 'high'], 583 [(4, 16), (4, 27)], [1], [1], 584 dl_mimo_config='N2X2', 585 ul_mimo_config='N2X2', 586 schedule_scenario="FULL_TPUT", 587 traffic_direction='UL', 588 transform_precoding=1)) 589 self.tests.extend( 590 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 591 ['low', 'mid', 'high'], 592 [(4, 16), (4, 27)], [2], [2], 593 dl_mimo_config='N2X2', 594 ul_mimo_config='N2X2', 595 schedule_scenario="FULL_TPUT", 596 traffic_direction='UL', 597 transform_precoding=1)) 598 self.tests.extend( 599 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 600 ['low', 'mid', 'high'], 601 [(4, 16), (4, 27)], [3], [3], 602 dl_mimo_config='N2X2', 603 ul_mimo_config='N2X2', 604 schedule_scenario="FULL_TPUT", 605 traffic_direction='UL', 606 transform_precoding=1)) 607 self.tests.extend( 608 self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], 609 ['low', 'mid', 'high'], 610 [(4, 16), (4, 27)], [4], [4], 611 dl_mimo_config='N2X2', 612 ul_mimo_config='N2X2', 613 schedule_scenario="FULL_TPUT", 614 traffic_direction='UL', 615 transform_precoding=1)) 616 617 618class Cellular5GFR2_DL_FrequecySweep_ThroughputTest(Cellular5GFR2ThroughputTest 619 ): 620 621 def __init__(self, controllers): 622 super().__init__(controllers) 623 dl_frequency_sweep_params = self.user_params['throughput_test_params'][ 624 'dl_frequency_sweep'] 625 self.tests = self.generate_test_cases(dl_frequency_sweep_params, 626 [(16, 4), (27, 4)], 627 schedule_scenario="FULL_TPUT", 628 traffic_direction='DL', 629 transform_precoding=0, 630 dl_mimo_config='N2X2', 631 ul_mimo_config='N1X1') 632 633 def generate_test_cases(self, dl_frequency_sweep_params, mcs_pair_list, 634 **kwargs): 635 """Function that auto-generates test cases for a test class.""" 636 test_cases = ['test_load_scpi'] 637 638 for band, band_config in dl_frequency_sweep_params.items(): 639 for num_dl_cells_str, sweep_config in band_config.items(): 640 num_dl_cells = int(num_dl_cells_str[0]) 641 num_ul_cells = 1 642 freq_vector = numpy.arange(sweep_config[0], sweep_config[1], 643 sweep_config[2]) 644 for freq in freq_vector: 645 for mcs_pair in mcs_pair_list: 646 test_name = 'test_nr_throughput_bler_{}_{}MHz_DL_{}CC_mcs{}_UL_{}CC_mcs{}'.format( 647 band, freq, num_dl_cells, mcs_pair[0], 648 num_ul_cells, mcs_pair[1]) 649 test_params = collections.OrderedDict( 650 band=band, 651 channel=freq, 652 dl_mcs=mcs_pair[0], 653 ul_mcs=mcs_pair[1], 654 num_dl_cells=num_dl_cells, 655 num_ul_cells=num_ul_cells, 656 dl_cell_list=list(range(1, num_dl_cells + 1)), 657 ul_cell_list=list(range(1, num_ul_cells + 1)), 658 **kwargs) 659 setattr( 660 self, test_name, 661 partial(self._test_nr_throughput_bler, 662 test_params)) 663 test_cases.append(test_name) 664 return test_cases 665