openerp/addons/base/test/test_ir_cron.py

   1 # -*- coding: utf-8 -*-
   2 ##############################################################################
   3 #
   4 #    OpenERP, Open Source Management Solution
   5 #    Copyright (C) 2011-TODAY OpenERP S.A. <http://www.openerp.com>
   6 #
   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.
  11 #
  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.
  16 #
  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/>.
  19 #
  20 ##############################################################################
  21
  22 import time
  23 from datetime import datetime
  24 from dateutil.relativedelta import relativedelta
  25
  26 import openerp
  27
  28 JOB = {
  29     'function': u'_0_seconds',
  30     'interval_type': u'minutes',
  31     'user_id': 1,
  32     'name': u'test',
  33     'args': False,
  34     'numbercall': 1,
  35     'nextcall': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
  36     'priority': 5,
  37     'doall': True,
  38     'active': True,
  39     'interval_number': 1,
  40     'model': u'ir.cron'
  41 }
  42
  43 class test_ir_cron(openerp.osv.osv.osv):
  44     """ Add a few handy methods to test cron jobs scheduling. """
  45     _inherit = "ir.cron"
  46
  47     def _0_seconds(a, b, c):
  48         print ">>> _0_seconds"
  49
  50     def _20_seconds(self, cr, uid):
  51         print ">>> in _20_seconds"
  52         time.sleep(20)
  53         print ">>> out _20_seconds"
  54
  55     def _80_seconds(self, cr, uid):
  56         print ">>> in _80_seconds"
  57         time.sleep(80)
  58         print ">>> out _80_seconds"
  59
  60     def test_0(self, cr, uid):
  61         now = datetime.now()
  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))
  68
  69     def test_1(self, cr, uid):
  70         now = datetime.now()
  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))
  73
  74     def test_2(self, cr, uid):
  75         now = datetime.now()
  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))
  78
  79     def test_3(self, cr, uid):
  80         now = datetime.now()
  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))
  87
  88     # This test assumes 4 cron threads.
  89     def test_00(self, cr, uid):
  90         self.test_00_set = set()
  91         now = datetime.now()
  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))
  98
  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)
 102         time.sleep(to_sleep)
 103         self.test_00_set.discard(to_add)
 104         assert self.test_00_set == to_expect_out
 105
 106     def _20_seconds_A(self, cr, uid):
 107         self._expect(cr, uid, 'A', 20, set(), set(['B', 'C']))
 108
 109     def _20_seconds_B(self, cr, uid):
 110         self._expect(cr, uid, 'B', 20, set('A'), set('C'))
 111
 112     def _20_seconds_C(self, cr, uid):
 113         self._expect(cr, uid, 'C', 20, set(['A', 'B']), set())
 114
 115 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
 116