Source code for autotest.client.cmdparser

Autotest command parser

:copyright: Don Zickus <> 2011

import logging
import os
import re
import sys

from autotest.client import harness
from autotest.client import os_dep, utils
from autotest.client.shared import logging_config, logging_manager
from autotest.client.shared import packages, error
from autotest.client.shared.settings import settings

GLOBALDIRTEST = settings.get_value('COMMON', 'test_dir', default="")

    autodir = os.path.abspath(os.environ['AUTODIR'])
except KeyError:
    autodir = settings.get_value('COMMON', 'autotest_top_path')
tmpdir = os.path.join(autodir, 'tmp')

output_dir = settings.get_value('COMMON', 'test_output_dir', default=tmpdir)

FETCHDIRTEST = os.path.join(output_dir, 'site_tests')

if not os.path.isdir(FETCHDIRTEST):

DEBUG = False

[docs]class CmdParserLoggingConfig(logging_config.LoggingConfig): """ Used with the sole purpose of providing convenient logging setup for the KVM test auxiliary programs. """
[docs] def configure_logging(self, results_dir=None, verbose=False): super(CmdParserLoggingConfig, self).configure_logging(use_console=True, verbose=verbose)
[docs]class CommandParser(object): """ A client-side command wrapper for the autotest client. """ COMMAND_LIST = ['help', 'list', 'run', 'fetch', 'bootstrap'] @classmethod def _print_control_list(cls, pipe, path): """ Print the list of control files available. :param pipe: Pipe opened to an output stream (may be a pager) :param path: Path we'll walk through """ if not os.path.isdir(path): pipe.write("Test directory not available\n") return pipe.write(" %-50s %s\n" % ("[Control]", "[Description]")) # The strategy here is to walk the root directory # looking for "*control*" files in some directory # and printing them out for root, _, files in sorted(os.walk(path)): for name in files: if"control", name): # strip full path basename = re.sub(path + "/", "", root) text = "%s/%s" % (basename, name) desc = "None" if name == "control": # Imply /control by listing only directory name text = "%s" % basename is_control = False for line in open(root + "/" + name).readlines(): if re.match("NAME", line): # We have a description line desc = re.split("=\s*", line, maxsplit=1)[1].rstrip() try: desc = desc[1:-1] except IndexError: pass is_control = True break if not is_control: # The current file is not a control file continue pipe.write(' %-50s %s\n' % (text, desc))
[docs] def fetch(self, args, options): """ fetch a remote control file or packages """ if not len(args): url = args.pop(0) if not utils.is_url(url):"Not a remote url, nothing to fetch (%s)", url) if len(args): name = args.pop(0) else: name = """Fetching file %s:%s", url, name) pkg_dir = os.path.join(output_dir, 'packages') install_dir = os.path.join(FETCHDIRTEST, name) pkgmgr = packages.PackageManager(output_dir, run_function_dargs={'timeout': 3600}) pkgmgr.install_pkg(name, 'test', pkg_dir, install_dir, repo_url=url) raise SystemExit(0)
[docs] @classmethod def help(cls): """ List the commands and their usage strings. :param args is not used here. """"Commands:")"bootstrap [-H <harness>] Use harness to fetch control file")"\tcan also export AUTOTEST_HARNESS=<harness> to avoid -H option")"fetch <url> [<file>]\tFetch a remote file/package and install it")"\tgit://...:[<branch>] [<file/directory>]")"\thttp://... [<file>]")"help\t\t\tOutput a list of supported commands")"list\t\t\tOutput a list of available tests")"run <test> [<args>]\tFind given <test> in path and run with args") raise SystemExit(0)
[docs] @classmethod def list_tests(cls): """ List the available tests for users to choose from """ # One favorite feature from git :-) try: less_cmd = os_dep.command('less') pipe = os.popen('%s -FRSX' % less_cmd, 'w') except ValueError: pipe = sys.stdout pipe.write("List of tests available\n") pipe.write("Unless otherwise specified, outputs imply /control files\n") pipe.write("\n") # Walk local ./tests directory dirtest = os.path.join(os.path.abspath(os.path.curdir), LOCALDIRTEST) # Don't repeat autodirtest results if not os.environ['AUTODIRTEST']: dirtest = os.environ['AUTODIRTEST'] pipe.write("Local tests (%s)\n" % dirtest) cls._print_control_list(pipe, dirtest) pipe.write("\n") # Walk fetchdirtest directory if FETCHDIRTEST and os.path.isdir(FETCHDIRTEST): dirtest = FETCHDIRTEST pipe.write("Remotely fetched tests (%s)\n" % dirtest) cls._print_control_list(pipe, dirtest) pipe.write("\n") # Walk globaldirtests directory if GLOBALDIRTEST and os.path.isdir(GLOBALDIRTEST): dirtest = GLOBALDIRTEST pipe.write("Globally imported tests (%s)\n" % dirtest) cls._print_control_list(pipe, dirtest) pipe.write("\n") # Walk customtest directory if 'CUSTOM_DIR' in os.environ: dirtest = os.environ['CUSTOM_DIR'] pipe.write("Custom Test Directory (%s)\n" % dirtest) cls._print_control_list(pipe, dirtest) pipe.write("\n") # Walk autodirtest directory dirtest = os.environ['AUTODIRTEST'] pipe.write("Autotest prepackaged tests (%s)\n" % dirtest) cls._print_control_list(pipe, dirtest) pipe.close() raise SystemExit(0)
[docs] def parse_args(self, args, options): """ Process a client side command. :param args: Command line args. """ logging_manager.configure_logging(CmdParserLoggingConfig(), verbose=options.verbose) if len(args) and args[0] in self.COMMAND_LIST: cmd = args.pop(0) else: # Do things the traditional way return args # List is a python reserved word if cmd == 'list': cmd = 'list_tests' try: try: args = getattr(self, cmd)(args, options) except TypeError: args = getattr(self, cmd)() except SystemExit, return_code: sys.exit(return_code.code) except Exception, error_detail: if DEBUG: raise sys.stderr.write("Command failed: %s -> %s\n" % (cmd, error_detail)) sys.exit(1) # Args are cleaned up, return to process the traditional way return args
[docs] def run(self, args, options): """ Wrap args with a path and send it back to autotest. """ if not len(args): test = args.pop(0) # Autotest works on control files if not"control", test): test = test + "/control" localdir = os.path.join(os.path.abspath(os.path.curdir), LOCALDIRTEST) fetchdir = FETCHDIRTEST globaldir = GLOBALDIRTEST autodir = os.environ['AUTODIRTEST'] if 'CUSTOM_DIR' in os.environ: customtestdir = os.environ['CUSTOM_DIR'] else: customtestdir = "" for dirtest in [localdir, fetchdir, globaldir, autodir, customtestdir]: if dirtest: d = os.path.join(dirtest, test) if os.path.isfile(d): args.insert(0, d) return args logging.error("Can not find test %s", test) raise SystemExit(1)
[docs] def bootstrap(self, args, options): """ Bootstrap autotest by fetching the control file first and pass it back Currently this relies on a harness to retrieve the file """ def harness_env(): try: return os.environ['AUTOTEST_HARNESS'] except KeyError: return None def harness_args_env(): try: return os.environ['AUTOTEST_HARNESS_ARGS'] except KeyError: return None class stub_job(object): def config_set(self, name, value): return if not options.harness and not harness_env(): if options.harness: harness_name = options.harness elif harness_env(): harness_name = harness_env() options.harness = harness_name if options.harness_args: harness_args = options.harness_args else: harness_args = harness_args_env() options.harness_args = harness_args myjob = stub_job() # let harness initialize itself try: myharness =, myjob, harness_args) if not getattr(myharness, 'bootstrap'): raise error.HarnessError("Does not support bootstrapping\n") except Exception, error_detail: if DEBUG: raise sys.stderr.write("Harness %s failed to initialize -> %s\n" % (harness_name, error_detail)) sys.exit(1) # get remote control file and stick it in FETCHDIRTEST try: control = myharness.bootstrap(FETCHDIRTEST) except Exception, ex: sys.stderr.write("bootstrap failed -> %s\n" % ex) raise SystemExit(1) logging.debug("bootstrap passing control file %s to run" % control) if not control: # nothing to do politely abort # trick to work around various harness quirks raise SystemExit(0) args.append(control) return args