| # Copyright 2015 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Ping connection test. |
| |
| Description |
| ----------- |
| Tests network connection by pinging a host for a period of time. Then checks |
| the percentage of successful pings is above some threshold or not. |
| |
| If moving_window_size is set, it will check the percentage within the moving |
| window during the test. It always checks the percentage of successful pings for |
| the whole duration at the end of the test. |
| |
| Test Procedure |
| -------------- |
| This is an automated test without user interaction. |
| |
| Dependency |
| ---------- |
| The program ``ping``. |
| |
| Examples |
| -------- |
| To ping 192.168.0.1 every 2 seconds for 10 seconds, and checks the successful |
| pings are >= 70% at the end of the test (The default values), add this in test |
| list:: |
| |
| { |
| "pytest_name": "ping_test", |
| "args": { |
| "host": "192.168.0.1" |
| } |
| } |
| |
| To ping 192.168.0.1 every 10 seconds for 120 seconds, and checks the successful |
| pings are >= 60% at the end of the test:: |
| |
| { |
| "pytest_name": "ping_test", |
| "args": { |
| "duration_secs": 120, |
| "host": "192.168.0.1", |
| "interval_secs": 10, |
| "ping_success_percent": 60 |
| } |
| } |
| |
| To ping 192.168.0.1 on interface eth0 every 10 seconds for 120 seconds, checks |
| the successful pings are >= 70% within the moving window of 5 pings, and also |
| checks the successful pings are >= 70% overall:: |
| |
| { |
| "pytest_name": "ping_test", |
| "args": { |
| "duration_secs": 120, |
| "host": "192.168.0.1", |
| "interface": "eth0", |
| "interval_secs": 10, |
| "moving_window_size": 5 |
| } |
| } |
| """ |
| |
| import logging |
| |
| from cros.factory.test import test_case |
| from cros.factory.test import test_ui |
| from cros.factory.utils.arg_utils import Arg |
| from cros.factory.utils import process_utils |
| from cros.factory.utils import time_utils |
| |
| |
| class PingTest(test_case.TestCase): |
| related_components = tuple() |
| ARGS = [ |
| Arg('host', str, 'The IP address or hostname to ping.'), |
| Arg('interface', str, 'Source interface address, may be numeric IP ' |
| 'address or name of network device, ex: eth0.', |
| default=None), |
| Arg('interval_secs', int, 'The interval in seconds between two ' |
| 'consecutive pings.', |
| default=2), |
| Arg('duration_secs', int, 'The duration of the ping test in seconds.', |
| default=10), |
| Arg('ping_success_percent', int, 'The percentage of successful pings ' |
| 'to meet. It will be checked at the end of the test.', |
| default=70), |
| Arg('moving_window_size', int, 'The size of the moving window in number ' |
| 'of ping attempts. If it is set, the percentage of successful ' |
| 'pings will be checked within the moving window in addition to ' |
| 'that at the end of the test.', |
| default=None), |
| Arg('verbose', bool, 'Dumps stdout of ping commands.', |
| default=False), |
| Arg('packet_size', int, 'Specifies the number of data bytes to be sent.', |
| default=None), |
| ] |
| |
| ui_class = test_ui.ScrollableLogUI |
| |
| def _CheckSuccessPercentage(self, success_count, total_count, title=''): |
| """Checks the percentage of successful pings is within the range.""" |
| success_percentage = (success_count / total_count) * 100 |
| # yapf: disable |
| if success_percentage < self.args.ping_success_percent: # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| self.FailTask( |
| # yapf: disable |
| f'Failed to meet ping success percentage: {success_percentage:.2f}% ' # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| f'(expected: {int(self.args.ping_success_percent)}%).') |
| logging.info('%s%.2f%% packets received.', title, success_percentage) |
| |
| def runTest(self): |
| """Tests the network connection by pinging a host for a period of time. |
| |
| Pings a host and counts the percentage of successful pings at the end of |
| the test. If moving_window_size is set, it will also check the successful |
| percentage within the moving window during the ping tests. |
| """ |
| # yapf: disable |
| window_size = self.args.moving_window_size # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| moving_queue = [] |
| moving_success_count = 0 |
| total_success_count = 0 |
| total_count = 0 |
| |
| # yapf: disable |
| ping_command = f'ping {self.args.host} -c 1' # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| # yapf: disable |
| if self.args.interface: # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| # yapf: disable |
| ping_command += f' -I {self.args.interface}' # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| # yapf: disable |
| if self.args.packet_size: # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| # yapf: disable |
| ping_command += f' -s {int(self.args.packet_size)}' # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| |
| # yapf: disable |
| end_time = time_utils.MonotonicTime() + self.args.duration_secs # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| while time_utils.MonotonicTime() < end_time: |
| # yapf: disable |
| if self.args.verbose: # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| p = process_utils.Spawn(ping_command, |
| shell=True, log=True, read_stdout=True) |
| logging.info(p.stdout_data) |
| if total_count % 10 == 0: |
| # yapf: disable |
| self.ui.ClearLog() # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| # yapf: disable |
| self.ui.AppendLog(p.stdout_data + '\n') # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| else: |
| p = process_utils.Spawn(ping_command, shell=True, call=True, |
| ignore_stdout=True, ignore_stderr=True) |
| result = int(p.returncode == 0) |
| |
| if window_size is not None: |
| moving_success_count += result |
| moving_queue.append(result) |
| if len(moving_queue) > window_size: |
| moving_success_count -= moving_queue.pop(0) |
| if len(moving_queue) == window_size: |
| self._CheckSuccessPercentage( |
| moving_success_count, window_size, 'Moving average: ') |
| |
| total_success_count += result |
| total_count += 1 |
| # yapf: disable |
| self.Sleep(self.args.interval_secs) # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| |
| self._CheckSuccessPercentage(total_success_count, total_count, 'Overall: ') |
