test/unittest_hooks.py
author David Douard <david.douard@logilab.fr>
Thu, 13 Nov 2014 14:55:11 +0100
changeset 1752 31c34cf9f19f
parent 1740 f3137466457a
child 1701 7419bf11f329
permissions -rw-r--r--
[testutils] small improvements

#!/usr/bin/python
import os
import re
import shutil

from logilab.common.testlib import unittest_main

from cubicweb.devtools.testlib import MAILBOX
from cubicweb import Binary

from cubes.vcsfile import bridge
from cubes.apycot.hooks import start_period_tests
from cubes.apycot.testutils import ApycotBaseTC

os.environ['HGRCPATH'] = os.devnull

def clean_str(string):
    url_filtered = re.sub('lgc/[0-9]*', 'lgc/<EID>', string.strip())
    return re.sub('[0-9]', 'X', url_filtered)


class NotificationTC(ApycotBaseTC):

    def setup_database(self):
        super(NotificationTC, self).setup_database()
        with self.admin_access.repo_cnx() as cnx:
            cnx.create_entity('EmailAddress',
                              address=u'admin@cubicweb.org',
                              reverse_use_email=cnx.user)
            cnx.commit()

    def start_lgc_tests(self):
        with self.admin_access.client_cnx() as cnx:
            lgc = cnx.entity_from_eid(self.lgc)
            lgce = cnx.entity_from_eid(self.lgce)
            plan = lgc.start(lgce)
            cnx.commit()
            return plan.eid

    def test_exec_status_change(self):
        planeid = self.start_lgc_tests()
        with self.new_access('narval').client_cnx() as cnx:
            plan = cnx.entity_from_eid(planeid)
            self.dumb_execution(cnx, plan, [('unittest', 'success'), ('coverage', 'success')])
            plan.cw_adapt_to('IWorkflowable').fire_transition('start')
            cnx.commit()
            plan.cw_adapt_to('IWorkflowable').fire_transition('end')
            cnx.commit()
            self.assertEqual(MAILBOX, [])
            self.assertEqual(len(MAILBOX), 0)
        planeid = self.start_lgc_tests()
        with self.new_access('narval').client_cnx() as cnx:
            plan = cnx.entity_from_eid(planeid)
            self.dumb_execution(cnx, plan, [('unittest', 'success'), ('coverage', 'success')])
            plan.cw_adapt_to('IWorkflowable').fire_transition('start')
            cnx.commit()
            plan.cw_adapt_to('IWorkflowable').fire_transition('end')
            cnx.commit()
            self.assertEqual(len(MAILBOX), 0)
            cnx.commit()
            self.assertEqual(len(MAILBOX), 0)
        planeid = self.start_lgc_tests()
        with self.new_access('narval').client_cnx() as cnx:
            plan = cnx.entity_from_eid(planeid)
            self.dumb_execution(cnx, plan, [('unittest', 'failure'), ('coverage', 'failure')])
            self.assertEqual(len(MAILBOX), 0, MAILBOX)
            cnx.commit()
            plan.cw_adapt_to('IWorkflowable').fire_transition('start')
            cnx.commit()
            plan.cw_adapt_to('IWorkflowable').fire_transition('end')
            cnx.commit()
            self.assertEqual(len(MAILBOX), 1)
            self.assertEqual(MAILBOX[0].recipients, ['admin@cubicweb.org'])
            self.assertEqual(MAILBOX[0].message.get('Subject'),
                             '[data] lgce/lgc#default now has 2 failure')
            self.assertMultiLineEqual(clean_str(MAILBOX[0].message.get_payload(decode=True)),
                                      '''The following changes occured between executions on branch default:

* coverage status changed from success to failure
* unittest status changed from success to failure

Imported changes occured between XXXX/XX/XX XX:XX and XXXX/XX/XX XX:XX:
* no change found in known repositories

URL: http://XXX.X.X.X:XXXX/testexecution/XXXX''')


    def test_exec_one_status_change(self):
        planeid = self.start_lgc_tests()
        with self.new_access('narval').client_cnx() as cnx:
            plan = cnx.entity_from_eid(planeid)
            self.dumb_execution(cnx, plan, [('unittest', 'success'), ('coverage', 'success')])
            cnx.commit()
            plan.cw_adapt_to('IWorkflowable').fire_transition('start')
            cnx.commit()
            plan.cw_adapt_to('IWorkflowable').fire_transition('end')
            cnx.commit()
        planeid = self.start_lgc_tests()
        with self.new_access('narval').client_cnx() as cnx:
            plan = cnx.entity_from_eid(planeid)
            self.dumb_execution(cnx, plan, [('unittest', 'failure')])
            cnx.commit()
            plan.cw_adapt_to('IWorkflowable').fire_transition('start')
            cnx.commit()
            plan.cw_adapt_to('IWorkflowable').fire_transition('end')
            cnx.commit()
            self.assertEqual(len(MAILBOX), 1)
            self.assertEqual(MAILBOX[0].recipients, ['admin@cubicweb.org'])
            self.assertEqual(MAILBOX[0].message.get('Subject'),
                             '[data] lgce/lgc#default: success -> failure (unittest)')
            self.assertMultiLineEqual(clean_str(MAILBOX[0].message.get_payload(decode=True)),
                                      '''The following changes occured between executions on branch default:

* unittest status changed from success to failure

Imported changes occured between XXXX/XX/XX XX:XX and XXXX/XX/XX XX:XX:
* no change found in known repositories

URL: http://XXX.X.X.X:XXXX/testexecution/XXXX''')

    def test_nosy_list_propagation(self):
        planeid = self.start_lgc_tests()
        with self.admin_access.client_cnx() as cnx:
            cnx.execute('SET X interested_in PE WHERE X login "anon", PE eid %(pe)s',
                         {'pe': self.lgce})
            cnx.commit()
            # do not propagate to existing test execution
            plan = cnx.entity_from_eid(planeid)
            self.assertNotIn('anon', [u.login for u in plan.nosy_list])


