atheist 0.20110402-2.1 / usr / lib / python2.7 / dist-packages / atheist / manager.py

# -*- mode: python; coding: utf-8 -*-

# atheist
#
# Copyright (C) 2009,2010,2011 David Villa Alises
#
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 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 General Public License for more details.
#
# You should have received a copy of the GNU 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


#import codecs, locale
#sys.stdout = codecs.getwriter(locale.getpreferredencoding())(sys.stdout)
#sys.stderr = codecs.getwriter(locale.getpreferredencoding())(sys.stderr)



import os
import logging
import sys
import time
import inspect
import random
import optparse
import fnmatch
import ConfigParser
from itertools import chain
import signal


from pyarco.Pattern import Singleton
from pyarco.Conio import *
from pyarco.iniparser import IniParser

import atheist
import atheist.const
from atheist.utils import *
import atheist.plugins

from atheist import gvar
Log = gvar.Log


class ConfigFacade(object):
    '''forwards missing attributes and key-items to holder objects'''
    def __init__(self):
        self._holders = []

    def init(self, holder=None):
        if holder:
            self.add_holder(holder)


    def add_holder(self, holder):
        self._holders.append(holder)

    def __getattr__(self, name):
        for holder in self._holders:
            try:
                return getattr(holder, name)
            except (AttributeError, ConfigParser.NoSectionError), e:
                pass

        raise AttributeError(name)

    def __getitem__(self, name):
        for holder in self._holders:
            if not hasattr(holder, '__getitem__'):
                continue

            try:
                return holder[name]
            except KeyError:
                pass

        raise KeyError(name)


#    def get(self, name, value=None):
#        try:
#            return getattr(self, name)
#        except AttributeError,e:
#            pass
#
#        return value


class ReporterManager(list):

    def report_case(self, case):
        for r in self:
            r.add_case(case)



class Config(ConfigFacade):

    # FIXME: El parser debería crearse aquí
    def init(self, options, parser):
        ConfigFacade.init(self, options)
        self.parser = parser

        if self.describe:
            self.RunnerClass = atheist.DescribeRunner
        elif self.list_only:
            self.RunnerClass = atheist.ListOnlyRunner
        else:
            self.RunnerClass = atheist.ThreadPoolRunner


        self.reporters = ReporterManager()

        self._check_sanity()

        if not self.quiet:
            self.reporters.append((atheist.ConsoleReporter()))

        self.pid = str(os.getpid())
        self.init_time = time.time()


    def _check_sanity(self):
        if self.stdout and self.out_on_fail:
            self.parser.error("-o/--stdout and -f/--out-on-fail are incompatible options")

        if self.stderr and self.out_on_fail:
            self.parser.error("-e/--stderr and -f/--out-on-fail are incompatible options")

        if self.verbosity and self.quiet:
            self.parser.error("-q/--quiet and -v/--verbose are incompatible options")

        if self.use_timetag and self.verbosity == 0 or self.quiet:
            self.parser.error("-t/--timetag requires some verbosity")

        if self.workers < 0:
            self.parser.error("-w/--workers can not be negative.")

        if self.verbosity:
            self.report_detail = 10

        for path in self.pluginpath:
            if not os.path.exists(path):
                self.parser.error("plugin directory '{0}' not exist".format(path))


    def load_user_config(self):
        if os.path.exists(self.config_file):
            ini = IniParser(self.config_file)
            self.add_holder(ini)
        else:
            if self.config_file != ATHEIST_CFG:
                Log.warning("Config file '%s' does not exists" % self.config_file)


    def load_exclude_patterns(self):
        self.exclude = EXCLUDE
        if self.ignore:
            self.exclude += self.ignore.split(',')

        try:
            self.exclude += self.ui.ignore.split(',')
        except AttributeError:
            pass


    def configure_loglevel(self):
        self.loglevel = logging.WARNING
        if self.verbosity == 1:
            self.loglevel = logging.INFO
        elif self.verbosity >= 2:
            self.loglevel = logging.DEBUG
        if self.quiet:
            self.loglevel = logging.ERROR

        Log.setLevel(self.loglevel)


    def configure_timetag(self):
        self.timetag = ''
        if self.use_timetag:
            self.timetag = '%(asctime)s '
            formatter = atheist.log.CapitalLoggingFormatter(\
                self.timetag + '[%(levelcapital)s] %(message)s',
                atheist.log.DATEFORMAT)
            Log.handlers[0].setFormatter(formatter) # change default formatter



def public(fn):
    fn.public = True
    return fn


class TaskDelegate:
    def __init__(self, cls, mng):
        self.cls = cls
        self.mng = mng

    def __call__(self, *args, **kargs):
