testutils.py
author David Douard <david.douard@logilab.fr>
Wed, 14 May 2014 22:28:57 +0200
changeset 1520 0abe241164e2
parent 1515 2a12802efe66
child 1523 7157179d08dd
permissions -rw-r--r--
[test] fix tests when executed in an apycot environment ie. when tests are executed from sources towards an installed apycot. The fix consist mainly in "moving" the input_path function from testutils.py (which is part of the installed modules) into unittest_checkers.py so we are sure we look for test data files relatively to the test directory (and not the location of the installation).

import sys
import shutil
import os
from datetime import datetime
from os.path import join, dirname, abspath

from logilab.common.testlib import mock_object

from cubicweb.devtools.testlib import CubicWebTC

from cubes.vcsfile.testutils import init_vcsrepo
from cubes.narval.testutils import NarvalBaseTC
from cubes.apycot.recipes import quick_script

try:
    import apycotlib
except:
    from cubes.apycot import _apycotlib as apycotlib
sys.modules['apycotlib'] = apycotlib

import narvalbot
if narvalbot.MODE == 'dev':
    PLUGINSDIR = join(dirname(__file__), '_narval')
else:
    from cubes.narval.__pkginfo__ import NARVAL_DIR
    PLUGINSDIR = join(narvalbot.INSTALL_PREFIX, NARVAL_DIR)
sys.path.append(PLUGINSDIR)

from apycotlib.writer import CheckDataWriter, BaseDataWriter


class DummyStack(object):

    def __init__(self):
        self.msg = None
        self.clear()

    def __getitem__(self, idx):
        return self

    def __len__(self):
        return 0

    def clear(self):
        self.msg = []
        self.append = self.msg.append


class MockBaseWriter(BaseDataWriter):

    def __init__(self):
        super(MockBaseWriter, self).__init__( MockConnection, None)

    def skip(self, *args, **kwargs):
        pass

    def _debug(self, *args, **kwargs):
        print args, kwargs

    def set_exec_status(self, status):
        self._logs.append('<internal> SETTING EXEC STATUS: %s' % status)

    raw = execution_info = skip
    close = skip


class MockTestWriter(MockBaseWriter):
    """fake apycot.IWriter class, ignore every thing"""

    def make_check_writer(self):
        return MockCheckWriter()

    link_to_revision = MockBaseWriter.skip


class MockCheckWriter(MockBaseWriter):
    """fake apycot.IWriter class, ignore every thing"""

    def start(self, checker):
        self._logs.append('<internal>STARTING %s' % checker.id)

    def clear_writer(self):
        self._log_stack = DummyStack()


class MockTest(object):
    """fake apycot.Test.Test class"""
    def __init__(self, repo=None):
        self.repo = repo
        self.tmpdir = 'data'
        self.environ = {}
        self.checkers = []
        self._apycot_config = {}

    def project_path(self, subpath=False):
        return self.repo.co_path

    @property
    def tconfig(self):
        return mock_object(testconfig={}, name='bob', subpath=None)

    def apycot_config(self, something=None):
        return self._apycot_config


class MockVCSFile(dict):
    def __init__(self, _type, source_url=None, path=None):
        super(MockVCSFile, self).__init__(
                type=_type, source_url=source_url, path=path, local_cache=None)


class MockRepository:
    """fake apycot.IRepository class"""
    branch = None
    def __init__(self, attrs=None, **kwargs):
        self.__dict__.update(kwargs)
        self.co_path = self.path

    def co_command(self):
        return self.command

    def co_move_to_branch_command(self):
        return None

    def __repr__(self):
        return '<MockRepository %r>' % self.__dict__

    def revision(self):
        pass


class MockConnection(object):
    """fake pyro connexion"""
    def close(self):
        pass
    def execute(*args, **kwargs):
        pass
    def commit(self):
        pass


class ApycotBaseTC(NarvalBaseTC):

    recipescript = quick_script

    def setup_database(self):
        """ self.repo: used to get the session to connect to cw
            self.vcsrepo: new entity
        """

        req = self.request()
        self.lgce = req.create_entity(
            'ProjectEnvironment', name=u'lgce',
            check_config=u'install=python_setup\nenv-option=value',
            check_environment=u'SETUPTOOLS=1\nDISPLAY=:2.0'
            )
        self.vcsrepo = req.create_entity('Repository', type=u'mercurial',
                                         # use path to avoid clone attempt when using url
                                         path=unicode(self.datapath('project')),
                                         reverse_local_repository=self.lgce)
        self.pyp = req.create_entity('TestConfig', name=u'PYTHONPACKAGE',
                                     check_config=u'python_lint_treshold=7\n'
                                     'python_lint_ignore=thirdparty\n'
                                     'python_test_coverage_treshold=70\n',
                                     check_environment=u'NO_SETUPTOOLS=1\nDISPLAY=:1.0')
        self.recipe = req.execute('Recipe X WHERE X name "apycot.recipe.quick"').get_entity(0, 0)
        # reset vcsrepo (using the session )
        init_vcsrepo(self.repo)
        # reset recipe content
        self.recipe.set_attributes(script=self.recipescript)
        self.lgc = self.add_test_config(u'lgc', env=self.lgce, group=self.pyp, use_recipe=self.recipe)

        self.repo.threaded_task = lambda func: func() # XXX move to cw

    def add_test_config(self, name,
                        check_config=u'python_lint_treshold=8\npouet=5',
                        env=None, group=None, **kwargs):
        """add a TestConfig instance"""
        req = self.request()
        if group is not None:
            kwargs['refinement_of'] = group
        if env is not None:
            kwargs['use_environment'] = env
        return req.create_entity('TestConfig', name=name,
                                 check_config=check_config, **kwargs)

    def dumb_execution(self, ex, check_defs, setend=True):
        """add a TestExecution instance"""
        req = self.request()
        for name, status in check_defs:
            cr = req.create_entity('CheckResult', name=unicode(name), status=unicode(status))
            req.execute('SET X during_execution Y WHERE X eid %(x)s, Y eid %(e)s',
                        {'x': cr.eid, 'e': ex.eid})
        if setend:
            req.execute('SET X status "success" '
                        'WHERE X eid %(x)s', {'x': ex.eid})