path: root/bin/insanity-run

#!/usr/bin/env python
# -*- mode: python; -*-
#
# Copyright (c) 2008 Nokia Corporation
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
# Boston, MA 02110-1301, USA.
#
# Authors: Rene Stadler <rene.stadler@nokia.com>
#

import sys
import os
import argparse

import pygtk
pygtk.require("2.0")
del pygtk

import insanity
import insanity.config

from insanity.client import CommandLineTesterClient
from insanity.testrun import TestRun, XmlTestRun

from insanity.storage.sqlite import SQLiteStorage
from insanity.generators.filesystem import FileSystemGenerator, URIFileSystemGenerator
from insanity.generators.playlist import PlaylistGenerator
from insanity.generators.external import ExternalGenerator
from insanity.generators.constant import ConstantGenerator
from insanity.monitor import ValgrindMemCheckMonitor, GDBMonitor, TerminalRedirectionMonitor

generators = {"filesystem": FileSystemGenerator,
              "urifilesystem": URIFileSystemGenerator,
              "playlist": PlaylistGenerator,
              "external": ExternalGenerator,
              "constant": ConstantGenerator}

class Client(CommandLineTesterClient):

    __software_name__ = "insanity-run"

    def __init__(self, verbose=False, singlerun=True, *a, **kw):

        CommandLineTesterClient.__init__(self, verbose=verbose, singlerun=singlerun, *a, **kw)

class ArgumentParser(argparse.ArgumentParser):

    def __init__(self):

        argparse.ArgumentParser.__init__(self)

        self.add_argument("-s",
                        "--storage",
                        dest="storage",
                        action="store",
                        help="configure data storage (default: sqlite:testrun.db)",
                        metavar="SPEC",
                        default="sqlite:testrun.db")
        self.add_argument("-o",
                        "--output",
                        dest="output",
                        action="store",
                        help="output directory (default: current)",
                        metavar="DIRECTORY",
                        default=".")
        self.add_argument("-x",
                        "--xmlpath",
                        dest="xmlpath",
                        action="store",
                        help="Path to an XML file describing the tests to run",
                        metavar="XMLPATH",
                        default=None)
        self.add_argument("-T",
                        "--tests",
                        dest="tests",
                        action="store",
                        help="tests directory (default: %s)" % insanity.config.Config.test_dir,
                        metavar="TESTS",
                        default=insanity.config.Config.test_dir)
        self.add_argument("-l",
                        "--substitutes-list",
                        dest="substitutes",
                        help="List of words to substitues in the XML file " \
                            "in the form of '-l old:new,old1:new1'",
                        metavar="SUBSTITUTES",
                        default=None)
        self.add_argument("-t",
                        "--test",
                        dest="test",
                        help="test or scenario to run (pass help for list of tests)",
                        metavar="TESTNAME",
                        default=None)
        self.add_argument("-a",
                        "--args",
                        dest="args",
                        nargs="+",
                        action="store",
                        help="set test arguments (pass help for list of arguments)",
                        metavar="SPEC",
                        default=None)
        self.add_argument("--gdb",
                        dest="gdb",
                        action="store_true",
                        help="Use gdb to gather a stack trace after a crash",
                        default=None)
        self.add_argument("--valgrind",
                        dest="valgrind",
                        action="store_true",
                        help="run tests on valgrind",
                        default=None)
        self.add_argument("--valgrind-supp",
                        dest="supp",
                        action="append",
                        help="add a valgrind suppression file to use",
                        metavar="SUPP",
                        default=None)
        self.add_argument("--compress-output-files",
                        dest="compress_output",
                        action="store_true",
                        help="Whether to compress the output files",
                        default=False)

    def parse_args(self, a):
        options = argparse.ArgumentParser.parse_args(self, a)
        options.storage = self.__parse_storage(options.storage)
        options.args = self.__parse_args(options.args)
        options.substitutes = self.__parse_subsitutes(options.substitutes)

        return options

    def __parse_subsitutes(self, value):
        dic = {}

        if not value:
            return dic

        for sub in value.split(","):
            try:
                o, n = sub.split(":")
            except ValueError, e:
                print "Wrong key value pair: %s, Reason %s" %(sub, e)
                continue
            dic[o] = n

        return dic

    def __parse_storage(self, value):

        if not value or value == "help" or not ":" in value:
            return "help"

        type_ = value.split(":")[0]
        arg = value[len(type_)+1:]

        return (type_, arg,)

    def __parse_args(self, args):
        if args is None:
            return None

        if args == "help" or "help" in args:
            return "help"

        result = []
        for arg in args:
            if not ":" in arg:
                return "help"
            (arg_name, rest) = arg.split(":", 1)

            found = False
            for generator in generators.keys():
                if rest.startswith(generator + ":"):
                    found = True
                    break

            if not found:
                gen_name = "constant"
                gen_args = rest
            else:
                (gen_name, gen_args) = rest.split(":", 1)
            result.append((arg_name, gen_name, gen_args,))

        return result

