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

import os
import re

from logilab.common.testlib import unittest_main

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


from utils import INPUTS_DIR, ApycotBaseTC


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()
        self.dumb_execution(plan, [('unittest', 'success'), ('coverage', 'success')])
        self.assertEqual(len(MAILBOX), 0)
        self.commit()
        self.assertEqual(len(MAILBOX), 0)
        plan = self.start_lgc_tests()
        self.dumb_execution(plan, [('unittest', 'success'), ('coverage', 'success')])
        self.assertEqual(len(MAILBOX), 0)
        self.commit()
        self.assertEqual(len(MAILBOX), 0)
        plan = self.start_lgc_tests()
        self.dumb_execution(plan, [('unittest', 'failure'), ('coverage', 'failure')])
        self.assertEqual(len(MAILBOX), 0, MAILBOX)
        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()
        self.dumb_execution(plan, [('unittest', 'success'), ('coverage', 'success')])
        self.commit()
        plan = self.start_lgc_tests()
        self.dumb_execution(plan, [('unittest', 'failure')])
        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])


HGREPO = os.path.join(INPUTS_DIR, u'hgrepo')
def setupModule(*args, **kwargs):
    if not os.path.exists(HGREPO):
        os.mkdir(HGREPO)
        os.system('cd %s && hg init' % HGREPO)

def teardownModule(*args, **kwargs):
    if os.path.exists(HGREPO):
        os.system('rm -rf %s' % HGREPO)


class StartTestTC(ApycotBaseTC):

    def grant_write_perm(self, repo, group):
        req = self.request()
        managers = req.execute('CWGroup G WHERE G name %(group)s',
                               {'group': group}).get_entity(0, 0)
        req.create_entity(
            'CWPermission', name=u"write", label=u'repo x write perm',
            reverse_require_permission=repo,
            require_group=managers)

    def test_new_vc_trigger(self):
        self.skipTest('XXX need update')
        self.lgc.set_attributes(start_mode=u'on new revision')
        lgc2 = self.add_test_config(u'lgc2', start_mode=u'manual', env=self.lgce)
        lgc3 = self.add_test_config(u'lgc3', start_mode=u'on new revision',
                                    check_config=u'branch=stable', env=self.lgce)
        lgce2 = self.request().create_entity(
            'ProjectEnvironment', name=u'lgce2',
            check_config=u'install=setup_install',
            vcs_path=u'dir1',
            )
        req = self.request()
        req.create_entity('Repository', type=u'mercurial',
                          source_url=u'http://www.logilab.org/src/logilab/common',
                          reverse_local_repository=lgce2)
        lgc4 = self.add_test_config(u'lgc4', start_mode=u'on new revision',
                                    check_config=u'branch=default',
                                    env=lgce2, start_rev_deps=True)
        lgc5 = self.add_test_config(u'lgc5', start_mode=u'manual',
                                    env=lgce2)
        self.commit()
        r = self.request().create_entity('Repository', path=HGREPO, type=u'mercurial', encoding=u'utf8')
        self.execute('SET PE local_repository R WHERE PE is ProjectEnvironment, R is Repository')
        self.grant_write_perm(r, 'managers')
        self.commit()
        # now test
        r.vcs_add(u'dir1', u'tutu.png', Binary('data'))
        self.commit()
        self.assertEqual(sorted(proxy.queued),
                          # data is the pyro instance id
                          [(('lgce', 'lgc'), {'cwinstid': ':cubicweb.data',
                                              'branch': 'default',
                                              'start_rev_deps': False}),
                           (('lgce2', 'lgc4'), {'cwinstid': ':cubicweb.data',
                                                'branch': 'default',
                                                'start_rev_deps': True})])
        proxy.queued = []
        r.vcs_add(u'dir1', u'tutu.png', Binary('data'), branch=u'stable')
        self.commit()
        self.assertEqual(proxy.queued, [(('lgce', 'lgc'), {'cwinstid': ':cubicweb.data',
                                                            'branch': 'stable',
                                                            'start_rev_deps': False}),
                                         (('lgce', 'lgc3'), {'cwinstid': ':cubicweb.data',
                                                             'branch': 'stable',
                                                             'start_rev_deps': False})])
        proxy.queued = []
        r.vcs_add(u'dir2', u'tutu.png', Binary('data'))
        self.commit()
        self.assertEqual(proxy.queued, [(('lgce', 'lgc'), {'cwinstid': ':cubicweb.data',
                                                            'branch': 'default',
                                                            'start_rev_deps': False})])

    def test_datetime_trigger(self):
        self.skipTest('XXX need update')
        self.lgc.set_attributes(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',
            )
        req.create_entity('Repository', type=u'mercurial',
                          source_url=u'http://www.logilab.org/src/logilab/common',
                          reverse_local_repository=lgce2)
        lgc3 = self.add_test_config(u'lgc3', start_mode=u'hourly',
                                    check_config=u'branch=default',
                                    env=lgce2, start_rev_deps=True)
        r = req.create_entity('Repository', path=HGREPO, type=u'mercurial', encoding=u'utf8')
        self.grant_write_perm(r, 'managers')
        r.vcs_add(u'dir1', u'tutu.png', Binary('data'))
        self.execute('SET PE local_repository R WHERE PE name "lgce2"')
        self.commit()
        session = self.session
        session.set_pool()
        self.assertEqual(start_test(session, 'daily'),
                          set())
        self.assertEqual(start_test(session, 'hourly'),
                          set((('lgce', 'lgc', False, None),
                               ('lgce', 'lgc2', False, None),
                               ('lgce2', 'lgc3', True, 'default')))
                          )
        lgc2.cw_adapt_to('IWorkflowable').fire_transition('deactivate')
        self.commit()
        session.set_pool()
        self.assertEqual(start_test(session, 'hourly'),
                          set((('lgce', 'lgc', False, None),
                               ('lgce2', 'lgc3', True, 'default')))
                          )
        self.login('narval', password='narval0')
        ex = self.dumb_execution('lgc3', (), setend=True)
        ex.set_relations(using_revision=r.branch_head())
        self.commit()
        self.restore_connection()
        session.set_pool()
        self.assertEqual(start_test(session, 'hourly'),
                          set((('lgce', 'lgc', False, None),))
                          )

class ComputedStartModeTC(ApycotBaseTC):

    def test_start_mode(self):
        self.pyp.set_attributes(start_mode=u'monthly')
        self.commit()
        self.assertEqual(self.pyp.computed_start_mode, self.pyp.start_mode)
        self.lgc.set_attributes(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.set_attributes(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()