blob: 164bb402f1a9f011a9736ab3321db116bf0132a9 [file] [log] [blame]
#!/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())