testutils.py
author David Douard <david.douard@logilab.fr>
Sun, 09 Nov 2014 20:07:17 +0100
changeset 1772 139e0c756385
parent 1769 8e817371cba1
permissions -rw-r--r--
[recipes] allow update_recipes() to update an existing one of the same name Useful for the migration script.

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

from logilab.common.testlib import mock_object
from logilab.common.shellutils import unzip
from logilab.common.decorators import classproperty

from cubicweb.devtools import BASE_URL
from cubicweb.devtools.testlib import CubicWebTC

from cubes.vcsfile import bridge
from cubes.vcsfile.hooks import repo_cache_dir
from cubes.vcsfile.testutils import setup_repos, HGRCMixin

from cubes.narval.testutils import NarvalBaseTC

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)
PPATH = [PLUGINSDIR]
if 'PYTHONPATH' in os.environ:
    PPATH.append(os.environ['PYTHONPATH'])
os.environ['PYTHONPATH'] = ':'.join(PPATH)

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 MockCheckWriter(BaseDataWriter):

    def __init__(self):
        super(MockCheckWriter, self).__init__(None, 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)

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

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

    raw = execution_info = skip
    close = skip




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

    def project_path(self):
        return self._project_path

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

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


class ApycotBaseTC(HGRCMixin, NarvalBaseTC):
    _repo_path = ()
    repo_import_revision_content = True

    @classproperty
    def test_db_id(cls):
        if cls._repo_path is None:
            return None
        ids = list(cls._repo_path)
        if not cls.repo_import_revision_content:
            ids.append('nocontent')
        return '-'.join(ids)

    @classmethod
    def pre_setup_database(cls, cnx, config):
        super(ApycotBaseTC, cls).pre_setup_database(cnx, config)
        setup_repos(*[join(cls.datadir, path) for path in cls._repo_path])
        pyp_tc = cnx.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')
        quickrecipe = cnx.find('Recipe', name='apycot.python.quick').one()
        setuprecipe = cnx.find('Recipe', name='apycot.setup.distutils').one()
        tc = cls.add_test_config(cnx, u'tc_quick', group=pyp_tc,
                                 use_recipe=quickrecipe)
        for path in cls._repo_path:
            repo = cnx.create_entity(
                'Repository', type=u'mercurial',
                source_url=u'file://' + join(cls.datadir, path), title=path,
                import_revision_content=True)
            pe = cnx.create_entity(
                'ProjectEnvironment', name=u'pe_%s'%path,
                check_config=u'env-option=value',
                check_environment=u'SETUPTOOLS=1\nDISPLAY=:2.0',
                setup_recipe=setuprecipe,
                local_repository=repo)
            cls.add_test_config(cnx, u'tc_%s'%path, env=pe,
                                group=pyp_tc, use_recipe=quickrecipe)
        cnx.commit()

    def setUp(self):
        setup_repos(*[join(self.datadir, path) for path in self._repo_path])
        super(ApycotBaseTC, self).setUp()
        lcache = repo_cache_dir(self.vreg.config)
        shutil.rmtree(lcache, ignore_errors=True)
        self.refresh()

    def tearDown(self):
        super(ApycotBaseTC, self).tearDown()
        for path in self._repo_path:
            shutil.rmtree(join(self.datadir, path), ignore_errors=True)

    def setup_database(self):
        """ self.repo: used to get the session to connect to cw
            self.vcsrepo: new entity
        """
        self.repo.threaded_task = lambda func: func() # XXX move to cw

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

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

    def hgrepo(self, reponame, local=True):
        with self.admin_access.repo_cnx() as cnx:
            repo = cnx.find('Repository', title=reponame).one()
            if local:
                repopath = repo.localcachepath
            else:
                repopath = repo.source_url
        return hgopen(repopath)

    def refresh(self):
        with self.repo.internal_cnx() as cnx:
            bridge.import_content(cnx, commitevery=1, raise_on_error=True)