run

#!/usr/bin/python
"""
Run virt tests outside the autotest client harness.

@copyright: Red Hat 2012
"""
import os, sys, time


class StreamProxy(object):
    """
    Mechanism to supress stdout output, while keeping the original stdout.
    """

    def __init__(self, filename='/dev/null', stream=sys.stdout):
        """
        Keep 2 streams to write to, and eventually switch.
        """
        self.terminal = stream
        if filename is None:
            self.log = stream
        else:
            self.log = open(filename, "a")
        self.stream = self.log

    def write(self, message):
        """
        Write to the current stream.
        """
        self.stream.write(message)

    def flush(self):
        """
        Flush the current stream.
        """
        self.stream.flush()

    def switch(self):
        """
        Switch between the 2 currently available streams.
        """
        if self.stream == self.log:
            self.stream = self.terminal
        else:
            self.stream = self.log


def create_config_files(options):
    test_dir = os.path.dirname(sys.modules[__name__].__file__)
    shared_dir = os.path.abspath(os.path.join(test_dir, 'shared', 'cfg'))
    if options.type:
        test_dir = os.path.abspath(os.path.join(os.path.dirname(test_dir),
                                                options.type))
    elif options.config:
        test_dir = os.path.abspath(os.path.dirname(os.path.dirname(options.config)))
    utils_misc.create_config_files(test_dir, shared_dir, interactive=False)


def bootstrap(options, cartesian_parser):
    bcolors = standalone_test.bcolors
    standalone_test.print_stdout(bcolors.HEADER + "SETUP:" + bcolors.ENDC,
                                 end=False)
    t_begin = time.time()
    test_dir = os.path.dirname(sys.modules[__name__].__file__)
    if options.type:
        test_dir = os.path.abspath(os.path.join(os.path.dirname(test_dir),
                                                options.type))
    elif options.config:
        test_dir = os.path.abspath(os.path.dirname(os.path.dirname(options.config)))
    base_dir = data_dir.get_data_dir()
    default_userspace_paths = None
    check_modules = ["kvm", "kvm-%s" % utils_misc.get_cpu_vendor(verbose=False)]
    online_docs_url = "https://github.com/autotest/virt-test/wiki"
    failed = False
    reason = None

    try:
        utils_misc.virt_test_assistant(options.type, test_dir, base_dir,
                                       default_userspace_paths, check_modules,
                                       online_docs_url,
                                       restore_image=options.restore,
                                       interactive=False)
    except Exception, e:
        reason = e
        failed = True

    t_end = time.time()
    t_elapsed = t_end - t_begin

    if not failed:
        standalone_test.print_pass(t_elapsed)
    else:
        standalone_test.print_fail(t_elapsed)
        standalone_test.print_stdout("Setup error: %s" % reason)
        sys.exit(-1)

    standalone_test.print_stdout(bcolors.HEADER +
                             "DATA DIR: %s" % data_dir.get_backing_data_dir() +
                             bcolors.ENDC)