#        try:
        return self.cls(self.mng, *args, **kargs)
#        except TypeError,e:
#            raise TypeError("%s: %s" % (self.cls.__name__, e))

    def __str__(self):
        return "<TaskDelegate '{0}'>".format(self.cls.__name__)


class InternalAPI:
    '''Namespace for atheist API'''

    def __init__(self, mng):
        self.mng = mng

        self._temp_n = 0
        self._temp_lock = threading.Lock()

        self.symbols = [x[1] for x in inspect.getmembers(
                self, inspect.ismethod) if hasattr(x[1], 'public')]

        pluginpath = gvar.the_config.pluginpath + \
            [os.path.join(os.path.dirname(atheist.plugins.__file__))]

        self.plugins = []
        for dname in pluginpath:
            self.plugins.extend(atheist.plugins.Loader(dname))

        self.symbols.extend([x for x in self.plugins
                             if issubclass(x, (atheist.Task, atheist.Condition))])

        # add built-ins classes to the scope
        for name,symbol in inspect.getmembers(atheist, inspect.isclass):
            if issubclass(symbol, atheist.Public):
                self.symbols.append(symbol)

        self.scope = {}
        for s in self.symbols:
            symbol_name = s.__name__
            if inspect.isclass(s) and issubclass(s, atheist.Task):
                s = TaskDelegate(s, mng)
            self.scope[symbol_name] = s


    def get_Task_classes(self):
        return [x for x in self.symbols if inspect.isclass(x) and issubclass(x, atheist.Task)]

    @public
    def get_task(self, _id):
        for t in chain(self.mng.ts):
            if t.tid == _id: return t
        Log.error("There is no task with id '%s'" % _id)
        return None

    @public
    def load(self, fname):
        _globals = self.mng.exec_env.copy()
        globals_before_keys = _globals.keys()

        atheist.exec_file(fname, _globals)

        # FIXME: better imp.new_module?
        retval = types.ModuleType(fname)
        for key,val in _globals.items():
            if key not in globals_before_keys:
                setattr(retval, key, val)

        return retval

    @public
    def temp_name(self, prefix=''):
        with self._temp_lock:
            self._temp_n += 1
            return os.path.join(ATHEIST_TMP, 'temp-{0}{1}'.format(
                    prefix + '-' if prefix else '',
                    self._temp_n))



class Manager(object):
    pass


class DefaultManager(Manager):
    instances = 0

    class NoTaskDefined(Exception): pass

    def __init__(self, argv=[]):

        assert DefaultManager.instances == 0, "Just one manager"
        DefaultManager.instances = 1

        signal.signal(signal.SIGPIPE, signal.SIG_DFL)
        signal.signal(signal.SIGINT, self.abort_handler)
        signal.signal(signal.SIGTERM, self.abort_handler)

        self.aborted = False
        self.abort_observers = []

        self.ts = []
        self.suite = atheist.Suite()

        self.parser = OptParser(usage=USAGE, version=atheist.const.VERSION)
        opt_cfg, args = self.parser.parse_args(argv)
#        self.cfg = Config(opt_cfg, self.parser)

        gvar.the_config.init(opt_cfg, self.parser)
        self.cfg = gvar.the_config


        # FIXME: debaería hacerlo main()
        if self.cfg.gen_template:
            print atheist.file_template()
            sys.exit(0)

        self.cfg.configure_loglevel()
        self.cfg.configure_timetag()

        if self.cfg.log:
            filelog = logging.FileHandler(self.cfg.log)
#            filelog.setFormatter(formatter)
            filelog.setLevel(logging.DEBUG)
            Log.addHandler(filelog)

        Log.debug('Log level is %s' % logging.getLevelName(Log.level))


        # make a new parser to get plugin options
        self.parser = OptParser(usage=USAGE, version=atheist.const.VERSION)
        self.parser.ignore_unknown = False

        self.api = InternalAPI(self)

        self.cfg.load_user_config()

        plugin_opts = optparse.OptionGroup(self.parser, "Plugin options")

        for plugin in self.api.plugins:
            plugin_opts.add_options(plugin.get_options(self.cfg, self.parser))

        self.parser.add_option_group(plugin_opts)

        opt_cfg, self.args = self.parser.parse_args(argv)
        self.cfg.add_holder(opt_cfg)


        for plugin in self.api.plugins:
            plugin.config(self.cfg, self.parser)

        self.exec_env = self.api.scope
        self.exec_env['args'] = self.cfg.task_args.split(',')


        self.cfg.curdir = os.getcwd()
        relpath.root = os.path.abspath(self.cfg.basedir)
        relpath.enable = (os.getcwd() == relpath.root)
