author Sylvain Thénault <sylvain.thenault@logilab.fr>
Wed, 09 Nov 2011 18:04:25 +0100
changeset 791 5ac76ce656dd
parent 783 caf114cc9c86
child 837 69daedbf31c8
permissions -rw-r--r--
update to cw 3.14 api. Closes #2063529

import shutil
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.apycot import entities

from apycotlib.writer import CheckDataWriter, BaseDataWriter

INPUTS_DIR = abspath(join(dirname(__file__), 'data'))

def input_path(file=''):
    return join(INPUTS_DIR, file)

class DummyStack(object):

    def __init__(self):
        self.msg = None

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

    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

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

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

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

class MockRepository:
    """fake apycot.IRepository class"""
    branch = None
    def __init__(self, attrs=None, **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):

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

class ApycotBaseTC(CubicWebTC):

    def setup_database(self):
        req = self.request()
        self.lgce = req.create_entity(
            'ProjectEnvironment', name=u'lgce',
        self.vcsrepo = req.create_entity('Repository', type=u'mercurial',
                                         # use path to avoid clone attempt when using url
        self.pyp = req.create_entity('TestConfig', name=u'PYTHONPACKAGE',
        self.lgc = self.add_test_config(u'lgc', env=self.lgce, group=self.pyp)
        self.recipe = req.create_entity('Recipe', name=u'functest.noop',
        step1 = self.recipe.add_step(u'action', u'basic.noop', initial=True)
        step2 = self.recipe.add_step(u'action', u'basic.noop', final=True)
        tr = self.recipe.add_transition(step1, step2)

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

    def add_test_config(self, name,
                        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 starttime %(et)s, X endtime %(et)s, X status "success" '
                        'WHERE X eid %(x)s', {'et': datetime.now(), 'x': ex.eid})