crostestutils: Fix lint errors in files under au_test_harness and ctest
The result is `cros lint au_test_harness/*.py` and `cros lint ctest/*.py` don't
complain. There are over 900 lint errors in files under other directories, but I
don't forsee myself touching any of them in the near future, so I am leaving
them alone...
BUG=None
TEST=`cros lint`, manual ctest run, and trybot
Change-Id: I0ac117a442d652bafe9a11488090181a945e2697
Reviewed-on: https://chromium-review.googlesource.com/325393
Commit-Ready: Daniel Wang <wonderfly@google.com>
Tested-by: Daniel Wang <wonderfly@google.com>
Reviewed-by: Mike Frysinger <vapier@chromium.org>
diff --git a/au_test_harness/au_test.py b/au_test_harness/au_test.py
index a4c5758..c0f0de1 100644
--- a/au_test_harness/au_test.py
+++ b/au_test_harness/au_test.py
@@ -4,6 +4,8 @@
"""Module containing a test suite that is run to test auto updates."""
+from __future__ import print_function
+
import os
import re
import time
@@ -71,7 +73,7 @@
port_in=proxy_port,
address_out='127.0.0.1',
port_out=dev_server_wrapper.DEFAULT_PORT,
- filter=update_filter
+ data_filter=update_filter
)
proxy.serve_forever_in_thread()
try:
@@ -149,6 +151,7 @@
def __init__(self):
"""Defines variable shared across all connections."""
self.close_count = 0
+ self.data_size = 0
def setup(self):
"""Called once at the start of each connection."""
@@ -212,6 +215,10 @@
It does this by inserting 3 20 second delays when transmitting
data after 2M has been sent.
"""
+ def __init__(self):
+ """Defines variable shared across all connections."""
+ self.data_size = 0
+ self.delay_count = 0
def setup(self):
"""Called once at the start of each connection."""
diff --git a/au_test_harness/au_worker.py b/au_test_harness/au_worker.py
index 88ec805..4b3417a 100644
--- a/au_test_harness/au_worker.py
+++ b/au_test_harness/au_worker.py
@@ -9,13 +9,14 @@
various types of target. Types of targets include VM's, real devices, etc.
"""
+from __future__ import print_function
+
import inspect
import os
from chromite.lib import cros_build_lib
from chromite.lib import cros_logging as logging
from chromite.lib import dev_server_wrapper
-from chromite.lib import path_util
from crostestutils.au_test_harness import update_exception
@@ -41,6 +42,16 @@
self.verify_suite = 'suite:%s' % (options.verify_suite_name or 'smoke')
self.ssh_private_key = options.ssh_private_key
+ # Attributes to be initialized in Initialize.
+ self._ssh_port = None
+ self._kvm_pid_file = None
+
+ # Initialize test results directory.
+ self.test_name = None
+ self.all_results_directory = None
+ self.fail_results_directory = None
+ self.results_count = 0
+
def CleanUp(self):
"""Called at the end of every test."""
@@ -52,8 +63,10 @@
else:
msg = 'Performing a full update to %s' % update_target
- if from_vm: msg += ' in a VM'
- if proxy: msg += ' using a proxy on port ' + str(proxy)
+ if from_vm:
+ msg += ' in a VM'
+ if proxy:
+ msg += ' using a proxy on port ' + str(proxy)
return msg
def PrepareBase(self, image_path, signed_base):
@@ -74,10 +87,7 @@
"""Implementation of an actual update.
Subclasses must override this method with the correct update procedure for
- the class.
-
- Args:
- See PerformUpdate for description of args.
+ the class. See PerformUpdate for args.
"""
def UpdateUsingPayload(self, update_path, stateful_change='old',
@@ -88,11 +98,11 @@
the class.
Args:
- update_path: Path to the image to update with. This directory should
- contain both update.gz, and stateful.image.gz
+ update_path: Path to the image to update with. This directory should
+ contain both update.gz, and stateful.image.gz
stateful_change: How to perform the stateful update.
- proxy_port: Port to have the client connect to. For use with
- CrosTestProxy.
+ proxy_port: Port to have the client connect to. For use with
+ CrosTestProxy.
"""
def VerifyImage(self, unittest, percent_required_to_pass=100, test=''):
@@ -102,11 +112,11 @@
override this method with the correct update procedure for the class.
Args:
- unittest: pointer to a unittest to fail if we cannot verify the image.
- percent_required_to_pass: percentage required to pass. This should be
- fall between 0-100.
- test: test that will be used to verify the image. If omitted or equal to
- the empty string the code will use self.verify_suite.
+ unittest: Pointer to a unittest to fail if we cannot verify the image.
+ percent_required_to_pass: Percentage required to pass. This should be
+ fall between 0-100.
+ test: Test that will be used to verify the image. If omitted or equal to
+ the empty string the code will use self.verify_suite.
Returns:
Returns the percent that passed.
@@ -122,20 +132,21 @@
instead.
Args:
- image_path: Path to the image to update with. This image must be a test
- image.
- src_image_path: Optional. If set, perform a delta update using the
- image specified by the path as the source image.
+ image_path: Path to the image to update with. This image must be a test
+ image.
+ src_image_path: Optional. If set, perform a delta update using the
+ image specified by the path as the source image.
stateful_change: How to modify the stateful partition. Values are:
- 'old': Don't modify stateful partition. Just update normally.
- 'clean': Uses clobber-state to wipe the stateful partition with the
- exception of code needed for ssh.
- proxy_port: Port to have the client connect to. For use with
- CrosTestProxy.
+ 'old': Don't modify stateful partition. Just update normally.
+ 'clean': Uses clobber-state to wipe the stateful partition with the
+ exception of code needed for ssh.
+ proxy_port: Port to have the client connect to. For use with
+ CrosTestProxy.
payload_signing_key: Path to the private key to use to sign payloads.
Raises an update_exception.UpdateException if _UpdateImage returns an error.
"""
- if not self.use_delta_updates: src_image_path = ''
+ if not self.use_delta_updates:
+ src_image_path = ''
key_to_use = payload_signing_key
self.UpdateImage(image_path, src_image_path, stateful_change, proxy_port,
@@ -184,11 +195,16 @@
Modifies cmd in places by appending appropriate items given args.
Args:
- See PerformUpdate for description of args.
+ cmd: See PerformUpdate.
+ image_path: See PerformUpdate.
+ src_image_path: See PerformUpdate.
+ proxy_port: See PerformUpdate.
+ payload_signing_key: See PerformUpdate.
for_vm: Additional optional argument to say that the payload is intended
- for vm usage (so we don't patch the kernel).
+ for vm usage (so we don't patch the kernel).
"""
- if proxy_port: cmd.append('--proxy_port=%s' % proxy_port)
+ if proxy_port:
+ cmd.append('--proxy_port=%s' % proxy_port)
update_id = dev_server_wrapper.GenerateUpdateId(
image_path, src_image_path, payload_signing_key,
for_vm=for_vm)
@@ -204,11 +220,12 @@
def RunUpdateCmd(self, cmd, log_directory=None):
"""Runs the given update cmd given verbose options.
- Raises an update_exception.UpdateException if the update fails.
-
Args:
- cmd: The shell cmd to run.
- log_directory: Where to store the logs for this cmd.
+ cmd: The shell cmd to run.
+ log_directory: Where to store the logs for this cmd.
+
+ Raises:
+ update_exception.UpdateException: If the update fails.
"""
kwds = dict(print_cmd=False, combine_stdout_stderr=True, error_code_ok=True)
if not self.verbose:
@@ -227,7 +244,8 @@
unittest: the unittest object running this test.
output: stdout from a test run.
percent_required_to_pass: percentage required to pass. This should be
- fall between 0-100.
+ fall between 0-100.
+
Returns:
percent that passed.
"""
@@ -235,7 +253,7 @@
self.TestInfo('Percent passed: %d vs. Percent required: %d' % (
percent_passed, percent_required_to_pass))
if percent_passed < percent_required_to_pass:
- print output
+ print(output)
unittest.fail('%d percent of tests are required to pass' %
percent_required_to_pass)
@@ -250,7 +268,7 @@
Each test needs to specify a unique ssh port.
Args:
- port: Unique port for ssh access.
+ port: Unique port for ssh access.
"""
# Initialize port vars.
self._ssh_port = port
@@ -277,6 +295,7 @@
Args:
label: The label used to describe this test phase.
+
Returns:
Returns a path for the results directory to use for this label.
"""
@@ -295,6 +314,7 @@
Args:
output: Output string for generate_test_report.py.
+
Returns:
The percentage of tests that passed.
"""
diff --git a/au_test_harness/cros_au_test_harness.py b/au_test_harness/cros_au_test_harness.py
index 8f5c388..be6bd07 100755
--- a/au_test_harness/cros_au_test_harness.py
+++ b/au_test_harness/cros_au_test_harness.py
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python2
# Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
@@ -12,8 +12,10 @@
being run during the update process.
"""
+from __future__ import print_function
+
+import argparse
import functools
-import optparse
import os
import pickle
import sys
@@ -91,17 +93,17 @@
return None
-def _PrepareTestSuite(options):
- """Returns a prepared test suite given by the options and test class."""
- au_test.AUTest.ProcessOptions(options)
+def _PrepareTestSuite(opts):
+ """Returns a prepared test suite given by the opts and test class."""
+ au_test.AUTest.ProcessOptions(opts)
test_loader = unittest.TestLoader()
- test_loader.testMethodPrefix = options.test_prefix
+ test_loader.testMethodPrefix = opts.test_prefix
return test_loader.loadTestsFromTestCase(au_test.AUTest)
-def _RunTestsInParallel(options):
- """Runs the tests given by the options in parallel."""
- test_suite = _PrepareTestSuite(options)
+def _RunTestsInParallel(opts):
+ """Runs the tests given by the opts in parallel."""
+ test_suite = _PrepareTestSuite(opts)
steps = []
for test in test_suite:
test_name = test.id()
@@ -110,113 +112,113 @@
logging.info('Running tests in test suite in parallel.')
try:
- parallel.RunParallelSteps(steps, max_parallel=options.jobs)
+ parallel.RunParallelSteps(steps, max_parallel=opts.jobs)
except parallel.BackgroundFailure as ex:
cros_build_lib.Die(ex)
-def CheckOptions(parser, options, leftover_args):
- """Assert given options are valid.
+def CheckOpts(parser, opts):
+ """Assert given opts are valid.
Args:
- parser: Parser used to parse options.
- options: Parsed options.
- leftover_args: Args left after parsing.
+ parser: Parser used to parse opts.
+ opts: Parsed opts.
"""
- if leftover_args: parser.error('Found unsupported flags ' + leftover_args)
- if not options.type in ['real', 'vm', 'gce']:
+ if not opts.type in ['real', 'vm', 'gce']:
parser.error('Failed to specify valid test type.')
def _IsValidImage(image):
- """Asserts that |image_path| is a valid image file for |options.type|."""
+ """Asserts that |image_path| is a valid image file for |opts.type|."""
return (image is not None) and os.path.isfile(image)
- if not _IsValidImage(options.target_image):
+ if not _IsValidImage(opts.target_image):
parser.error('Testing requires a valid target image.\n'
'Given: type=%s, target_image=%s.' %
- (options.type, options.target_image))
+ (opts.type, opts.target_image))
- if not options.base_image:
+ if not opts.base_image:
logging.info('No base image supplied. Using target as base image.')
- options.base_image = options.target_image
+ opts.base_image = opts.target_image
- if not _IsValidImage(options.base_image):
+ if not _IsValidImage(opts.base_image):
parser.error('Testing requires a valid base image.\n'
'Given: type=%s, base_image=%s.' %
- (options.type, options.base_image))
+ (opts.type, opts.base_image))
- if (options.payload_signing_key and not
- os.path.isfile(options.payload_signing_key)):
+ if (opts.payload_signing_key and not
+ os.path.isfile(opts.payload_signing_key)):
parser.error('Testing requires a valid path to the private key.')
- if options.ssh_private_key and not os.path.isfile(options.ssh_private_key):
+ if opts.ssh_private_key and not os.path.isfile(opts.ssh_private_key):
parser.error('Testing requires a valid path to the ssh private key.')
- if options.test_results_root:
- if not 'chroot/tmp' in options.test_results_root:
+ if opts.test_results_root:
+ if not 'chroot/tmp' in opts.test_results_root:
parser.error('Must specify a test results root inside tmp in a chroot.')
- if not os.path.exists(options.test_results_root):
- os.makedirs(options.test_results_root)
+ if not os.path.exists(opts.test_results_root):
+ os.makedirs(opts.test_results_root)
else:
chroot_tmp = os.path.join(constants.SOURCE_ROOT, 'chroot', 'tmp')
- options.test_results_root = tempfile.mkdtemp(
+ opts.test_results_root = tempfile.mkdtemp(
prefix='au_test_harness', dir=chroot_tmp)
def main():
test_helper.SetupCommonLoggingFormat()
- parser = optparse.OptionParser()
- parser.add_option('-b', '--base_image',
- help='path to the base image.')
- parser.add_option('-r', '--board',
- help='board for the images.')
- parser.add_option('--no_delta', action='store_false', default=True,
- dest='delta',
- help='Disable using delta updates.')
- parser.add_option('--no_graphics', action='store_true',
- help='Disable graphics for the vm test.')
- parser.add_option('-j', '--jobs', default=test_helper.CalculateDefaultJobs(),
- type=int, help='Number of simultaneous jobs')
- parser.add_option('--payload_signing_key', default=None,
- help='Path to the private key used to sign payloads with.')
- parser.add_option('-q', '--quick_test', default=False, action='store_true',
- help='Use a basic test to verify image.')
- parser.add_option('-m', '--remote',
- help='Remote address for real test.')
- parser.add_option('-t', '--target_image',
- help='path to the target image.')
- parser.add_option('--test_results_root', default=None,
- help='Root directory to store test results. Should '
- 'be defined relative to chroot root.')
- parser.add_option('--test_prefix', default='test',
- help='Only runs tests with specific prefix i.e. '
- 'testFullUpdateWipeStateful.')
- parser.add_option('-p', '--type', default='vm',
- help='type of test to run: [vm, real, gce]. Default: vm.')
- parser.add_option('--verbose', default=True, action='store_true',
- help='Print out rather than capture output as much as '
- 'possible.')
- parser.add_option('--whitelist_chrome_crashes', default=False,
- dest='whitelist_chrome_crashes', action='store_true',
- help='Treat Chrome crashes as non-fatal.')
- parser.add_option('--verify_suite_name', default=None,
- help='Specify the verify suite to run.')
- parser.add_option('--parallel', default=False, dest='parallel',
- action='store_true',
- help='Run multiple test stages in parallel (applies only '
- 'to vm tests). Default: False')
- parser.add_option('--ssh_private_key', default=None,
- help='Path to the private key to use to ssh into the image '
- 'as the root user.')
- (options, leftover_args) = parser.parse_args()
+ parser = argparse.ArgumentParser()
+ parser.add_argument('-b', '--base_image',
+ help='path to the base image.')
+ parser.add_argument('-r', '--board',
+ help='board for the images.')
+ parser.add_argument('--no_delta', action='store_false', default=True,
+ dest='delta',
+ help='Disable using delta updates.')
+ parser.add_argument('--no_graphics', action='store_true',
+ help='Disable graphics for the vm test.')
+ parser.add_argument('-j', '--jobs',
+ default=test_helper.CalculateDefaultJobs(), type=int,
+ help='Number of simultaneous jobs')
+ parser.add_argument('--payload_signing_key', default=None,
+ help='Path to the private key used to sign payloads '
+ 'with.')
+ parser.add_argument('-q', '--quick_test', default=False, action='store_true',
+ help='Use a basic test to verify image.')
+ parser.add_argument('-m', '--remote',
+ help='Remote address for real test.')
+ parser.add_argument('-t', '--target_image',
+ help='path to the target image.')
+ parser.add_argument('--test_results_root', default=None,
+ help='Root directory to store test results. Should '
+ 'be defined relative to chroot root.')
+ parser.add_argument('--test_prefix', default='test',
+ help='Only runs tests with specific prefix i.e. '
+ 'testFullUpdateWipeStateful.')
+ parser.add_argument('-p', '--type', default='vm',
+ help='type of test to run: [vm, real, gce]. Default: vm.')
+ parser.add_argument('--verbose', default=True, action='store_true',
+ help='Print out rather than capture output as much as '
+ 'possible.')
+ parser.add_argument('--whitelist_chrome_crashes', default=False,
+ dest='whitelist_chrome_crashes', action='store_true',
+ help='Treat Chrome crashes as non-fatal.')
+ parser.add_argument('--verify_suite_name', default=None,
+ help='Specify the verify suite to run.')
+ parser.add_argument('--parallel', default=False, dest='parallel',
+ action='store_true',
+ help='Run multiple test stages in parallel (applies only '
+ 'to vm tests). Default: False')
+ parser.add_argument('--ssh_private_key', default=None,
+ help='Path to the private key to use to ssh into the '
+ 'image as the root user.')
+ opts = parser.parse_args()
- CheckOptions(parser, options, leftover_args)
+ CheckOpts(parser, opts)
# Generate cache of updates to use during test harness.
- update_cache = _ReadUpdateCache(options.type, options.target_image)
+ update_cache = _ReadUpdateCache(opts.type, opts.target_image)
if not update_cache:
msg = ('No update cache found. Update testing will not work. Run '
' cros_generate_update_payloads if this was not intended.')
@@ -236,15 +238,15 @@
if update_cache:
my_server = dev_server_wrapper.DevServerWrapper(
port=dev_server_wrapper.DEFAULT_PORT,
- log_dir=options.test_results_root)
+ log_dir=opts.test_results_root)
my_server.Start()
- if options.type == 'vm' or options.type == 'gce' and options.parallel:
- _RunTestsInParallel(options)
+ if opts.type == 'vm' or opts.type == 'gce' and opts.parallel:
+ _RunTestsInParallel(opts)
else:
# TODO(sosa) - Take in a machine pool for a real test.
# Can't run in parallel with only one remote device.
- test_suite = _PrepareTestSuite(options)
+ test_suite = _PrepareTestSuite(opts)
test_result = unittest.TextTestRunner().run(test_suite)
if not test_result.wasSuccessful():
cros_build_lib.Die('Test harness failed.')
diff --git a/au_test_harness/cros_test_proxy.py b/au_test_harness/cros_test_proxy.py
index 22c9dd9..9a105f6 100644
--- a/au_test_harness/cros_test_proxy.py
+++ b/au_test_harness/cros_test_proxy.py
@@ -4,6 +4,8 @@
"""Module containing classes pertaining to inserting a proxy in a test."""
+from __future__ import print_function
+
import os
import select
import socket
@@ -46,14 +48,14 @@
def setup(self):
"""Setup is called once for each connection proxied."""
- self.server.filter.setup()
+ self.server.data_filter.setup()
def handle(self):
"""Handles each incoming connection.
Opens a new connection to the port we are proxing to, then
passes each packet along in both directions after passing
- them through the filter object passed in.
+ them through the Filter object passed in.
"""
# Open outgoing socket
s_in = self.request
@@ -61,12 +63,13 @@
s_out.connect((self.server.address_out, self.server.port_out))
while True:
- rlist, wlist, xlist = select.select([s_in, s_out], [], [])
+ rlist, _, _ = select.select([s_in, s_out], [], [])
if s_in in rlist:
data = s_in.recv(1024)
- data = self.server.filter.InBound(data)
- if not data: break
+ data = self.server.data_filter.InBound(data)
+ if not data:
+ break
try:
# If there is any error sending data, close both connections.
s_out.sendall(data)
@@ -75,8 +78,9 @@
if s_out in rlist:
data = s_out.recv(1024)
- data = self.server.filter.OutBound(data)
- if not data: break
+ data = self.server.data_filter.OutBound(data)
+ if not data:
+ break
try:
# If there is any error sending data, close both connections.
s_in.sendall(data)
@@ -87,21 +91,21 @@
s_out.close()
def __init__(self,
- filter,
+ data_filter,
port_in=8081,
address_out='127.0.0.1', port_out=8080):
"""Configures the proxy object.
Args:
- filter: An instance of a subclass of Filter.
+ data_filter: An instance of a subclass of Filter.
port_in: Port on which to listen for incoming connections.
address_out: Address to which outgoing connections will go.
- address_port: Port to which outgoing connections will go.
+ port_out: Port to which outgoing connections will go.
"""
self.port_in = port_in
self.address_out = address_out
self.port_out = port_out
- self.filter = filter
+ self.data_filter = data_filter
# Used to coordinate startup/shutdown in a new thread.
self.__is_started = threading.Event()
@@ -148,7 +152,7 @@
# connecting to the socket to wake this up instead of
# polling. Polling reduces our responsiveness to a
# shutdown request and wastes cpu at all other times.
- r, w, e = select.select([self], [], [], poll_interval)
+ r, _, _ = select.select([self], [], [], poll_interval)
if r:
self._handle_request_noblock()
@@ -167,4 +171,3 @@
"""
self.__serving = False
self.__is_shut_down.wait()
-
diff --git a/au_test_harness/gce_au_worker_unittest.py b/au_test_harness/gce_au_worker_unittest.py
index e88de00..20ba25b 100755
--- a/au_test_harness/gce_au_worker_unittest.py
+++ b/au_test_harness/gce_au_worker_unittest.py
@@ -306,15 +306,16 @@
return_value=(0, None, None))
# Fake resource existance.
- remote_instance = 'fake-remote-instance'
- self.PatchObject(worker.gce_context, 'InstanceExists', autospec=True,
- side_effect=lambda instance: remote_instance is not None)
+ remote_instances = ('fake-remote-instance',)
+ self.PatchObject(
+ worker.gce_context, 'InstanceExists', autospec=True,
+ side_effect=lambda instance: instance in remote_instances)
# Fake resource deletors.
self.PatchObject(worker.gce_context, 'DeleteImage', autospec=True)
# Make DeleteInstance delete the remote resource.
- def _OverrideDeleteInstance(_):
- remote_instance = None
+ def _OverrideDeleteInstance(instance):
+ remote_instances.remove(instance)
self.PatchObject(worker.gce_context, 'DeleteInstance', autospec=True,
side_effect=_OverrideDeleteInstance)
diff --git a/au_test_harness/real_au_worker.py b/au_test_harness/real_au_worker.py
index 6da9ab9..b6fa0a2 100644
--- a/au_test_harness/real_au_worker.py
+++ b/au_test_harness/real_au_worker.py
@@ -4,6 +4,8 @@
"""Module containing class that implements an au_worker for a test device."""
+from __future__ import print_function
+
import constants
from chromite.lib import cros_build_lib
from chromite.lib import path_util
@@ -47,13 +49,15 @@
stateful_change_flag,
'--verify',
]
- if proxy_port: cmd.append('--proxy_port=%s' % proxy_port)
+ if proxy_port:
+ cmd.append('--proxy_port=%s' % proxy_port)
self.RunUpdateCmd(cmd)
def VerifyImage(self, unittest, percent_required_to_pass=100, test=''):
"""Verifies an image using test_that with verification suite."""
test_directory, _ = self.GetNextResultsPath('autotest_tests')
- if not test: test = self.verify_suite
+ if not test:
+ test = self.verify_suite
cmd = ['test_that', '--no-quickmerge', '--results_dir=%s' % test_directory,
self.remote, test]
@@ -67,4 +71,3 @@
return self.AssertEnoughTestsPassed(unittest, result.output,
percent_required_to_pass)
-
diff --git a/au_test_harness/vm_au_worker.py b/au_test_harness/vm_au_worker.py
index 4bfa724..7471978 100644
--- a/au_test_harness/vm_au_worker.py
+++ b/au_test_harness/vm_au_worker.py
@@ -4,6 +4,8 @@
"""Module containing implementation of an au_worker for virtual machines."""
+from __future__ import print_function
+
import os
import shutil
import tempfile
@@ -23,7 +25,8 @@
"""Processes vm-specific options."""
super(VMAUWorker, self).__init__(options, test_results_root)
self.graphics_flag = ''
- if options.no_graphics: self.graphics_flag = '--no_graphics'
+ if options.no_graphics:
+ self.graphics_flag = '--no_graphics'
if not self.board:
cros_build_lib.Die('Need board to convert base image to vm.')
self.whitelist_chrome_crashes = options.whitelist_chrome_crashes
@@ -125,7 +128,8 @@
'--ssh_port=%s' % self._ssh_port,
stateful_change_flag,
]
- if proxy_port: cmd.append('--proxy_port=%s' % proxy_port)
+ if proxy_port:
+ cmd.append('--proxy_port=%s' % proxy_port)
self.TestInfo(self.GetUpdateMessage(update_path, None, True, proxy_port))
try:
self.RunUpdateCmd(cmd, log_directory)
@@ -137,10 +141,8 @@
payload_signing_key, for_vm=False):
"""Appends common args to an update cmd defined by an array.
- Calls super function with for_vm set to True.
-
- Args:
- See AppendUpdateFlags for description of args.
+ Calls super function with for_vm set to True. See base class for
+ descriptions for arguments.
"""
super(VMAUWorker, self).AppendUpdateFlags(
cmd, image_path, src_image_path, proxy_port, payload_signing_key,
@@ -150,7 +152,7 @@
# VMAUWorker disregards |unittest| and |percent_required_to_pass|.
return self._VerifyImage(test)
- # pylint: disable-msg=W0221
+ # pylint: disable=W0221
def _VerifyImage(self, test=''):
"""Runs vm smoke suite or any single test to verify image.
@@ -160,7 +162,8 @@
(_, _, log_directory_in_chroot) = log_directory.rpartition('chroot')
# image_to_live already verifies lsb-release matching. This is just
# for additional steps.
- if not test: test = self.verify_suite
+ if not test:
+ test = self.verify_suite
command = ['./bin/cros_run_vm_test',
'--board=%s' % self.board,
@@ -172,7 +175,8 @@
'--verbose=0',
test,
]
- if self.graphics_flag: command.append(self.graphics_flag)
+ if self.graphics_flag:
+ command.append(self.graphics_flag)
if self.whitelist_chrome_crashes:
command.append('--whitelist_chrome_crashes')
if self.ssh_private_key is not None:
@@ -186,7 +190,7 @@
# If the command failed or printed warnings, print the output.
if result.returncode != 0 or '@@@STEP_WARNINGS@@@' in result.output:
- print result.output
+ print(result.output)
self._HandleFail(log_directory, fail_directory)
return result.returncode == 0
diff --git a/ctest/ctest.py b/ctest/ctest.py
index 940077f..7cc0675 100755
--- a/ctest/ctest.py
+++ b/ctest/ctest.py
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python2
#
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
@@ -6,8 +6,9 @@
"""Wrapper for tests that are run on builders."""
-import logging
-import optparse
+from __future__ import print_function
+
+import argparse
import os
import sys
@@ -17,6 +18,7 @@
sys.path.append(constants.CROS_PLATFORM_ROOT)
from chromite.lib import cros_build_lib
+from chromite.lib import cros_logging as logging
from crostestutils.lib import image_extractor
from crostestutils.lib import test_helper
@@ -46,25 +48,25 @@
whitelist_chrome_crashes: Whether to treat Chrome crashes as non-fatal.
"""
- def __init__(self, options):
+ def __init__(self, opts):
"""Initializes the test object.
Args:
- options: Parsed options for module.
+ opts: Parsed args for module.
"""
self.base = None
- self.board = options.board
- self.archive_dir = options.archive_dir
+ self.board = opts.board
+ self.archive_dir = opts.archive_dir
self.crosutils_root = os.path.join(constants.SOURCE_ROOT, 'src', 'scripts')
- self.no_graphics = options.no_graphics
- self.remote = options.remote
+ self.no_graphics = opts.no_graphics
+ self.remote = opts.remote
# TODO(sosa): Remove once signed payload bug is resolved.
- #self.sign_payloads = not options.cache
+ #self.sign_payloads = not opts.cache
self.sign_payloads = False
- self.target = options.target_image
- self.test_results_root = options.test_results_root
- self.type = options.type
- self.whitelist_chrome_crashes = options.whitelist_chrome_crashes
+ self.target = opts.target_image
+ self.test_results_root = opts.test_results_root
+ self.type = opts.type
+ self.whitelist_chrome_crashes = opts.whitelist_chrome_crashes
self.public_key = None
if self.sign_payloads:
@@ -74,11 +76,11 @@
else:
self.payload_signing_key = None
- self.jobs = options.jobs
- self.nplus1_archive_dir = options.nplus1_archive_dir
+ self.jobs = opts.jobs
+ self.nplus1_archive_dir = opts.nplus1_archive_dir
# An optional ssh private key used for testing.
- self.ssh_private_key = options.ssh_private_key
+ self.ssh_private_key = opts.ssh_private_key
def GeneratePublicKey(self):
"""Returns the path to a generated public key from the UE private key."""
@@ -143,7 +145,8 @@
else:
cmd.append('--basic_suite')
- if self.type != 'vm': cmd.append('--novm')
+ if self.type != 'vm':
+ cmd.append('--novm')
try:
cros_build_lib.RunCommand(cmd, cwd=self.crosutils_root)
except cros_build_lib.RunCommandError:
@@ -164,6 +167,8 @@
Args:
only_verify: Only verify the target image.
quick_update: Do a quick update test.
+ suite: The suite of tests to run.
+
Raises:
TestException: If the cros_au_test_harness command returns an error code.
"""
@@ -189,10 +194,12 @@
elif quick_update:
cmd.append('--test_prefix=SimpleTestUpdateAndVerify')
- if self.test_results_root: cmd.append('--test_results_root=%s' %
- self.test_results_root)
- if self.no_graphics: cmd.append('--no_graphics')
- if self.whitelist_chrome_crashes: cmd.append('--whitelist_chrome_crashes')
+ if self.test_results_root:
+ cmd.append('--test_results_root=%s' % self.test_results_root)
+ if self.no_graphics:
+ cmd.append('--no_graphics')
+ if self.whitelist_chrome_crashes:
+ cmd.append('--whitelist_chrome_crashes')
# We did not generate signed payloads if this is a |quick_update| test.
if not quick_update and self.sign_payloads:
@@ -208,78 +215,74 @@
def main():
test_helper.SetupCommonLoggingFormat()
- parser = optparse.OptionParser()
- parser.add_option('-b', '--board',
- help='board for the image to compare against.')
- parser.add_option('--archive_dir',
- help='Directory containing previously archived images.')
- parser.add_option('--cache', default=False, action='store_true',
- help='Cache payloads')
- parser.add_option('--jobs', default=test_helper.CalculateDefaultJobs(),
- type=int,
- help='Number of threads to run in parallel.')
- parser.add_option('--no_graphics', action='store_true', default=False,
- help='Disable graphics for the vm test.')
- parser.add_option('--only_verify', action='store_true', default=False,
- help='Only run basic verification suite.')
- parser.add_option('--quick_update', action='store_true',
- help='Run a quick update test. This will run a subset of '
- 'test suite after running autoupdate from target '
- 'image to itself.')
- parser.add_option('--nplus1_archive_dir', default=None,
- help='If set, directory to archive nplus1 payloads.')
- parser.add_option('--remote', default='0.0.0.0',
- help='For real tests, ip address of the target machine.')
- parser.add_option('--target_image', default=None,
- help='Target image to test.')
- parser.add_option('--suite', default=None, help='Test suite to run.')
- parser.add_option('--test_results_root', default=None,
- help='Root directory to store test results. Should '
- 'be defined relative to chroot root.')
- parser.add_option('--type', default='vm',
- help='type of test to run: [vm, real, gce]. Default: vm.')
- parser.add_option('--verbose', default=False, action='store_true',
- help='Print out added debugging information')
- parser.add_option('--whitelist_chrome_crashes', default=False,
- dest='whitelist_chrome_crashes', action='store_true',
- help='Treat Chrome crashes as non-fatal.')
- parser.add_option('--ssh_private_key', default=None,
- help='Path to the private key to use to ssh into the image '
- 'as the root user')
+ parser = argparse.ArgumentParser()
+ parser.add_argument('-b', '--board',
+ help='board for the image to compare against.')
+ parser.add_argument('--archive_dir',
+ help='Directory containing previously archived images.')
+ parser.add_argument('--cache', default=False, action='store_true',
+ help='Cache payloads')
+ parser.add_argument('--jobs', default=test_helper.CalculateDefaultJobs(),
+ type=int,
+ help='Number of threads to run in parallel.')
+ parser.add_argument('--no_graphics', action='store_true', default=False,
+ help='Disable graphics for the vm test.')
+ parser.add_argument('--only_verify', action='store_true', default=False,
+ help='Only run basic verification suite.')
+ parser.add_argument('--quick_update', action='store_true',
+ help='Run a quick update test. This will run a subset of '
+ 'test suite after running autoupdate from target '
+ 'image to itself.')
+ parser.add_argument('--nplus1_archive_dir', default=None,
+ help='If set, directory to archive nplus1 payloads.')
+ parser.add_argument('--remote', default='0.0.0.0',
+ help='For real tests, ip address of the target machine.')
+ parser.add_argument('--target_image', default=None,
+ help='Target image to test.')
+ parser.add_argument('--suite', default=None, help='Test suite to run.')
+ parser.add_argument('--test_results_root', default=None,
+ help='Root directory to store test results. Should '
+ 'be defined relative to chroot root.')
+ parser.add_argument('--type', default='vm',
+ help='type of test to run: [vm, real, gce]. Default: vm.')
+ parser.add_argument('--verbose', default=False, action='store_true',
+ help='Print out added debugging information')
+ parser.add_argument('--whitelist_chrome_crashes', default=False,
+ dest='whitelist_chrome_crashes', action='store_true',
+ help='Treat Chrome crashes as non-fatal.')
+ parser.add_argument('--ssh_private_key', default=None,
+ help='Path to the private key to use to ssh into the '
+ 'image as the root user')
- # Set the usage to include flags.
- def _ParserError(msg):
- print >> sys.stderr, parser.format_help()
- print >> sys.stderr, 'Error: %s' % msg
- sys.exit(2)
- parser.error = _ParserError
- (options, args) = parser.parse_args()
+ opts = parser.parse_args()
- if args: parser.error('Extra args found %s.' % args)
- if not options.board: parser.error('Need board for image to compare against.')
- if options.only_verify and options.quick_update: parser.error(
- 'Only one of --only_verify or --quick_update should be specified.')
+ if not opts.board:
+ parser.error('Need board for image to compare against.')
+ if opts.only_verify and opts.quick_update:
+ parser.error(
+ 'Only one of --only_verify or --quick_update should be specified.')
- # force absolute path for these options, since a chdir occurs deeper in the
+ # force absolute path for these opts, since a chdir occurs deeper in the
# codebase.
for x in ('nplus1_archive_dir', 'target_image', 'test_results_root'):
- if x == 'target_image' and options.type == 'gce':
+ if x == 'target_image' and opts.type == 'gce':
# In this case |target_image| is a Google Storage path.
continue
- val = getattr(options, x)
+ val = getattr(opts, x)
if val is not None:
- setattr(options, x, os.path.abspath(val))
+ setattr(opts, x, os.path.abspath(val))
- ctest = CTest(options)
- if ctest.sign_payloads: ctest.GeneratePublicKey()
+ ctest = CTest(opts)
+ if ctest.sign_payloads:
+ ctest.GeneratePublicKey()
ctest.FindTargetAndBaseImages()
- if not options.only_verify:
- ctest.GenerateUpdatePayloads(not options.quick_update)
+ if not opts.only_verify:
+ ctest.GenerateUpdatePayloads(not opts.quick_update)
try:
- ctest.RunAUTestHarness(options.only_verify, options.quick_update,
- options.suite)
+ ctest.RunAUTestHarness(opts.only_verify, opts.quick_update,
+ opts.suite)
except TestException as e:
- if options.verbose:
+ if opts.verbose:
cros_build_lib.Die(str(e))
sys.exit(1)
diff --git a/ctest/ctest_unittest.py b/ctest/ctest_unittest.py
index 3732f57..d2a5656 100755
--- a/ctest/ctest_unittest.py
+++ b/ctest/ctest_unittest.py
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python2
#
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
@@ -6,6 +6,8 @@
"""Test module containing unittests for CTest."""
+from __future__ import print_function
+
import mox
import os
import sys
@@ -37,12 +39,15 @@
# TODO(sosa): Can we create mock objects but call a real method in an easier
# way?
- ctest.CTest.__init__()
+ ctest.CTest.__init__(None)
+ # pylint: disable=no-value-for-parameter
+ # For some reason pylint is not happy with the way mox mocks out functions.
image_extractor.ImageExtractor.GetLatestImage(
'target_version').AndReturn(None)
+ # pylint: enable=no-value-for-parameter
self.mox.ReplayAll()
- ctester = ctest.CTest() # Calls mocked out __init__.
+ ctester = ctest.CTest(None) # Calls mocked out __init__.
ctester.target = 'some_image/target_version/file.bin'
ctester.base = None
ctester.archive_dir = self.ARCHIVE_DIR
@@ -64,14 +69,16 @@
latest_base_path = os.path.join(
'/some/fake/path', image_extractor.ImageExtractor.IMAGE_TO_EXTRACT)
- ctest.CTest.__init__()
+ ctest.CTest.__init__(None)
+ # pylint: disable=no-value-for-parameter
image_extractor.ImageExtractor.GetLatestImage('target_version').AndReturn(
latest_base_dir)
image_extractor.ImageExtractor.UnzipImage(latest_base_dir).AndReturn(
latest_base_path)
+ # pylint: enable=no-value-for-parameter
self.mox.ReplayAll()
- ctester = ctest.CTest() # Calls mocked out __init__.
+ ctester = ctest.CTest(None) # Calls mocked out __init__.
ctester.target = 'some_image/target_version/file.bin'
ctester.base = None
ctester.archive_dir = self.ARCHIVE_DIR
@@ -94,15 +101,17 @@
fake_crosutils = os.path.join(self.FAKE_ROOT, 'src', 'scripts')
- ctest.CTest.__init__()
+ ctest.CTest.__init__(None)
cros_build_lib.RunCommand(
mox.In('./get_latest_image.sh'), cwd=fake_crosutils, print_cmd=False,
redirect_stdout=True).AndReturn(fake_result)
+ # pylint: disable=no-value-for-parameter
image_extractor.ImageExtractor.GetLatestImage('latest_version').AndReturn(
None)
+ # pylint: enable=no-value-for-parameter
self.mox.ReplayAll()
- ctester = ctest.CTest() # Calls mocked out __init__.
+ ctester = ctest.CTest(None) # Calls mocked out __init__.
ctester.target = None
ctester.base = None
ctester.archive_dir = self.ARCHIVE_DIR
diff --git a/lib/constants.py b/lib/constants.py
index 9c01596..f8f7055 100644
--- a/lib/constants.py
+++ b/lib/constants.py
@@ -2,6 +2,10 @@
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
+"""Module containing contants to be used across crostestutils."""
+
+from __future__ import print_function
+
import os
_TEST_LIB_PATH = os.path.realpath(__file__)
