odoo
/
odoo
2
0
Fork 0
Code Pull Requests Releases Activity

[IMP] cron: added some code to test the new threaded jobs, should be moved elsewhere. 

bzr revid: vmt@openerp.com-20110715094051-7z5v2xu91uid0jtm
This commit is contained in:
Vo Minh Thu 2011-07-15 11:40:51 +02:00
parent 4b8708fb10
commit ed910b5898
1 changed files with 53 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
openerp/addons/base/ir/ir_cron.py
View File

@ -44,6 +44,21 @@ _intervalTypes = {
'minutes': lambda interval: relativedelta(minutes=interval),
}
JOB = {
'function': u'f',
'interval_type': u'minutes',
'user_id': 1,
'name': u'test',
'args': False,
'numbercall': 1,
'nextcall': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'priority': 5,
'doall': True,
'active': True,
'interval_number': 1,
'model': u'ir.cron'
}
class ir_cron(osv.osv):
""" This is the ORM object that periodically executes actions.
"""
@ -240,6 +255,44 @@ class ir_cron(osv.osv):
cr.commit()
openerp.cron.schedule_in_advance(1, self.pool.db.dbname)
def _20_seconds(self, cr, uid):
print ">>> in _20_seconds"
time.sleep(20)
print ">>> out _20_seconds"
def _80_seconds(self, cr, uid):
print ">>> in _80_seconds"
time.sleep(80)
print ">>> out _80_seconds"
def test_0(self, cr, uid):
now = datetime.now()
t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
t2 = (now + relativedelta(minutes=1, seconds=5)).strftime('%Y-%m-%d %H:%M:%S')
t3 = (now + relativedelta(minutes=1, seconds=10)).strftime('%Y-%m-%d %H:%M:%S')
self.create(cr, uid, dict(JOB, name='test_0 _20_seconds A', function='_20_seconds', nextcall=t1))
self.create(cr, uid, dict(JOB, name='test_0 _20_seconds B', function='_20_seconds', nextcall=t2))
self.create(cr, uid, dict(JOB, name='test_0 _20_seconds C', function='_20_seconds', nextcall=t3))
def test_1(self, cr, uid):
now = datetime.now()
t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
self.create(cr, uid, dict(JOB, name='test_1 _20_seconds * 3', function='_20_seconds', nextcall=t1, numbercall=3))
def test_2(self, cr, uid):
now = datetime.now()
t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
self.create(cr, uid, dict(JOB, name='test_2 _80_seconds * 2', function='_80_seconds', nextcall=t1, numbercall=2))
def test_3(self, cr, uid):
now = datetime.now()
t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
t2 = (now + relativedelta(minutes=1, seconds=5)).strftime('%Y-%m-%d %H:%M:%S')
t3 = (now + relativedelta(minutes=1, seconds=10)).strftime('%Y-%m-%d %H:%M:%S')
self.create(cr, uid, dict(JOB, name='test_3 _80_seconds A', function='_80_seconds', nextcall=t1))
self.create(cr, uid, dict(JOB, name='test_3 _20_seconds B', function='_20_seconds', nextcall=t2))
self.create(cr, uid, dict(JOB, name='test_3 _20_seconds C', function='_20_seconds', nextcall=t3))
def create(self, cr, uid, vals, context=None):
res = super(ir_cron, self).create(cr, uid, vals, context=context)
self.update_running_cron(cr)