| #!/usr/bin/env python |
| # |
| # Copyright (C) 2015 The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| """Script to run `adb` tests for Brillo. |
| |
| This script provides an easy way to run adb tests against a Brillo |
| device. By default it will just run the functional test once and exit. |
| If a duration is specified, the script will also run a more data- |
| intensive file test for the given duration. |
| |
| Ctrl+C can be used to stop the file test at any time. |
| |
| Example usage: |
| # Functional tests only. |
| $ brillo_adb_test.py |
| |
| # Functional tests + file test 3 times. |
| $ brillo_adb_test.py --duration 3 |
| |
| # Functional tests + file test for 20 minutes. |
| $ brillo_adb_test.py --duration 20m |
| |
| # Only file test for 4.5 hours. |
| $ brillo_adb_test.py --duration 4.5h --skip-functional-tests |
| """ |
| |
| |
| from __future__ import print_function |
| |
| import argparse |
| import os |
| import subprocess |
| import sys |
| import time |
| import unittest |
| |
| from adb import test_brillo_device |
| |
| |
| def time_string(seconds): |
| """Takes seconds and returns a simple H:MM:SS time string""" |
| minutes, seconds = divmod(int(seconds), 60) |
| hours, minutes = divmod(minutes, 60) |
| return '{:01}:{:02}:{:02}'.format(hours, minutes, seconds) |
| |
| |
| class TestRepeater(object): |
| """Manages running a test multiple times""" |
| |
| def __init__(self, iterations, duration, skip_functional_tests): |
| """Constructor. |
| |
| Args: |
| iterations: how many times to run the test, or None to use a time |
| duration instead. |
| duration: how long to run the test in seconds, or None to use an |
| iteration count instead. |
| skip_functional_tests: True to skip the initial functional tests. |
| """ |
| self.start_time = None |
| self.iterations = None |
| self.duration = None |
| if iterations is not None: |
| self.iterations = int(iterations) |
| elif duration is not None: |
| self.duration = duration |
| else: |
| raise ValueError('Must specify test iterations or duration.') |
| self.iteration_start_time = self.start_time |
| self.successes = 0 |
| self.failures = 0 |
| self.skip_functional_tests = skip_functional_tests |
| |
| def _start_iteration(self): |
| """Returns True if we should start another test iteration.""" |
| if self.iterations is not None: |
| self.iterations -= 1 |
| return self.iterations >= 0 |
| else: |
| now = time.time() |
| if now >= self.start_time + self.duration: |
| return False |
| self.iteration_start_time = now |
| return True |
| |
| def _result_string(self): |
| """Returns the current results as a string. |
| |
| Returned string depends on whether we're counting iterations or |
| time, and looks like this: |
| [Fri Oct 16 16:59:30 2015] 3/3 | pass: 2 fail: 0 |
| or this: |
| [Fri Oct 16 16:59:56 2015] 0:00:03/0:00:05 | pass: 2 fail: 0 |
| """ |
| timestamp = time.strftime('%c') |
| if self.iterations is not None: |
| iter_count = self.successes + self.failures + 1 |
| progress = '{}/{}'.format(iter_count, iter_count + self.iterations) |
| else: |
| progress = '{}/{}'.format( |
| time_string(self.iteration_start_time - self.start_time), |
| time_string(self.duration)) |
| return '[{}] {} | pass: {} fail: {}'.format( |
| timestamp, progress, self.successes, self.failures) |
| |
| def run_test(self): |
| """Run the functional and transfer tests. |
| |
| Returns: |
| An exit code to return from the script; 0 if all tests passed, |
| 1 if one or more test failed. |
| """ |
| if not self.skip_functional_tests: |
| suite = test_brillo_device.suite() |
| # unittest verbosity levels: |
| # 0 = result line only when done. |
| # 1 = 1-character results as tests run. |
| # 2 = test descriptions and results as they run. |
| result = unittest.TextTestRunner(verbosity=2).run(suite) |
| if not result.wasSuccessful(): |
| return 1 |
| print() |
| |
| data_file_name = None |
| self.start_time = time.time() |
| try: |
| while self._start_iteration(): |
| if not data_file_name: |
| # Currently crashes are triggered when a lot of data is |
| # sent from the device to the host, which this will |
| # replicate. Later on we could add host->device transfers |
| # also if needed. |
| size = 20 * 1000 * 1000 |
| data_file_name = test_brillo_device.create_data_file(size) |
| print('Pushing file to device') |
| print('-------------------------') |
| try: |
| subprocess.check_call( |
| ['adb', 'push', data_file_name, |
| test_brillo_device.FileTest.DEVICE_TEMP_FILE]) |
| except subprocess.CalledProcessError as e: |
| print('"{}" failed, aborting test'.format( |
| ' '.join(e.cmd))) |
| break |
| print() |
| print('Starting file pull loop') |
| print('-------------------------') |
| |
| try: |
| print(self._result_string()) |
| subprocess.check_call( |
| ['adb', 'pull', |
| test_brillo_device.FileTest.DEVICE_TEMP_FILE, |
| data_file_name]) |
| self.successes += 1 |
| except subprocess.CalledProcessError as e: |
| self.failures += 1 |
| print('Failure: "{}" exited with code {}'.format( |
| e.cmd, e.returncode)) |
| print(e.output) |
| print() |
| |
| except KeyboardInterrupt: |
| print('\nStopping test due to keyboard interrupt') |
| |
| if data_file_name: |
| print('Test finished in {} with {}/{} success'.format( |
| time_string(time.time() - self.start_time), |
| self.successes, self.successes + self.failures)) |
| os.remove(data_file_name) |
| try: |
| subprocess.check_call( |
| ['adb', 'shell', 'rm', '-f', |
| test_brillo_device.FileTest.DEVICE_TEMP_FILE]) |
| except subprocess.CalledProcessError: |
| # If previous failures happened this may not work which is fine. |
| pass |
| |
| if self.failures > 0: |
| return 1 |
| else: |
| return 0 |
| |
| |
| def parse_time_argument(arg, default_unit=''): |
| """Parses user input to get a time value. |
| |
| Args: |
| arg: user input string. |
| default_unit: unit to assume if unspecified. |
| |
| Returns: |
| A (value, seconds) tuple. Value will be set if no unit was |
| specified, otherwise seconds will be set. |
| |
| Raises: |
| ValueError: invalid |arg| value. |
| """ |
| # We can be a little loose with parsing, assume anything starting |
| # with 's' is seconds, 'm' for minutes, etc. |
| arg += default_unit |
| for char, multiplier in (('s', 1), ('m', 60), ('h', 3600), ('d', 86400)): |
| value, sep, _ = arg.partition(char) |
| if sep: |
| return (None, float(value) * multiplier) |
| return (float(arg), None) |
| |
| |
| def parse_arguments(): |
| """Parses the command-line arguments. |
| |
| Returns: |
| A TestRepeater object. |
| """ |
| parser = argparse.ArgumentParser( |
| description=__doc__, |
| formatter_class=argparse.RawDescriptionHelpFormatter) |
| parser.add_argument('-d', '--duration', default='0', |
| type=parse_time_argument, |
| help='How long to run the test. <n> = iterations,' |
| ' <n>m = minutes, <n>h = hours.') |
| parser.add_argument('-s', '--skip-functional-tests', action='store_true', |
| help='Skip the initial functional tests.') |
| options = parser.parse_args() |
| |
| return TestRepeater(options.duration[0], options.duration[1], |
| options.skip_functional_tests) |
| |
| |
| def main(): |
| test_repeater = parse_arguments() |
| return test_repeater.run_test() |
| |
| |
| if __name__ == '__main__': |
| sys.exit(main()) |
