test/unittest_hooks.py
author Julien Cristau <julien.cristau@logilab.fr>
Tue, 23 Dec 2014 14:18:51 +0100
changeset 1675 533a3f8bbebb
parent 1674 7e66dbf04a09
child 1680 e128d10b0a4e
child 1824 82ad4a58ea49
permissions -rw-r--r--
[test] set up evolve extension so vcsfile import can do its thing Related to #4786982

#!/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.vcsfile.testutils import HGRCMixin
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://testing.fr/cubicweb/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://testing.fr/cubicweb/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(HGRCMixin, ApycotBaseTC):

    def setUp(self):
        super(StartTestTC, self).setUp()
        r = self.repo
        with self.admin_access.repo_cnx() as cnx:
            vcsrepo = cnx.entity_from_eid(self.vcsrepo)
            r.path = vcsrepo.localcachepath
        if os.path.exists(r.path):
            shutil.rmtree(r.path)
        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 tearDown(self):
        r = self.repo
        os.system('rm -rf %s/.hg' % r.path)
        os.system('rm -rf %s/*.png' % r.path)
        super(StartTestTC, self).tearDown()

    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_name(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, cnx, 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.
        """

        start_period_tests(cnx, period)
        cnx.commit() #put new te in base
        ## make sure they are not waiting execution and REV is filled
        for te, rev in cnx.execute('Any TE, REV WHERE TE status %(st)s, '
                                    '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',
                                    {'st': 'waiting execution'}):
            te_e = cnx.entity_from_eid(te)
            rev_e = cnx.entity_from_eid(rev)
            te_e.cw_adapt_to('IWorkflowable').fire_transition('start')
            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:
            lgc = cnx.entity_from_eid(self.lgc)
            lgc.cw_set(start_mode=u'on new revision')
            lgc2 = self.add_test_config(cnx, u'lgc2',
                                        start_mode=u'manual',
                                        env=self.lgce,
                                        use_recipe=self.recipe)
            ## same repo, branch stable
            lgc3 = self.add_test_config(cnx, u'lgc3',
                                        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)]))
            ## run everything 

            self.launch_all_tests(cnx, 'on new revision')

            nb_tc_lgc_before = len(self.get_te_for_tc_name(cnx, self.lgc))
            nb_tc_lgc3_before = len(self.get_te_for_tc_name(cnx, lgc3.eid))
            nb_tc_lgc4_before = len(self.get_te_for_tc_name(cnx, lgc4.eid))
            nb_tc_lgc5_before = len(self.get_te_for_tc_name(cnx, lgc5.eid))
            ##check you do not add anything on rerun (no new revision)
            self.launch_all_tests(cnx, 'on new revision')
            self.assertEqual(nb_tc_lgc_before,
                             len(self.get_te_for_tc_name(cnx, self.lgc)))
            self.assertEqual(nb_tc_lgc3_before,
                             len(self.get_te_for_tc_name(cnx, lgc3.eid)))
            self.assertEqual(nb_tc_lgc4_before,
                             len(self.get_te_for_tc_name(cnx, lgc4.eid)))
            self.assertEqual(nb_tc_lgc5_before,
                             len(self.get_te_for_tc_name(cnx, lgc5.eid)))
            ## add a new revision in stable
            #r.vcs_add(u'dir1', u'tutu.png', Binary('data'), branch=u'stable')
            cnx.commit()

        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)

        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.launch_all_tests(cnx, 'on new revision')
            self.assertEqual(nb_tc_lgc_before + 1,
                             len(self.get_te_for_tc_name(cnx, self.lgc)))
            self.assertEqual(nb_tc_lgc3_before + 1,
                             len(self.get_te_for_tc_name(cnx, lgc3.eid)))
            self.assertEqual(nb_tc_lgc4_before,
                              len(self.get_te_for_tc_name(cnx, lgc4.eid)))
            self.assertEqual(nb_tc_lgc5_before,
                              len(self.get_te_for_tc_name(cnx, lgc5.eid)))

            ## add a new revision in default
            #r.vcs_add(u'dir1', u'tutu1.png', Binary('data'))
            cnx.commit()
        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)
        ## it should add te to the tc on default (lgc and lgc4)
        with self.admin_access.client_cnx() as cnx:
            self.launch_all_tests(cnx, 'on new revision')
            self.assertEqual(nb_tc_lgc_before + 2,
                             len(self.get_te_for_tc_name(cnx, self.lgc)))
            self.assertEqual(nb_tc_lgc3_before + 1,
                             len(self.get_te_for_tc_name(cnx, lgc3.eid)))
            self.assertEqual(nb_tc_lgc4_before + 1,
                             len(self.get_te_for_tc_name(cnx, lgc4.eid)))
            self.assertEqual(nb_tc_lgc5_before,
                             len(self.get_te_for_tc_name(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_name(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_name(cnx, self.lgc)
            rset_before_lgc3 = self.get_te_for_tc_name(cnx, lgc3.eid)
            self.launch_all_tests(cnx, 'hourly')
            rset_after_lgc = self.get_te_for_tc_name(cnx, self.lgc)
            rset_after_lgc3 = self.get_te_for_tc_name(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(HGRCMixin, 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')

if __name__ == '__main__':
    unittest_main()