#!/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):
_repo_path = (u'project1',)
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):
self.refresh()
with self.admin_access.client_cnx() as cnx:
pe = cnx.find('ProjectEnvironment').one()
tc = cnx.find('TestConfig', name='tc_quick').one()
plan = tc.start(pe)
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] pe_project1/tc_quick#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] pe_project1/tc_quick#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"')
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 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:
pe = cnx.find('ProjectEnvironment').one()
tc = cnx.find('TestConfig', name='tc_quick').one()
te = tc.start(pe, 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()
self.assertEqual(u'error', 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()