| # Copyright (c) 2021 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.server.cros.network import iperf_runner |
| from autotest_lib.server.cros.network import iperf_session |
| from autotest_lib.server.cros.network import netperf_runner |
| from autotest_lib.server.cros.network import netperf_session |
| |
| |
| class PerfTestTypes(object): |
| """These are the different performance test types that are supported by |
| autotest. The are defined from perspective of the Device Under Test, so for |
| example 'tcp_rx' refers to a performance test of data transfer from a remote |
| server to the DUT using the TCP protocol. |
| """ |
| TEST_TYPE_TCP_TX = 'tcp_tx' |
| TEST_TYPE_TCP_RX = 'tcp_rx' |
| TEST_TYPE_TCP_BIDIRECTIONAL = 'tcp_bidirectional' |
| TEST_TYPE_UDP_TX = 'udp_tx' |
| TEST_TYPE_UDP_RX = 'udp_rx' |
| TEST_TYPE_UDP_BIDIRECTIONAL = 'udp_bidirectional' |
| |
| |
| class PerfTestManager(object): |
| """Manager for Performance tests. This class provides a unified API to allow |
| callers run performance tests using the supported tools. |
| """ |
| |
| # TODO(b:195574472): Add support for iperf in this class. |
| |
| DEFAULT_TEST_TIME = 10 |
| |
| def __init__(self, use_iperf): |
| """Construct a PerfTestManager. |
| |
| TODO(b:198343041) iperf2 bidirectional tests are unreliable, so we |
| always use netperf for bidirectional tests. |
| |
| @param bool use_iperf True if the tests should use iperf, false if the |
| tests should use netperf. |
| """ |
| self._use_iperf = use_iperf |
| |
| def get_config(self, test_type, test_time=DEFAULT_TEST_TIME): |
| """Get a config object for a performance tests based on the test |
| type and other parameters. Will return either a NetperfConfig or |
| IperfConfig based on the use_iperf value of the class. |
| |
| @param test_type string, test type from performance_test_types. |
| @param test_time int number of seconds to run the test for. |
| |
| @return NetperfConfig or IperfConfig object. |
| """ |
| # (b:198343041): Always use netperf for bidirectional tests. |
| if self._use_iperf and test_type not in [ |
| PerfTestTypes.TEST_TYPE_TCP_BIDIRECTIONAL, |
| PerfTestTypes.TEST_TYPE_UDP_BIDIRECTIONAL |
| ]: |
| return iperf_runner.IperfConfig( |
| self._iperf_type_from_perf_type(test_type)) |
| return netperf_runner.NetperfConfig( |
| self._netperf_type_from_perf_type(test_type), |
| test_time=test_time) |
| |
| def get_session(self, |
| test_type, |
| test_device_proxy, |
| peer_device_proxy, |
| test_device_interface=None, |
| peer_device_interface=None, |
| ignore_failures=False): |
| """Get a Session object for a set of performance tests. Will return |
| either a NetperfSession or IperfSession based on the use_iperf value of |
| the class. |
| |
| @param test_device_proxy: WiFiClient object for the device-under-test. |
| @param peer_device_proxy: LinuxSystem object for the performance testing |
| peer of the DUT. |
| @param test_device_interface Interface object for the test device. |
| @param peer_device_interface Interface object for the peer device. |
| |
| @return NetperfSession object. |
| """ |
| # (b:198343041) Always use netperf for bidirectional tests. |
| if self._use_iperf and test_type not in [ |
| PerfTestTypes.TEST_TYPE_TCP_BIDIRECTIONAL, |
| PerfTestTypes.TEST_TYPE_UDP_BIDIRECTIONAL |
| ]: |
| if test_type in [ |
| PerfTestTypes.TEST_TYPE_TCP_TX, |
| PerfTestTypes.TEST_TYPE_UDP_TX |
| ]: |
| return iperf_session.IperfSession( |
| test_device_proxy, |
| peer_device_proxy, |
| client_interface=test_device_interface, |
| server_interface=peer_device_interface, |
| ignore_failures=ignore_failures) |
| if test_type in [ |
| PerfTestTypes.TEST_TYPE_TCP_RX, |
| PerfTestTypes.TEST_TYPE_UDP_RX |
| ]: |
| return iperf_session.IperfSession( |
| peer_device_proxy, |
| test_device_proxy, |
| client_interface=peer_device_interface, |
| server_interface=test_device_interface, |
| ignore_failures=ignore_failures) |
| |
| raise error.TestFail( |
| 'Test type %s is not supported by this test.' % test_type) |
| |
| return netperf_session.NetperfSession( |
| test_device_proxy, |
| peer_device_proxy, |
| client_interface=test_device_interface, |
| server_interface=peer_device_interface, |
| ignore_failures=ignore_failures) |
| |
| def get_result(self, results): |
| """Get a single performance result from a list of results. |
| |
| @param results list of IperfResults or NetperfResults. |
| |
| @return a single IperfResult or NetperfResult which represents the |
| distribution of results. |
| """ |
| # All the results will be of the same type, so we can safely check the |
| # first result only. |
| if isinstance(results[0], iperf_runner.IperfResult): |
| return iperf_runner.IperfResult.from_samples(results) |
| if isinstance(results[0], netperf_runner.NetperfResult): |
| return netperf_runner.NetperfResult.from_samples(results) |
| raise error.TestFail('Invalid test result type: %s' % type(results)) |
| |
| def _netperf_type_from_perf_type(self, test_type): |
| """Convert a performance test type to a netperf test type. |
| |
| @param test_type string, test type from PerfTestTypes. |
| |
| @return string netperf test type that corresponds to the generic test type. |
| """ |
| if test_type == PerfTestTypes.TEST_TYPE_TCP_TX: |
| return netperf_runner.NetperfConfig.TEST_TYPE_TCP_STREAM |
| elif test_type == PerfTestTypes.TEST_TYPE_TCP_RX: |
| return netperf_runner.NetperfConfig.TEST_TYPE_TCP_MAERTS |
| elif test_type == PerfTestTypes.TEST_TYPE_TCP_BIDIRECTIONAL: |
| return netperf_runner.NetperfConfig.TEST_TYPE_TCP_BIDIRECTIONAL |
| elif test_type == PerfTestTypes.TEST_TYPE_UDP_TX: |
| return netperf_runner.NetperfConfig.TEST_TYPE_UDP_STREAM |
| elif test_type == PerfTestTypes.TEST_TYPE_UDP_RX: |
| return netperf_runner.NetperfConfig.TEST_TYPE_UDP_MAERTS |
| elif test_type == PerfTestTypes.TEST_TYPE_UDP_BIDIRECTIONAL: |
| return netperf_runner.NetperfConfig.TEST_TYPE_UDP_BIDIRECTIONAL |
| raise error.TestFail( |
| 'Test type %s is not supported by netperf_runner.' % test_type) |
| |
| def _iperf_type_from_perf_type(self, test_type): |
| """Convert a performance test type to an iperf test type |
| |
| @param test_type string, test type from PerfTestTypes. |
| |
| @return string iperf test type that corresponds to the generic test type. |
| """ |
| if test_type == PerfTestTypes.TEST_TYPE_TCP_TX: |
| return iperf_runner.IperfConfig.IPERF_TEST_TYPE_TCP_TX |
| elif test_type == PerfTestTypes.TEST_TYPE_TCP_RX: |
| return iperf_runner.IperfConfig.IPERF_TEST_TYPE_TCP_RX |
| elif test_type == PerfTestTypes.TEST_TYPE_TCP_BIDIRECTIONAL: |
| return iperf_runner.IperfConfig.IPERF_TEST_TYPE_TCP_BIDIRECTIONAL |
| elif test_type == PerfTestTypes.TEST_TYPE_UDP_TX: |
| return iperf_runner.IperfConfig.IPERF_TEST_TYPE_UDP_TX |
| elif test_type == PerfTestTypes.TEST_TYPE_UDP_RX: |
| return iperf_runner.IperfConfig.IPERF_TEST_TYPE_UDP_RX |
| elif test_type == PerfTestTypes.TEST_TYPE_UDP_BIDIRECTIONAL: |
| return iperf_runner.IperfConfig.IPERF_TEST_TYPE_UDP_BIDIRECTIONAL |
| raise error.TestFail( |
| 'Test type %s is not supported by netperf_runner.' % test_type) |