if __name__ == '__main__':
    import optparse
    option_parser = optparse.OptionParser()
    option_parser.add_option("-v", "--verbose",
                      action="store_true", dest="verbose",
                      help="Exhibit debug messages")
    option_parser.add_option("-t", "--type",
                      action="store", dest="type",
                      help="Choose test type (kvm, libvirt, v2v)")
    option_parser.add_option("-c", "--config",
                      action="store", dest="config",
                      help="Explicitly choose a cartesian config")
    option_parser.add_option("-r", "--restore-image",
                      action="store_true", dest="restore",
                      help="Restore image from pristine image")
    option_parser.add_option("--qemu-bin", action="store", dest="qemu",
                      help="Choose a custom qemu path to be tested")
    option_parser.add_option("--tests", action="store", dest="tests",
                      help=('List of space separated tests to be executed'
                            ' - example: --tests "boot, reboot, shutdown"'))
    option_parser.add_option("--list-tests", action="store_true", dest="list",
                      help="List tests available")
    option_parser.add_option("--data-dir", action="store", dest="datadir",
                      help="Path to a data dir (that locates ISOS and images)")
    options, args = option_parser.parse_args()

    if (not options.type) and (not options.config):
        print("No type (-t) or config (-c) options specified, aborting...")
        option_parser.print_help()
        sys.exit(1)

    if not options.verbose:
        # Effectively points the stderr FD (2) to /dev/null, silencing
        # stderr.
        out_fd = os.open('/dev/null', os.O_WRONLY | os.O_CREAT)
        try:
            os.dup2(out_fd, 2)
        finally:
            os.close(out_fd)
        sys.stderr = os.fdopen(2, 'w')
        # Replace stdout by our StreamProxy object, so we can supress all output
        # but the one we specifically want.
        sys.stdout = StreamProxy(filename="/dev/null", stream=sys.stdout)
    else:
        # We have full stdout output
        sys.stdout = StreamProxy(filename=None, stream=sys.stdout)

    try:
        from autotest.client import setup_modules
    except ImportError:
        try:
            autotest_dir = os.environ['AUTOTEST_PATH']
        except KeyError:
            sys.stdout.switch()
            print("Environment variable $AUTOTEST_PATH not set. "
                  "please set it to a path containing an autotest checkout")
            sys.exit(1)
        client_dir = os.path.join(autotest_dir, 'client')
        sys.path.insert(0, client_dir)
        import setup_modules
        sys.path.pop(0)
        setup_modules.setup(base_path=client_dir,
                            root_module_name="autotest.client")

    test_dir = os.path.abspath(os.path.dirname(sys.modules[__name__].__file__))
    sys.path.append(test_dir)

    from virttest import cartesian_config, standalone_test, utils_misc, data_dir

    standalone_test.configure_logging()
    standalone_test.configure_console_logging()

    if options.datadir:
        data_dir.set_backing_data_dir(options.datadir)

    create_config_files(options)

    cartesian_parser = cartesian_config.Parser()
    if options.config is not None:
        cfg = os.path.abspath(options.config)
    if options.config is None or options.list:
        cfg = os.path.join(test_dir, options.type, "cfg", "tests.cfg")

    cartesian_parser.parse_file(cfg)

    if options.list:
        try:
            less_cmd = utils_misc.find_command('less')
            pipe = os.popen('%s -FRSX' % less_cmd, 'w')
        except ValueError:
            pipe = sys.stdout
        index = 0
        for params in cartesian_parser.get_dicts():
            virt_test_type = params.get('virt_test_type', "")
            supported_virt_backends = virt_test_type.split(" ")
            if options.type in supported_virt_backends:
                index +=1
                shortname = params['shortname'].split(".")[6:]
                shortname = ".".join(shortname)
                pipe.write("%d %s\n" % (index, shortname))
        sys.exit(0)
    else:
        bootstrap(options, cartesian_parser)


    if options.qemu:
        d = os.path.dirname(options.qemu)
        cartesian_parser.parse_string("qemu_binary = %s" % options.qemu)
        cartesian_parser.parse_string("qemu_img_binary = %s" %
                            os.path.join(d, 'qemu-img'))
        cartesian_parser.parse_string("qemu_io_binary = %s" %
                                      os.path.join(d, 'qemu-io'))

    if options.type and options.tests:
        if options.config:
            standalone_test.print_stdout("You can't use -c together "
                                         "with -t or --tests")
            sys.exit(-1)
        tests = options.tests.split(" ")
        cartesian_parser.parse_string("only %s" % ", ".join(tests))
    elif options.type and not options.tests:
        if options.type == 'kvm':
            cartesian_parser.parse_string("only migrate")
        elif options.type == 'libvirt':
            cartesian_parser.parse_string("only "
                                          "unattended_install.import.import, "
                                          "virsh_domname, "
                                          "remove_guest.without_disk")

    standalone_test.run_tests(cartesian_parser)