testutils.py
author David Douard <david.douard@logilab.fr>
Sun, 26 Oct 2014 22:42:02 +0100
changeset 1767 b99269e6e477
parent 1764 8473295c50b1
child 1714 ec6937082db6
permissions -rw-r--r--
[apycotlib] fix 'pyversions' compliant with virtualenv When executed in virtualenv, there is only one python "version" available...

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

    def __init__(self):
        super(MockBaseWriter, self).__init__(MockConnection('narval0'), 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 = {}

    @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 HTTP handler"""
    def __init__(self, instance_id):
        self.instance_id = instance_id
        self.instance_url = BASE_URL

    def http_get(self, url, **params):
        pass
    def http_post(self, url, **params):
        pass
    def pending_plans(self):
        return ()
    def plan(self, eid):
        pass

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.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)