def storage_help():

    print "Possible arguments for --storage (-s):"
    # TODO: Just supporting sqlite for now:
    print "  sqlite:<DATABASE-FILENAME>"

def test_help():

    print "Possible arguments for --test (-t):"
    all_tests = list(insanity.utils.list_available_tests())
    all_tests.extend(insanity.utils.list_available_scenarios())
    for test in sorted(all_tests):
        print "  %s (%s)" % (test.__test_name__, test.__test_description__,)

def args_help(test_name):

    print "Usage for --args (-a) option:"
    print "  --args ARG ARG1..."
    print "Each ARG in the space separated list takes the following form:"
    print "  ARGLIST:[GENERATOR:]GENERATOR-ARGUMENTS"
    print "ARGLIST is a single argument. If no generator is provided a constant"
    print "value is used for the argument, otherwise the generator should generate"
    print "the appropriate number of arguments."
    print ""
    print "Possible generators and arguments:"
    print "  filesystem:PATH"
    print "  urifilesystem:PATH"
    print "  playlist:PATH"
    print "  external:COMMANDLINE"
    print ""
    print "Examples:"
    print "  uri:file://foo/bar"
    print "  uri:urifilesystem:/testclips"
    print "  uri:playlist:/home/user/playlist.txt"
    print "  uri:external:\"find `pwd` | sed -e s:^:file\\\://:\""
    print "  uri:urifilesystem:/testclips videodec:playlist:/decoders"

    if not test_name:
        return

    test = insanity.utils.get_test_metadata(test_name)
    print ""
    print "Arguments for test %s:" % test_name

    args = test.getFullArgumentList()
    for arg in args:
        print "  %s: %s" % (arg, args[arg]["description"])
        print "    description: %s" % (args[arg]["full_description"])
        print "    type: %s" % (args[arg]["type"])
        print "    global: %d" % (args[arg]["global"])
        print "    default value: %s" % (str(args[arg]["default_value"]))
        print ""

def storage_closed():
    pass

def main():

    error = False
    parser = ArgumentParser()
    options = parser.parse_args(sys.argv[1:])

    if options.storage == "help":
        storage_help()
        return True

    if options.args == "help":
        args_help(options.test)
        return True

    insanity.utils.scan_for_tests(options.tests)

    if options.test == "help":
        test_help()
        return True
    elif options.test is None and options.xmlpath is None:
        parser.print_help()
        return True

    if options.test:
        test = insanity.utils.get_test_metadata(options.test)

        # our monitors
        monitors = []

        if options.gdb:
            gdbscriptfile = os.path.join(os.path.dirname(os.path.abspath(__file__)), "gdb.instructions")
            if not os.path.exists (gdbscriptfile):
                gdbscriptfile = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "share", "insanity", "gdb.instructions")
            if not os.path.exists (gdbscriptfile):
                print
                return True
            else:
                monitors.append((GDBMonitor, {"gdb-script" : gdbscriptfile}))

        if options.valgrind:
            monitors.append((ValgrindMemCheckMonitor,
                             {"suppression-files":options.supp}))

        monitors.append((TerminalRedirectionMonitor,
                         {"compress-outputfiles":options.compress_output}))

        test_arguments = {}
        for arg_name, gen_name, gen_args in options.args or []:
            if not gen_name or not gen_name in generators.keys():
                args_help(options.test)
                return True

            # FIXME: Hardcoded list.
            gen_class = generators[gen_name]

            if gen_args:
                # FIXME:
                if gen_class == PlaylistGenerator:
                    gen = gen_class(location=gen_args)
                elif gen_class == ExternalGenerator:
                    gen = gen_class(command=gen_args)
                elif gen_class == ConstantGenerator:
                    gen = gen_class(constant=gen_args)
                else:
                    gen = gen_class(paths=[gen_args])
            else:
                gen = gen_class()

            test_arguments[arg_name] = gen

        test_run = TestRun(maxnbtests=1, workingdir=options.output)
        try:
            test_run.addTest(test, arguments=test_arguments, monitors=monitors)
        except Exception, e:
            print 'Error: exception adding test: ', e
            error = True
    else:
        try:
            test_run = XmlTestRun(options.xmlpath, substitutes=options.substitutes, workingdir=options.output)
        except Exception, e:
            print 'Error: creating XmlTestRun ', e
            error = True

    if not error:
        storage_name, storage_args = options.storage
        if storage_name == "sqlite":
            storage = SQLiteStorage(path=storage_args)
        else:
            # FIXME: Support other storage backends.
            storage_help()
            return True

            # From now on, when returning on error, call: storage.close(callback=storage_closed)

            storage.close(callback=storage_closed)
            error = True

        client = Client()
        client.setStorage(storage)
        client.addTestRun(test_run)
        client.run()

    return error

if __name__ == "__main__":
    if main():
        sys.exit(1)