test/unittest_task.py
author Sylvain Thénault <sylvain.thenault@logilab.fr>
Wed, 09 Nov 2011 18:04:25 +0100
branchstable
changeset 791 5ac76ce656dd
parent 727 f865240f4ea7
child 828 5d1b7e01c4bc
permissions -rw-r--r--
update to cw 3.14 api. Closes #2063529

import shutil
import tarfile
import sys
import os
from os.path import exists, join, abspath

from logilab.common.testlib import TestCase, unittest_main, mock_object, within_tempdir

# import this first will set import machinery on
from utils import MockTestWriter, MockRepository, MockConnection, MockVCSFile, input_path

from apycotlib import SetupException
from apycotlib import SUCCESS, FAILURE, PARTIAL, SKIPPED
from apycotlib.atest import Test as BaseTest
from apycotlib.repositories import SVNRepository


# manage temporary repo
def setUpModule():
    for repo in ('badpkg2_svn', 'goodpkg_svn'):
        path = input_path(repo)
        if exists(path):
            shutil.rmtree(path)
        tarfile.open(input_path('%s.tar.gz' % repo), 'r|gz').extractall(input_path(''))
    os.environ['APYCOT_ROOT'] = ''
    mockvcsfile = MockVCSFile('subversion', path='/home/cvs')
    global MOCKREPO, BADREPO, SVNREPO1, SVNREPO2, TPP
    MOCKREPO = MockRepository(repository=mockvcsfile,
                              path='soft/goodpkg',
                              command='cp -R %s .' % input_path('goodpkg'))
    BADREPO = MockRepository(repository=mockvcsfile,
                             path='soft/goodpkg', command='false')
    SVNREPO1 = SVNRepository({'repository': MockVCSFile('subversion',
                                                        source_url='file://%s' % input_path('goodpkg'))})
    SVNREPO2 = SVNRepository({'repository': MockVCSFile('subversion',
                                                        source_url='file://%s' % input_path('badpkg2'))})
    TPP = TouchTestPreprocessor()


def tearDownModule():
    del os.environ['APYCOT_ROOT']
    for repo in ('badpkg2', 'goodpkg'):
        path = input_path(repo)
        if exists(path):
            shutil.rmtree(path)

def Test(tconfig, *args, **kwargs):
    pps = kwargs.pop('preprocessors', None)
    checkers = kwargs.pop('checkers', None)
    repo = kwargs.pop('repo', MOCKREPO)
    environment = kwargs.pop('environment',
                             Environment(eid=tconfig.name, apycot_preprocessors={}))
    texec = mock_object(configuration=tconfig, environment=environment,
                        branch=environment.conf.get('branch'))
    test = BaseTest(texec, *args, **kwargs)
    if pps is not None:
        test.apycot_preprocessors = lambda x: pps
    if checkers is not None:
        test.checkers = checkers
    if repo is not None:
        test._repositories[environment.eid] = repo
    return test

class TestConfig:
    def __init__(self, name, dependencies=(), environ=None, conf=None):
        self.name = self.eid = name
        #self._repo = repo
        self._dependencies = dependencies
        self._environ = environ or {'ZIGUOUIGOUI': 'YOOOO'}
        self.all_checks = ()
        if conf is None:
            conf = {}
        self._configuration = conf

    def apycot_process_environment(self):
        return self._environ
    def apycot_configuration(self, pe):
        return self._configuration.copy()
    def dependencies(self):
        return self._dependencies

class Environment(object):
    def __init__(self, eid, apycot_preprocessors={},
                 repository=MockVCSFile('mercurial',
                                     source_url='http://bob.org/hg/toto/'),
                 conf=None):
        self.eid = eid
        self.apycot_preprocessors = apycot_preprocessors
        self.repository = repository
        if conf is None:
            conf = {}
        self.conf = conf
        self.name=''
        self.vcs_path = ''

    def apycot_configuration(self):
        return self.conf
    def apycot_process_environment(self):
        return {}

# mock objects ################################################################

class CleanRaisePreprocessor:
    id = 'clean_raise_preprocessor'
    def match(self, name):
        return 1

    def run(self, test, path=None):
        if path is None:
            return 1
        else:
            return 0

class SetupRaisePreprocessor:
    id = 'setup_raise_preprocessor'
    def match(self, name):
        return 1

    def run(self, test, path=None):
        if path is None:
            raise SetupException('in test_preprocessor.test_setup')
        else:
            raise SetupException('%s failed on %r' % (self.id, path))

class TouchTestPreprocessor:
    id = 'touch_preprocessor'
    file = None
    file2 = None

    def run(self, test, path=None):
        self.file = join(test.tmpdir, 'TestTC_pp')
        self.file2 = join(test.tmpdir, 'TestTC2_pp')
        f = open(self.file, 'w')
        f.close()
        f = open(self.file2, 'w')
        f.close()

