Source Code for Module expedient.common.timer.tests

  1  ''' 
  2  Created on Feb 16, 2011 
  3   
  4  @author: jnaous 
  5  ''' 
  6  from django.test import TestCase 
  7  from datetime import datetime 
  8  from expedient.common.timer.models import Job 
  9  from time import sleep 
 10  from expedient.common.timer.exceptions import JobAlreadyScheduled,\ 
 11      JobNotScheduled 
 12  from django.core.management import call_command 
 13  import logging 
 14  from django.db.models.signals import post_syncdb 
 15  from expedient.common.timer import models as timer_app 
 16   
 17  logger = logging.getLogger("timer.tests") 
 18   
 19  times = dict( 
 20      job1_time=None, 
 21      job2_time=None, 
 22      job3_time=None, 
 23      job4_time=None, 
 24      job5_time=None, 
 25      job6_time=None, 
 26  ) 
 27   
 28 -def jobx_call(i): 
 29      times["job%s_time" % i] = datetime.now() 
 30      logger.debug("Timer %s called." % i) 
 31   
 32 -def job1_call(): jobx_call(1) 
 33 -def job2_call(): jobx_call(2) 
 34 -def job3_call(): jobx_call(3) 
 35 -def job4_call(): jobx_call(4) 
 36 -def job5_call(): jobx_call(5) 
 37 -def job6_call(): jobx_call(6) 
 38   
 39  Job.objects.schedule_post_syncdb(4, job5_call) 
 40  Job.objects.schedule_post_syncdb(4, job6_call) 
 41   
 42 -class Tests(TestCase): 
 43       
 44 -    def test_methods(self): 
 45          Job.objects.exclude( 
 46              callable_name__regex=r'expedient.common.timer.tests.job.?_call$', 
 47          ).delete() 
 48          # create jobs 
 49          Job.objects.schedule(4, job1_call) 
 50          Job.objects.schedule(4, job2_call) 
 51          Job.objects.schedule(4, job3_call) 
 52          Job.objects.schedule(4, job4_call) 
 53           
 54          # check that there are 6 jobs now including 
 55          # the post syncdb one. 
 56          self.assertEqual( 
 57              Job.objects.count(), 6, 
 58              "Wrong number of jobs. " 
 59              "Existing job callables: %s" %  
 60              Job.objects.values_list("callable_name", flat=True), 
 61          ) 
 62           
 63          # try to schedule an already scheduled job 
 64          self.assertRaises(JobAlreadyScheduled, Job.objects.schedule, 1, job1_call) 
 65           
 66          # wait 2s, delete one of the jobs that are supposed to run 
 67          sleep(2) 
 68           
 69          Job.objects.unschedule(job4_call) 
 70           
 71          # check that there are 5 jobs now 
 72          self.assertEqual(Job.objects.count(), 5) 
 73           
 74          # try to unschedule a non-existent job 
 75          self.assertRaises(JobNotScheduled, Job.objects.unschedule, job4_call) 
 76           
 77          # check that conditional execution of a job not on 
 78          # time won't do anything 
 79          job1 = Job.objects.get_for(job1_call) 
 80          ret = job1.conditional_execute() 
 81          self.assertEqual(ret, (None, False)) 
 82          self.assertEqual(times["job1_time"], None) 
 83           
 84          # reset the timer for job 3 so it executes 2 seconds after job1 
 85          job3 = Job.objects.get_for(job3_call) 
 86          job3.reset_timer() 
 87           
 88          # check that jobs 1, 2, and 5 get executed on time 
 89          # using the command 
 90          self.assertEqual(times["job1_time"], None) 
 91          self.assertEqual(times["job2_time"], None) 
 92          self.assertEqual(times["job5_time"], None) 
 93          self.assertEqual(times["job6_time"], None) 
 94          sleep(2) 
 95          self.assertEqual(times["job1_time"], None) 
 96          self.assertEqual(times["job2_time"], None) 
 97          self.assertEqual(times["job5_time"], None) 
 98          self.assertEqual(times["job6_time"], None) 
 99          call_command("run_timer_jobs") 
100          self.assertNotEqual(times["job1_time"], None) 
101          self.assertNotEqual(times["job2_time"], None) 
102          self.assertNotEqual(times["job5_time"], None) 
103          self.assertEqual(times["job3_time"], None) 
104          self.assertEqual(times["job4_time"], None) 
105   
106          job1_time1 = times["job1_time"] 
107          job2_time1 = times["job2_time"] 
108          job5_time1 = times["job5_time"] 
109           
110          # Check that the job was rescheduled 
111          job1 = Job.objects.get_for(job1_call) 
112          self.assertTrue(job1.next_run_time > datetime.now()) 
113           
114          # Check that conditional_execute works on job 3 
115          sleep(2) 
116          ret = job3.conditional_execute() 
117          self.assertTrue(ret[1]) 
118          self.assertNotEqual(times["job3_time"], None) 
119           
120          # check that jobs 1 and 2 run again after their period but not before 
121          job1 = Job.objects.get_for(job1_call) 
122          ret = job1.conditional_execute() 
123          self.assertEqual(ret, (None, False)) 
124          self.assertEqual(times["job1_time"], job1_time1) 
125           
126          sleep(2) 
127          call_command("run_timer_jobs") 
128          job1_time2 = times["job1_time"] 
129          job2_time2 = times["job2_time"] 
130          job5_time2 = times["job2_time"] 
131   
132          self.assertTrue(job1_time2 > job1_time1) 
133          self.assertTrue(job2_time2 > job2_time1) 
134          self.assertTrue(job5_time2 > job5_time1) 
135