1 # -*- coding: utf-8 -*-
2 ##############################################################################
4 # OpenERP, Open Source Management Solution
5 # Copyright (C) 2011-TODAY OpenERP S.A. <http://www.openerp.com>
7 # This program is free software: you can redistribute it and/or modify
8 # it under the terms of the GNU Affero General Public License as
9 # published by the Free Software Foundation, either version 3 of the
10 # License, or (at your option) any later version.
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU Affero General Public License for more details.
17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
20 ##############################################################################
23 from datetime import datetime
24 from dateutil.relativedelta import relativedelta
29 'function': u'_0_seconds',
30 'interval_type': u'minutes',
35 'nextcall': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
43 class test_ir_cron(openerp.osv.osv.osv):
44 """ Add a few handy methods to test cron jobs scheduling. """
47 def _0_seconds(a, b, c):
48 print ">>> _0_seconds"
50 def _20_seconds(self, cr, uid):
51 print ">>> in _20_seconds"
53 print ">>> out _20_seconds"
55 def _80_seconds(self, cr, uid):
56 print ">>> in _80_seconds"
58 print ">>> out _80_seconds"
60 def test_0(self, cr, uid):
62 t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
63 t2 = (now + relativedelta(minutes=1, seconds=5)).strftime('%Y-%m-%d %H:%M:%S')
64 t3 = (now + relativedelta(minutes=1, seconds=10)).strftime('%Y-%m-%d %H:%M:%S')
65 self.create(cr, uid, dict(JOB, name='test_0 _20_seconds A', function='_20_seconds', nextcall=t1))
66 self.create(cr, uid, dict(JOB, name='test_0 _20_seconds B', function='_20_seconds', nextcall=t2))
67 self.create(cr, uid, dict(JOB, name='test_0 _20_seconds C', function='_20_seconds', nextcall=t3))
69 def test_1(self, cr, uid):
71 t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
72 self.create(cr, uid, dict(JOB, name='test_1 _20_seconds * 3', function='_20_seconds', nextcall=t1, numbercall=3))
74 def test_2(self, cr, uid):
76 t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
77 self.create(cr, uid, dict(JOB, name='test_2 _80_seconds * 2', function='_80_seconds', nextcall=t1, numbercall=2))
79 def test_3(self, cr, uid):
81 t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
82 t2 = (now + relativedelta(minutes=1, seconds=5)).strftime('%Y-%m-%d %H:%M:%S')
83 t3 = (now + relativedelta(minutes=1, seconds=10)).strftime('%Y-%m-%d %H:%M:%S')
84 self.create(cr, uid, dict(JOB, name='test_3 _80_seconds A', function='_80_seconds', nextcall=t1))
85 self.create(cr, uid, dict(JOB, name='test_3 _20_seconds B', function='_20_seconds', nextcall=t2))
86 self.create(cr, uid, dict(JOB, name='test_3 _20_seconds C', function='_20_seconds', nextcall=t3))
88 # This test assumes 4 cron threads.
89 def test_00(self, cr, uid):
90 self.test_00_set = set()
92 t1 = (now + relativedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
93 t2 = (now + relativedelta(minutes=1, seconds=5)).strftime('%Y-%m-%d %H:%M:%S')
94 t3 = (now + relativedelta(minutes=1, seconds=10)).strftime('%Y-%m-%d %H:%M:%S')
95 self.create(cr, uid, dict(JOB, name='test_00 _20_seconds_A', function='_20_seconds_A', nextcall=t1))
96 self.create(cr, uid, dict(JOB, name='test_00 _20_seconds_B', function='_20_seconds_B', nextcall=t2))
97 self.create(cr, uid, dict(JOB, name='test_00 _20_seconds_C', function='_20_seconds_C', nextcall=t3))
99 def _expect(self, cr, uid, to_add, to_sleep, to_expect_in, to_expect_out):
100 assert self.test_00_set == to_expect_in
101 self.test_00_set.add(to_add)
103 self.test_00_set.discard(to_add)
104 assert self.test_00_set == to_expect_out
106 def _20_seconds_A(self, cr, uid):
107 self._expect(cr, uid, 'A', 20, set(), set(['B', 'C']))
109 def _20_seconds_B(self, cr, uid):
110 self._expect(cr, uid, 'B', 20, set('A'), set('C'))
112 def _20_seconds_C(self, cr, uid):
113 self._expect(cr, uid, 'C', 20, set(['A', 'B']), set())
115 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: