test/unittest_hooks.py
author Julien Cristau <julien.cristau@logilab.fr>
Thu, 24 Jul 2014 14:39:57 +0200
changeset 1561 ea7b1e630c1b
parent 1553 ba466ae29ba8
child 1563 000605df69cf
permissions -rw-r--r--
Silence cubicweb 3.16 deprecation warnings set_attributes/set_relations became cw_set.

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

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()
        self.request().create_entity('EmailAddress',
                                     address=u'admin@cubicweb.org',
                                     reverse_use_email=self.user())

    def start_lgc_tests(self):
        plan = self.lgc.start(self.lgce)
        self.lgc._cw.cnx.commit()
        return plan

    def test_exec_status_change(self):
        self.login('narval', password='narval0')
        plan = self.start_lgc_tests()
        plan = self.request().entity_from_eid(plan.eid)
        self.dumb_execution(plan, [('unittest', 'success'), ('coverage', 'success')])
        plan.cw_adapt_to('IWorkflowable').fire_transition('start')
        self.commit()
        plan.cw_adapt_to('IWorkflowable').fire_transition('end')
        self.commit()
        self.assertEqual(MAILBOX, [])
        self.commit()
        self.assertEqual(len(MAILBOX), 0)
        plan = self.start_lgc_tests()
        plan = self.request().entity_from_eid(plan.eid)
        self.dumb_execution(plan, [('unittest', 'success'), ('coverage', 'success')])
        plan.cw_adapt_to('IWorkflowable').fire_transition('start')
        self.commit()
        plan.cw_adapt_to('IWorkflowable').fire_transition('end')
        self.commit()
        self.assertEqual(len(MAILBOX), 0)
        self.commit()
        self.assertEqual(len(MAILBOX), 0)
        plan = self.start_lgc_tests()
        plan = self.request().entity_from_eid(plan.eid)
        self.dumb_execution(plan, [('unittest', 'failure'), ('coverage', 'failure')])
        self.assertEqual(len(MAILBOX), 0, MAILBOX)
        self.commit()
        plan.cw_adapt_to('IWorkflowable').fire_transition('start')
        self.commit()
        plan.cw_adapt_to('IWorkflowable').fire_transition('end')
        self.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):
        self.login('narval', password='narval0')
        plan = self.start_lgc_tests()
        plan = self.request().entity_from_eid(plan.eid)
        self.dumb_execution(plan, [('unittest', 'success'), ('coverage', 'success')])
        self.commit()
        plan.cw_adapt_to('IWorkflowable').fire_transition('start')
        self.commit()
        plan.cw_adapt_to('IWorkflowable').fire_transition('end')
        self.commit()
        plan = self.start_lgc_tests()
        plan = self.request().entity_from_eid(plan.eid)
        self.dumb_execution(plan, [('unittest', 'failure')])
        self.commit()
        plan.cw_adapt_to('IWorkflowable').fire_transition('start')
        self.commit()
        plan.cw_adapt_to('IWorkflowable').fire_transition('end')
        self.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):
        self.login('narval', password='narval0')
        plan = self.start_lgc_tests()
        self.commit()
        self.restore_connection()
        self.execute('SET X interested_in PE WHERE X login "anon", PE eid %(pe)s',
                     {'pe': self.lgce.eid})
        self.commit()
        # do not propagate to existing test execution
        self.assertFalse('anon' in [u.login for u in plan.nosy_list])


class StartTestTC(ApycotBaseTC):

    def setUp(self):
        super(StartTestTC, self).setUp()
        r = self.repo
        r.path = self.vcsrepo.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_session(safe=True) as session:
            bridge.import_content(session, 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, 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 self.execute(rql))

    def get_te_for_tc_name(self, eid):
        rql = 'Any TE where TE using_config TC, TC eid %(eid)s'
        return self.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.
        """

        session = self.session
        session.set_cnxset()
        start_period_tests(session, period)
        self.commit() #put new te in base
        ## make sure they are not waiting execution and REV is filled
        for te, rev in self.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 = self.request().entity_from_eid(te)
            rev_e = self.request().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)
        self.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"""
        self.lgc.cw_set(start_mode=u'on new revision')
        lgc2 = self.add_test_config(u'lgc2',
                                    start_mode=u'manual',
                                    env=self.lgce,
                                    use_recipe=self.recipe)
        ## same repo, branch stable
        lgc3 = self.add_test_config(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(u'lgc4',
                                    start_mode=u'on new revision',
                                    check_config=u'branch=default',
                                    env=self.lgce,
                                    use_recipe=self.recipe)
        ## on different repo
        req = self.request()
        lgce2 = req.create_entity(
            'ProjectEnvironment', name=u'lgce2',
            check_config=u'install=setup_install',
            vcs_path=u'dir1', #dummy
            )
        lgc5 = self.add_test_config(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
        self.commit()
        ## not reaally necessary now, as they all appear, helps understanding,
        ## we check all the necessary testconfig are there
        self.assertEqual(self.get_tc(period=u'on new revision'),
                         set([('lgce', 'lgc', None),
                              ('lgce', 'lgc3', None),
                              ('lgce', 'lgc4', None),
                              ('lgce2', 'lgc5', 1)]))
        ## run everything 

        self.launch_all_tests('on new revision')

        nb_tc_lgc_before = len(self.get_te_for_tc_name(self.lgc.eid))
        nb_tc_lgc3_before = len(self.get_te_for_tc_name(lgc3.eid))
        nb_tc_lgc4_before = len(self.get_te_for_tc_name(lgc4.eid))
        nb_tc_lgc5_before = len(self.get_te_for_tc_name(lgc5.eid))
        ##check you do not add anything on rerun (no new revision)
        self.launch_all_tests('on new revision')
        self.assertEqual(nb_tc_lgc_before,
                         len(self.get_te_for_tc_name(self.lgc.eid)))
        self.assertEqual(nb_tc_lgc3_before,
                         len(self.get_te_for_tc_name(lgc3.eid)))
        self.assertEqual(nb_tc_lgc4_before,
                         len(self.get_te_for_tc_name(lgc4.eid)))
        self.assertEqual(nb_tc_lgc5_before,
                         len(self.get_te_for_tc_name(lgc5.eid)))
        r = self.repo

        ## add a new revision in stable
        #r.vcs_add(u'dir1', u'tutu.png', Binary('data'), branch=u'stable')
        self.commit()
        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_session(safe=True) as session:
            bridge.import_content(session, commitevery=1, raise_on_error=True)

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

        ## add a new revision in default
        #r.vcs_add(u'dir1', u'tutu1.png', Binary('data'))
        self.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_session(safe=True) as session:
            bridge.import_content(session, commitevery=1, raise_on_error=True)
        ## it should add te to the tc on default (lgc and lgc4)
        self.launch_all_tests('on new revision')
        self.assertEqual(nb_tc_lgc_before + 2,
                         len(self.get_te_for_tc_name(self.lgc.eid)))
        self.assertEqual(nb_tc_lgc3_before + 1,
                         len(self.get_te_for_tc_name(lgc3.eid)))
        self.assertEqual(nb_tc_lgc4_before + 1,
                         len(self.get_te_for_tc_name(lgc4.eid)))
        self.assertEqual(nb_tc_lgc5_before,
                         len(self.get_te_for_tc_name(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
        """
        self.lgc.cw_set(start_mode=u'hourly')
        lgc2 = self.add_test_config(u'lgc2', start_mode=u'hourly', env=self.lgce)
        req = self.request()
        lgce2 = req.create_entity(
            'ProjectEnvironment', name=u'lgce2',
            check_config=u'install=setup_install',
            vcs_path=u'dir1',
            )
        lgc3 = self.add_test_config(u'lgc3', start_mode=u'hourly',
                                    check_config=u'branch=default',
                                    env=lgce2, start_rev_deps=True,
                                    use_recipe=self.recipe)
        self.execute('SET PE local_repository R WHERE PE name "lgce2", R eid %(r)s',
                     {'r': self.vcsrepo.eid})
        self.commit()
        self.assertEqual(self.get_tc('daily'),
                          set([]))
        self.assertEqual(self.get_tc('hourly'),
                          set((('lgce', 'lgc', None),
                               ('lgce', 'lgc2', None),
                               ('lgce2', 'lgc3', 1)))
                          )
        lgc2.cw_adapt_to('IWorkflowable').fire_transition('deactivate')
        self.commit()
        self.assertEqual(self.get_tc('hourly'),
                          set((('lgce', 'lgc', None),
                               ('lgce2', 'lgc3', 1)))
                          )
        self.commit()
        rset_before_lgc = self.get_te_for_tc_name(self.lgc.eid)
        te = lgc3.start(lgce2, branch=u'default')
        lgc3._cw.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('hourly'),
                          set((('lgce', 'lgc', None),
                               ('lgce2', 'lgc3', 1)))
                          )
        rset_before_lgc = self.get_te_for_tc_name(self.lgc.eid)
        rset_before_lgc3 = self.get_te_for_tc_name(lgc3.eid)
        self.launch_all_tests('hourly')
        self.commit()
        rset_after_lgc = self.get_te_for_tc_name(self.lgc.eid)
        rset_after_lgc3 = self.get_te_for_tc_name(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"""
        te = self.lgc.start(self.lgce, branch=u'default')
        self.commit()
        te.cw_adapt_to('IWorkflowable').fire_transition('start')
        self.commit()
        te.cw_adapt_to('IWorkflowable').fire_transition('fail')
        self.commit()
        #the status should not be waiting execution !
        self.assertNotEqual(u'waiting execution', te.status)


class ComputedStartModeTC(ApycotBaseTC):

    def test_start_mode(self):
        self.pyp.cw_set(start_mode=u'monthly')
        self.commit()
        self.assertEqual(self.pyp.computed_start_mode, self.pyp.start_mode)
        self.lgc.cw_set(start_mode=u'inherited')
        self.commit()
        self.lgc.cw_clear_all_caches()
        self.assertEqual(self.lgc.computed_start_mode, self.pyp.start_mode)
        self.pyp.cw_set(start_mode=u'hourly')
        self.commit()
        self.pyp.cw_clear_all_caches()
        self.lgc.cw_clear_all_caches()
        self.assertEqual(self.pyp.computed_start_mode, u'hourly')
        self.assertEqual(self.lgc.computed_start_mode, u'hourly')

if __name__ == '__main__':
    unittest_main()