class SimplePreprocessor(object):

    id = 'simple_preprocessor'
    def __init__(self):
        self.processed    = {}

    def run(self, test, path=None):
        if path == None:
            path = test.project_path()
        self.processed.setdefault(path, 0)
        self.processed[path] += 1

class DummyTest(object):
    need_preprocessor = None

    def check(self, test, writer):
        return SUCCESS
    def check_options(self):
        pass

class SuccessTestChecker(DummyTest):
    id = 'success_test_checker'
    options = {}
    need_preprocessor = 'install'
    def check(self, test, writer):
        return SUCCESS

class FailureTestChecker(DummyTest):
    id = 'failure_test_checker'
    options = {}
    def check(self, test, writer):
        return FAILURE

class ErrorTestChecker(DummyTest):
    id = 'error_test_checker'
    options = {}
    def check(self, test, writer):
        raise Exception('never succeed!')


# real tests ##################################################################

class TestTC(TestCase):

    @within_tempdir
    def test_setup_installed(self):
        self.skipTest('to be done by pyves')
        pp = SimplePreprocessor()
        test = Test(TestConfig('yo', dependencies=(Environment('pypasax'),)),
                    MockTestWriter(), {},
                    checkers=[SuccessTestChecker()],
                    preprocessors={'install': pp})
        test._repositories['pypasax'] = SVNREPO2
        test.setup()
        self.assertEqual(pp.processed, {'soft/goodpkg': 1, 'badpkg2': 1})


    def test_setup_no_install(self):
        self.skipTest('to be done by pyves')
        test = Test(TestConfig('yo', dependencies=(Environment('pypasax'),)),
                    MockTestWriter(), {},
                    preprocessors={'install': TPP})
        test._repositories['pypasax'] = SVNREPO2
        # no checks requiring installation, main repo should be checked out though not installed,
        # and dependencies shouldn't be installed
        try:
            test.setup()
            self.assertEqual(os.environ['ZIGUOUIGOUI'], 'YOOOO')
            self.failUnless(exists('goodpkg'))
            self.failIf(exists('badpkg2'))
            self.failUnless(TPP.file is None or not exists(TPP.file))
        finally:
            os.environ.pop('ZIGUOUIGOUI', None)
            if exists('goodpkg'):
                shutil.rmtree('goodpkg')

    def test_python_setup(self):
        self.skipTest('to be done by pyves')
        test = Test(TestConfig('yo', dependencies=(Environment('pypasax'),)),
                    MockTestWriter(), {},
                    checkers=[SuccessTestChecker()],
                    preprocessors={'install': TPP})
        test._repositories['pypasax'] = SVNREPO2
        try:
            test.setup()
            tconfig = test.tconfig
            projectenvs = (tconfig.environment,) + tconfig.dependencies()
            for pe in projectenvs:
                test.checkout(pe)
            self.assertEqual(os.environ['ZIGUOUIGOUI'], 'YOOOO')
            self.failUnless(exists('goodpkg'))
            self.failUnless(exists('badpkg2'))
            self.failUnless(exists(TPP.file))
        finally:
            if exists('goodpkg'):
                shutil.rmtree('goodpkg')
            if exists('badpkg2'):
                shutil.rmtree('badpkg2')
            if TPP.file is not None and exists(TPP.file):
                os.remove(TPP.file)
            if TPP.file2 is not None and exists(TPP.file2):
                os.remove(TPP.file2)
            del os.environ['ZIGUOUIGOUI']

    @within_tempdir
    def _test_setup_ex(self, test, msg=None):
        self.skipTest('to be done by pyves')
        try:
            test.setup()
            self.failUnless(test._failed_pp, 'Preprocessors should have failed')
        except SetupException, ex:
            if msg:
                self.assertEqual(str(ex), msg)

    def test_setup_raise(self):
        self.skipTest('to be done by pyves')
        # test bad checkout command
        test = Test(TestConfig('yo'), MockTestWriter(), {}, repo=BADREPO)
        self._test_setup_ex(test, "`false` returned with status : 1")
        # test bad dependencies checkout
        test = Test(TestConfig('yo', dependencies=(Environment('toto'),),
                               ),
                    MockTestWriter(), {},
                    checkers=[SuccessTestChecker()])
        test._repositories['toto'] = BADREPO
        self._test_setup_ex(test)
        # test bad preprocessing
        test = Test(TestConfig('yo', dependencies=(Environment('pypasax'),),
                               ),
                    MockTestWriter(), {},
                    checkers=[SuccessTestChecker()],
                    preprocessors={'install': SetupRaisePreprocessor()})
        test._repositories['pypasax'] = SVNREPO2
        self._test_setup_ex(test)

    def test_clean(self):
        self.skipTest("We don't execute whole test anymore")
        test = Test(TestConfig('yo', dependencies=(Environment('pypasax'),),
                               ),
                    MockTestWriter(), {},
                    checkers=[SuccessTestChecker()],
                    preprocessors={'install': TPP})
        test._repositories['pypasax'] = SVNREPO2
        # clean should never fail
        # but most interesting things occurs after setup...
        test.execute()
        self.assertNotNone(TPP.file, "Preprocessors have not been run")
        self.failIf(exists('goodpkg'))
        self.failIf(exists('badpkg2'))
        self.failIf(exists(TPP.file))

    def test_execute_1(self):
        self.skipTest("We don't execute whole test anymore")
        test = Test(TestConfig('yo', dependencies=(Environment('Pypasax'),),
                               ),
                    MockTestWriter(), {}, repo=SVNREPO1,
                    checkers=[SuccessTestChecker(), FailureTestChecker(), ErrorTestChecker()],
                    preprocessors={'install': TPP})
        test._repositories['Pypasax'] = SVNREPO2
        test.execute()
        self.assertNotNone(TPP.file)
        self.failIf(exists(TPP.file))
        self.assertNotNone(TPP.file2)
        self.failIf(exists(TPP.file2))

    def test_execute_2(self):
        self.skipTest("We don't execute whole test anymore")
        test = Test(TestConfig('yo', dependencies=(Environment('Pypasax', SVNREPO2),),
                                     ),
                    MockTestWriter(), {}, repo=SVNREPO1,
                    checkers=[SuccessTestChecker(), FailureTestChecker(), ErrorTestChecker()],
                    preprocessors={'install:': SetupRaisePreprocessor()})
        test._repositories['Pypasax'] = SVNREPO2
        test.execute()

    def test_execute_0(self):
        self.skipTest("We don't execute whole test anymore")
        command = 'cp -R '+abspath('inputs/goodpkg')+' .'
        cwd = os.getcwd()
        test = Test(TestConfig('yo', dependencies=(Environment('Pypasax'),)),
                    MockTestWriter(), {})
        test._repositories['Pypasax'] = SVNREPO2
        self.failUnless(exists(test.tmpdir))
        test.execute()
        self.failIf(exists(test.tmpdir))
        self.assertEqual(os.getcwd(), cwd)

        test = Test(TestConfig('yo', dependencies=(Environment('Pypasax'),)),
                    MockTestWriter(), {'keep-test-dir':1},)
        test._repositories['Pypasax'] = SVNREPO2
        self.failUnless(exists(test.tmpdir))
        test.execute()
        self.failIf(exists(test.tmpdir))
        self.assertEqual(os.getcwd(), cwd)

        test = Test(TestConfig('yo', dependencies=(Environment('Pypasax'),),
                                     ),
                    MockTestWriter()(), {}, repo=SVNREPO1,
                    checkers=[SuccessTestChecker()],
                    preprocessors={'install': SetupRaisePreprocessor()})
        test._repositories['Pypasax'] = SVNREPO2
        self.failUnless(exists(test.tmpdir))
        test.execute()
        self.failIf(exists(test.tmpdir))
        self.assertEqual(os.getcwd(), cwd)

        test = Test(TestConfig('yo', dependencies=(Environment('Pypasax'),),
                                     ),
                    MockTestWriter()(), {}, repo=SVNREPO1,
                    checkers=[SuccessTestChecker()],
                    preprocessors={'install': CleanRaisePreprocessor()})
        test._repositories['Pypasax'] = SVNREPO2
        self.failUnless(exists(test.tmpdir))
        test.execute()
        self.failIf(exists(test.tmpdir))
        self.assertEqual(os.getcwd(), cwd)

    def test_branch(self):
        test = Test(TestConfig('yo'),
                    MockTestWriter(), {},
                    environment=Environment('babar', conf={'branch': 'bob'}),
                    checkers=[SuccessTestChecker()])
        del test._repositories['babar'] # XXX clean up this mess
        repo = test.apycot_repository()
        self.assertEqual(repo.branch, 'bob')

    def test_branch_deps_with_branch(self):
        dep = Environment('babar', conf={'branch': 'Onk'})
        test = Test(TestConfig('yo', dependencies=(dep, )),
                    MockTestWriter(), {},
                    checkers=[SuccessTestChecker()])
        repo = test.apycot_repository(dep)
        self.assertEqual(repo.branch, 'Onk')

    def test_branch_deps_without_branch(self):
        dep = Environment('babar')
        test = Test(TestConfig('yo', dependencies=(dep, )),
                    MockTestWriter(), {},
                    checkers=[SuccessTestChecker()])
        repo = test.apycot_repository(dep)
        # default should be the branch name (as none is defined)
        self.assertEqual(repo.branch, 'default')