class StartTestTC(ApycotBaseTC):
    _repo_path = (u'project1',)
    
    def setUp(self):
        super(StartTestTC, self).setUp()
        r = self.repo
        with self.admin_access.repo_cnx() as cnx:
            vcsrepo = cnx.find('Repository').one()
            path = vcsrepo.localcachepath
        return
    
        os.system('hg init %s' % r.path)
        os.system('echo data > %s/tutu.png' % r.path)
        os.system('hg -R %s add %s/tutu.png' % (r.path, r.path))
        os.system('hg -R %s commit -m rien -u narval' % r.path)
        # reset vcsrepo
        with r.internal_cnx() as cnx:
            bridge.import_content(cnx, commitevery=1, raise_on_error=True)


    def get_tc(self, cnx, period=None):
        rql = ('Any PEN, TCN, TCS  WHERE TC in_state S, '
               'TC use_environment PE, '
               'PE name PEN, '
               'S name "activated", '
               'TC name TCN, '
               'TC start_rev_deps TCS')
        if period:
            rql +=', TC computed_start_mode "%s"' % period
        return set(tuple(tc) for tc in cnx.execute(rql))

    def get_te_for_tc(self, cnx, eid):
        rql = 'Any TE where TE using_config TC, TC eid %(eid)s'
        return cnx.execute(rql, {'eid': eid})

    def launch_all_tests(self, period):
        """ launch all tests in the correct state and and make sure no
        TestExecutions are left in status waiting execution. Also fill-in the
        revision which should normally be filled by StartTestOp.
        """

        with self.admin_access.client_cnx() as cnx:
            start_period_tests(cnx, period)
            cnx.commit() # put new te in base

        with self.new_access('narval').client_cnx() as cnx:
            ## make sure they are not waiting execution and REV is filled
            for (te, rev) in cnx.execute('DISTINCT Any TE, REV WHERE TE in_state ST, '
                                         'ST name "ready", '
                                         'TE using_config TC, '
                                         'TE branch BR, '
                                         'TC use_environment PE, '
                                         'PE local_repository R, '
                                         'REV from_repository R, '
                                         'REV branch BR, '
                                         'NOT REV parent_revision REV2'):
                te_e = cnx.entity_from_eid(te)
                te_e.cw_adapt_to('IWorkflowable').fire_transition('start')
                rev_e = cnx.entity_from_eid(rev)
                # force status and revision since we do not have a narvabot eating
                # our TestExecutions here
                te_e.cw_set(status=u'running')
                te_e.cw_set(using_revision=rev)
            cnx.commit() # change new te status

    def test_new_vc_trigger(self):
        """ check the on new revision start mode. Run all testconfigs, add new
        revision, re-run and check there are new test configs"""
        with self.admin_access.client_cnx() as cnx:
            pe = cnx.find('ProjectEnvironment').one()
            tc = cnx.find('TestConfig', name='tc_project1').one()
            recipe = cnx.find('Recipe', name='apycot.quick')
            tc.cw_set(start_mode=u'on new revision')
            tc2 = self.add_test_config(cnx, u'tc2',
                                       start_mode=u'manual',
                                       env=pe,
                                       use_recipe=recipe)
            ## same repo, branch stable
            tc3 = self.add_test_config(cnx, u'tc3',
                                        start_mode=u'on new revision',
                                        check_config=u'branch=stable',
                                        env=self.lgce,
                                        use_recipe=self.recipe)
            ## same repo, branch default
            lgc4 = self.add_test_config(cnx, u'lgc4',
                                        start_mode=u'on new revision',
                                        check_config=u'branch=default',
                                        env=self.lgce,
                                        use_recipe=self.recipe)
            ## on different repo
            lgce2 = cnx.create_entity(
                'ProjectEnvironment', name=u'lgce2',
                check_config=u'install=setup_install',
                vcs_path=u'dir1', #dummy
                )
            lgc5 = self.add_test_config(cnx, u'lgc5',
                                        start_mode=u'on new revision',
                                        check_config=u'branch=default',
                                        env=lgce2,
                                        start_rev_deps=True,
                                        use_recipe=self.recipe)
            ## not sure this is still useful
            cnx.commit()
            ## not reaally necessary now, as they all appear, helps understanding,
            ## we check all the necessary testconfig are there
            self.assertEqual(self.get_tc(cnx, period=u'on new revision'),
                             set([('lgce', 'lgc', None),
                                  ('lgce', 'lgc3', None),
                                  ('lgce', 'lgc4', None),
                                  ('lgce2', 'lgc5', 1)]))
        ## create TestExecution entities
        self.launch_all_tests('on new revision')

        with self.admin_access.client_cnx() as cnx:
            nb_tc_lgc_before = len(self.get_te_for_tc(cnx, self.lgc))
            nb_tc_lgc3_before = len(self.get_te_for_tc(cnx, lgc3.eid))
            nb_tc_lgc4_before = len(self.get_te_for_tc(cnx, lgc4.eid))
            nb_tc_lgc5_before = len(self.get_te_for_tc(cnx, lgc5.eid))
            ## check you do not add anything on rerun (no new revision)

        self.launch_all_tests('on new revision')
        with self.admin_access.client_cnx() as cnx:
            self.assertEqual(nb_tc_lgc_before,
                             len(self.get_te_for_tc(cnx, self.lgc)))
            self.assertEqual(nb_tc_lgc3_before,
                             len(self.get_te_for_tc(cnx, lgc3.eid)))
            self.assertEqual(nb_tc_lgc4_before,
                             len(self.get_te_for_tc(cnx, lgc4.eid)))
            self.assertEqual(nb_tc_lgc5_before,
                             len(self.get_te_for_tc(cnx, lgc5.eid)))
        r = self.repo
        os.system('echo data > %s/tutu1.png' % r.path)
        os.system('hg -R %s branch stable' % r.path)
        os.system('hg -R %s add %s/tutu1.png' % (r.path, r.path))
        os.system('hg -R %s commit -m rien -u narval' % r.path)
        with r.internal_cnx() as cnx:
            bridge.import_content(cnx, commitevery=1, raise_on_error=True)
            cnx.commit()

        self.launch_all_tests('on new revision')
        with self.admin_access.client_cnx() as cnx:
            ## now it should add te for the tc linked to lgce and running on new
            ## revision and not the others
            self.assertEqual(nb_tc_lgc_before + 1,
                             len(self.get_te_for_tc(cnx, self.lgc)))
            self.assertEqual(nb_tc_lgc3_before + 1,
                             len(self.get_te_for_tc(cnx, lgc3.eid)))
            self.assertEqual(nb_tc_lgc4_before,
                              len(self.get_te_for_tc(cnx, lgc4.eid)))
            self.assertEqual(nb_tc_lgc5_before,
                              len(self.get_te_for_tc(cnx, lgc5.eid)))

        self.assertEqual(0, os.system('echo data > %s/tutu2.png' % r.path))
        self.assertEqual(0, os.system('hg -R %s up default' % r.path))
        self.assertEqual(0, os.system('hg -R %s add %s/tutu2.png' % (r.path, r.path)))
        self.assertEqual(0, os.system('hg -R %s commit -m rien -u narval' % r.path))
        with r.internal_cnx() as cnx:
            bridge.import_content(cnx, commitevery=1, raise_on_error=True)
            cnx.commit()

        ## it should add te to the tc on default (lgc and lgc4)
        self.launch_all_tests('on new revision')
        with self.admin_access.client_cnx() as cnx:
            self.assertEqual(nb_tc_lgc_before + 2,
                             len(self.get_te_for_tc(cnx, self.lgc)))
            self.assertEqual(nb_tc_lgc3_before + 1,
                             len(self.get_te_for_tc(cnx, lgc3.eid)))
            self.assertEqual(nb_tc_lgc4_before + 1,
                             len(self.get_te_for_tc(cnx, lgc4.eid)))
            self.assertEqual(nb_tc_lgc5_before,
                             len(self.get_te_for_tc(cnx, lgc5.eid)))

    def test_datetime_trigger(self):
        """test the registering of TC depending on their start_mode

        - check they are added correctly when created
        - check they are removed when deactivated
        - check they are removed when finished
        """
        with self.admin_access.client_cnx() as cnx:
            lgc = cnx.entity_from_eid(self.lgc)
            lgc.cw_set(start_mode=u'hourly')
            lgc2 = self.add_test_config(cnx, u'lgc2', start_mode=u'hourly', env=self.lgce)
            lgce2 = cnx.create_entity(
                'ProjectEnvironment', name=u'lgce2',
                check_config=u'install=setup_install',
                vcs_path=u'dir1',
                )
            lgc3 = self.add_test_config(cnx, u'lgc3', start_mode=u'hourly',
                                        check_config=u'branch=default',
                                        env=lgce2, start_rev_deps=True,
                                        use_recipe=self.recipe)
            cnx.execute('SET PE local_repository R WHERE PE name "lgce2", R eid %(r)s',
                         {'r': self.vcsrepo})
            cnx.commit()
            self.assertEqual(self.get_tc(cnx, 'daily'),
                              set([]))
            self.assertEqual(self.get_tc(cnx, 'hourly'),
                              set((('lgce', 'lgc', None),
                                   ('lgce', 'lgc2', None),
                                   ('lgce2', 'lgc3', 1)))
                              )
            lgc2.cw_adapt_to('IWorkflowable').fire_transition('deactivate')
            cnx.commit()
            self.assertEqual(self.get_tc(cnx, 'hourly'),
                              set((('lgce', 'lgc', None),
                                   ('lgce2', 'lgc3', 1)))
                              )
            cnx.commit()
            rset_before_lgc = self.get_te_for_tc(cnx, self.lgc)
            te = lgc3.start(lgce2, branch=u'default')
            cnx.commit()
            ###not really useful as the method directly check the existence of a
            ###relevant tc before running the tests,
            ### we should instead test the launched test configs
            self.assertEqual(self.get_tc(cnx, 'hourly'),
                              set((('lgce', 'lgc', None),
                                   ('lgce2', 'lgc3', 1)))
                              )
            rset_before_lgc = self.get_te_for_tc(cnx, self.lgc)
            rset_before_lgc3 = self.get_te_for_tc(cnx, lgc3.eid)
        self.launch_all_tests('hourly')
        with self.admin_access.client_cnx() as cnx:
            rset_after_lgc = self.get_te_for_tc(cnx, self.lgc)
            rset_after_lgc3 = self.get_te_for_tc(cnx, lgc3.eid)
            ### check there is one new TestConfig for lgc and the same number for lgc3
            #(alredy launched)
            self.assertEqual(len(rset_after_lgc3), len(rset_before_lgc3))
            self.assertEqual(len(rset_after_lgc), 1 + len(rset_before_lgc))

    def test_error_in_recipe(self):
        """this testcase should check what happens when a recipe does not have
        the correct code and crashes before sending any status, the returned
        status should NOT be waiting execution"""
        with self.admin_access.client_cnx() as cnx:
            lgc = cnx.entity_from_eid(self.lgc)
            lgce = cnx.entity_from_eid(self.lgce)
            te = lgc.start(lgce, branch=u'default')
            cnx.commit()
            te.cw_adapt_to('IWorkflowable').fire_transition('start')
            cnx.commit()
            te.cw_adapt_to('IWorkflowable').fire_transition('fail')
            cnx.commit()
            #the status should not be waiting execution !
            self.assertNotEqual(u'waiting execution', te.status)