#        relpath.enable = False
        os.chdir(self.cfg.basedir)

        self.cfg.load_exclude_patterns()

        Log.debug("excluding %s" % self.cfg.exclude)

        cout_config(self.cfg.plain)



    def abort_handler(self, signum, frame):
        if self.aborted:
            return

        print
        Log.warning("C-c pressed!")
        self.abort()

    def abort(self):
        self.aborted = True
        for ob in self.abort_observers: ob()


    def reload(self):
        def create_inline(name, code):
            Log.debug("inline script: %s" % code)
            fname = os.path.join(ATHEIST_TMP, 'inline-%s.test' % name)
            with file(fname, 'w') as fd:
                fd.write(code)

            return fname

        self.suite = atheist.Suite()

        inlines = self.cfg.inline[:]
        for cmd in self.cfg.commands:
            inlines.append('Test("%s", shell=True)' % cmd)

        for i, code in enumerate(inlines):
            fname = create_inline(i, code)
            try:
                self.suite.append(atheist.TaskCase(self, fname))
            finally:
                Log.debug("removing %s" % fname)
                os.remove(fname)

        self.args = [abspath(x, self.cfg.curdir) for x in self.args]

        files = [x for x in self.args if os.path.isfile(x)]
        dirs =  [x for x in self.args if os.path.isdir(x)]

        rest = set(self.args).difference(set(files).union(dirs))
        if rest:
            for f in rest:
                Log.warning("Ignored file: %s", relpath(f))

        for f in files:
            self.process_files('', [f], filter_excluded=False)

        for d in dirs:
            self.process_directory(d)

        if self.suite.num_tasks == 0:
            raise self.NoTaskDefined()


    def excluded(self, fname):
        return any(fnmatch.fnmatch(fname, pattern) for pattern in self.cfg.exclude)


    def process_files(self, root, files, filter_excluded=True):
        for f in sorted([x for x in files]):
            if filter_excluded and (self.excluded(f) or not f.endswith(ATHEIST_EXT)):
                continue

            fullname = os.path.join(root, f)
            if not os.path.exists(fullname):
                Log.warning("No such file or directory: '%s', skipped.   ", fullname)
                continue

            self.suite.append(atheist.TaskCase(self, fullname))


    def process_directory(self, dname):
        for root, dirs, files in os.walk(dname):
            dirs.sort()
            Log.info("Entering directory '%s'" % relpath(root))

            if not self.cfg.skip_hooks:
                dir_setup = os.path.join(root, DIR_SETUP)
                if os.path.exists(dir_setup):
                    self.suite.append(atheist.TaskCase(self, dir_setup, {},
                                                       atheist.task_mode.DIR_SETUP))

            self.process_files(root, files)

            if not self.cfg.skip_hooks:
                dir_teardown = os.path.join(root, DIR_TEARDOWN)
                if os.path.exists(dir_teardown):
                    self.suite.append(atheist.TaskCase(self, dir_teardown, {},
                                                       atheist.task_mode.DIR_TEARDOWN))

            for d in dirs[:]:
                if self.excluded(d):
                    dirs.remove(d)


    def run(self, runner_cls=None):

        if not self.args and not self.cfg.inline and not self.cfg.commands:
            self.parser.print_help()
            return 1

        if self.cfg.clean_only:
            atheist.remove_files_and_dirs(atheist.get_generated_from_finished_atheists())
            return 0

        atheist.assure_tmp_dir()

        if runner_cls == None:
            runner_cls = self.cfg.RunnerClass


        retval = 0

        try:
            runner = runner_cls(self)
            retval = runner.run()
        except self.NoTaskDefined:
            Log.warning("No testfiles found.")
            retval = 1
        except atheist.ExecutionError:
            retval = 1

#        self.clean_async(
#            os.path.join(ATHEIST_TMP_BASE, self.cfg.pid),
#            [ATHEIST_TMP])

        if self.cfg.clean:
            atheist.remove_files_and_dirs([ATHEIST_TMP])

        return retval


    def clean_async(self, fname, generated):
        generated = generated
        if self.cfg.clean:
            atheist.remove_files_and_dirs(generated)
        else:
            atheist.save_dirty_filelist(fname + '.gen', generated)