class EnvironmentTrackerMixinTC(TestCase):

    def setUp(self):
        os.environ['PYTHONPATH'] = ''
        self.tracker = Test(TestConfig('yo', dependencies=(Environment('pypasax'),)),
                            MockTestWriter(), {})

    def test_update_clean_env(self):
        lc_all = os.environ.get('LC_ALL')
        self.tracker.update_env('key', 'LC_ALL', 'XXXX')
        self.assertEqual(os.environ['LC_ALL'], 'XXXX')
        self.tracker.clean_env('key', 'LC_ALL')
        self.assertEqual(os.environ.get('LC_ALL'), lc_all)

        self.tracker.update_env('key', '__ENVIRONMENTTRACKERMIXINTC__', 'XXXX')
        self.assertEqual(os.environ['__ENVIRONMENTTRACKERMIXINTC__'], 'XXXX')
        self.tracker.clean_env('key', '__ENVIRONMENTTRACKERMIXINTC__')
        self.assertRaises(KeyError, os.environ.__getitem__,
                          '__ENVIRONMENTTRACKERMIXINTC__')

    def test_nested(self):
        lc_all = os.environ.get('LC_ALL')
        self.tracker.update_env('key', 'LC_ALL', 'XXXX')
        self.assertEqual(os.environ['LC_ALL'], 'XXXX')
        self.tracker.update_env('key2', 'LC_ALL', 'YYYY')
        self.assertEqual(os.environ['LC_ALL'], 'YYYY')
        self.tracker.clean_env('key2', 'LC_ALL')
        self.assertEqual(os.environ['LC_ALL'], 'XXXX')
        self.tracker.clean_env('key', 'LC_ALL')
        self.assertEqual(os.environ.get('LC_ALL'), lc_all)

    def test_update_clean_env_sep(self):
        path = os.environ['PATH']
        self.tracker.update_env('key', 'PATH', '/mybin', ':')
        self.assertEqual(os.environ['PATH'], '/mybin:' + path)
        self.tracker.clean_env('key', 'PATH')
        self.assertEqual(os.environ['PATH'], path)

    def test_nested_sep(self):
        path = os.environ['PATH']
        self.tracker.update_env('key', 'PATH', '/mybin', ':')
        if path:
            self.assertEqual(os.environ['PATH'], '/mybin:' + path)
        else:
            self.assertEqual(os.environ['PATH'], '/mybin')
        self.tracker.update_env('key2', 'PATH', '/myotherbin', ':')
        if path:
            self.assertEqual(os.environ['PATH'], '/myotherbin:/mybin:' + path)
        else:
            self.assertEqual(os.environ['PATH'], '/myotherbin:/mybin')
        self.tracker.clean_env('key2', 'PATH')
        if path:
            self.assertEqual(os.environ['PATH'], '/mybin:' + path)
        else:
            self.assertEqual(os.environ['PATH'], '/mybin')
        self.tracker.clean_env('key', 'PATH')
        self.assertEqual(os.environ['PATH'], path)

    def test_python_path_sync(self):
        self.tracker.update_env('key', 'PYTHONPATH', '/mylib', ':')
        self.assertEqual(os.environ['PYTHONPATH'], '/mylib')
        self.assertEqual(sys.path[0], '/mylib')
        self.tracker.update_env('key2', 'PYTHONPATH', '/otherlib', ':')
        self.assertEqual(os.environ['PYTHONPATH'], '/otherlib:/mylib')
        self.assertEqual(sys.path[0], '/otherlib')
        self.tracker.clean_env('key2', 'PYTHONPATH')
        self.assertEqual(os.environ['PYTHONPATH'], '/mylib')
        self.assertNotEqual(sys.path[0], '/otherlib')
        self.tracker.clean_env('key', 'PYTHONPATH')
        self.assertEqual(os.environ['PYTHONPATH'], '')
        self.assertNotEqual(sys.path[0], '/otherlib')

    def test_update_undefined_env(self):

        var = 'XNZOUACONFVESUHFJGSLKJ'
        while os.environ.get(var) is not None:
            var = ''.join(chr(randint(ord('A'), ord('Z') +1))
                for cnt in xrange(randint(10, 20)))

        self.tracker.update_env('key', var, 'to be or not to be', ':')
        self.assertMultiLineEqual(os.environ.get(var),  'to be or not to be')
        self.tracker.clean_env('key', var)
        self.assertEqual(os.environ.get(var), None)



if __name__ == '__main__':
    unittest_main()