class ComputedStartModeTC(ApycotBaseTC):

    def test_start_mode(self):
        with self.admin_access.client_cnx() as cnx:
            pyp = cnx.entity_from_eid(self.pyp)
            lgc = cnx.entity_from_eid(self.lgc)
            pyp.cw_set(start_mode=u'monthly')
            cnx.commit()
            self.assertEqual(pyp.computed_start_mode, pyp.start_mode)
            lgc.cw_set(start_mode=u'inherited')
            cnx.commit()
            lgc.cw_clear_all_caches()
            self.assertEqual(lgc.computed_start_mode, pyp.start_mode)
            pyp.cw_set(start_mode=u'hourly')
            cnx.commit()
            pyp.cw_clear_all_caches()
            lgc.cw_clear_all_caches()
            self.assertEqual(pyp.computed_start_mode, u'hourly')
            self.assertEqual(lgc.computed_start_mode, u'hourly')

class ReposirotyRecipeTC(ApycotBaseTC):

    def test_reposiroty_has_recipe(self):
        with self.admin_access.client_cnx() as cnx:
            repo = cnx.create_entity('Repository', type=u'mercurial',
                                     source_url=u'file://path/to/repo',)
            cnx.commit()
            self.assertEqual('apycot.checkout.mercurial', repo.checkout_recipe[0].name)

if __name__ == '__main__':
    unittest_main()