class OptParser(optparse.OptionParser):
    def __init__(self, *args, **kargs):
        self.ignore_unknown = True
        optparse.OptionParser.__init__(self, *args, **kargs)

        self.add_option('-a', '--task-args', dest='task_args', metavar='ARGS',
                        default='',
                        help='colon-separated options for the tasks')

        self.add_option('-b', '--base-dir', dest='basedir', metavar='PATH',
                        default=os.getcwd(),
                        help='set working directory')

        self.add_option('-C', '--clean-only', dest='clean_only',
                        action='store_true',
                        help="execute nothing, only remove generated files")

        self.add_option('-d', '--describe', dest='describe',
                        action='store_true',
                        help='execute nothing, only describe tasks')

        self.add_option('-e', '--stderr', dest='stderr',
                        action='store_true',
                        help='print task process stderr')

        self.add_option('-f', '--out-on-fail', dest='out_on_fail',
                        action='store_true',
                        help='print task output (stout/stderr) but only if it fail')

        self.add_option('-g', '--gen-template', dest='gen_template',
                        action='store_true',
                        help='generate a taskcase file template with default values')

# FIXME: test this
        self.add_option('-i', '--report-detail', metavar='LEVEL', default=1, type=int,
                        help='Report verbosity (0:nothing, [1:case], 2:task, 3:composite, 4:condition)')

        self.add_option('-j', '--skip-hooks', dest='skip_hooks',
                          action='store_true',
                        help='skip _setup and _teardown files')

        self.add_option('-k', '--keep-going', dest='keep_going',
                        action='store_true', default=False,
                        help='continue even with failed tasks')

        self.add_option('-l', '--list', dest='list_only',
                        action='store_true',
                        help='list tasks but do not execute them')

        self.add_option('-o', '--stdout', dest='stdout',
                        action='store_true',
                        help='print task stdout')

        # FIXME: probar esto
        self.add_option('-p', '--plugin-dir', dest='pluginpath', metavar='PATH',
                        action='append', default=[],
                        help='a directory containing plugins')

        self.add_option('-q', '--quiet', dest='quiet',
                        action='store_true',
                        help='do not show result summary nor warnings, only totals')

        self.add_option('-r', '--random', dest='random', type='int', default=None,
                        help='shuffle taskcases using the specified seed (0:random seed)')

        self.add_option('-s', '--script', dest='inline',
                        action='append', default=[],
                        help='specifies command line script')

        self.add_option('-t', '--time-tag', dest='use_timetag',
                        action='store_true',
                        help='include time info in the logs')

        self.add_option('-v', '--verbose', dest='verbosity',
                        action='count', default=0,
                        help='incresse verbosity')

        self.add_option('-w', '--workers', default=1, type=int,
                        help='number of simultaneous tasks (deafult:1) 0:auto-select.')

        self.add_option('-x', '--exec', dest='commands', metavar='COMMAND',
                        action='append', default=[],
                        help='exec like in "Test(COMMAND)"')

        self.add_option('--case-time', default=False,
                        action='store_true',
                        help="print case execution time in reports")

        self.add_option('--cols', dest='screen_width', default='80', metavar='WIDTH',
                          type=int, help="terminal width (in chars)")

        self.add_option('--config', dest='config_file',
                        default=ATHEIST_CFG, metavar='FILE',
                        help="alternate config file")

        self.add_option('--dirty', dest='clean',
                          action='store_false', default=True,
                          help="do not remove generated files after task execution")

        self.add_option('--disable-bar', action='store_true',
                          help="don't show progress bar")

        # FIXME: a test for this
        a= self.add_option('--ignore', metavar='PATTERN', dest='ignore', default='',
                        help="test files to ignore (glob patterns) separated with semicolon")

        self.add_option('--log', default=None,
                          help="log to specified filename")

        self.add_option('--plain', default=False, action='store_true',
                          help="avoid color codes in console output")

        self.add_option('--save-stdout', action='store_true', default=False,
                          help="save stdout of all tasks")

        self.add_option('--save-stderr', action='store_true', default=False,
                          help="save stderr of all tasks")



# YAGNI
#        self.add_option('-x', '--extension', dest='ext',
#                          default='.test',
#                          help='file extension for the task files')



    def _process_args(self, largs, rargs, values):
        "Override OptParse._process_args to give ability to ignore unknown args"
        while rargs:
            arg = rargs[0]
            if arg == "--":
                del rargs[0]
                return
            elif arg[0:2] == "--":
                try:
                    self._process_long_opt(rargs, values)
                except optparse.BadOptionError, e:
                    if not self.ignore_unknown:
                        raise

            elif arg[:1] == "-" and len(arg) > 1:
                try:
                    self._process_short_opts(rargs, values)
                except optparse.BadOptionError, e:
                    if not self.ignore_unknown:
                        raise

            elif self.allow_interspersed_args:
                largs.append(arg)
                del rargs[0]
            else:
                return