Commits

Tetsuya Morimoto committed 0679eb1

added DailyScheduler component and SchedulerType tests

Comments (0)

Files changed (3)

src/traccron/scheduler.py

 # -*- encoding: UTF-8 -*-
-'''
+"""
 Created on 28 oct. 2010
 
 @author: thierry
-'''
+"""
 ###############################################################################
 ##
 ##          O U T    O F    T H E    B O X    S C H E D U L E R
 
 class SchedulerType(ISchedulerType):
     """
-    Define a sort of scheduling. Base class for any scheduler type implementation
+    Define a sort of scheduling. Base class for any scheduler type
+    implementation
     """
-
     implements(ISchedulerType)
 
     def __init__(self):
         self.cronconf = CronConfig(self.env)
+        self._comp_debug = 'compareTime: %s, %s, %s'
 
     def getId(self):
         """
 
     def getHint(self):
         """
-        Return a description of what it is and the format used to defined the schedule
+        Return a description of what it is and the format used to defined
+        the schedule
         """
-        return ""
+        return ''
 
     def isTriggerTime(self, task, currentTime):
         """
         is time to fire the task
         """
         # read the configuration value for the task
-        self.env.log.debug("looking for schedule of type: %s" % self.getId())
+        self.env.log.debug('looking for schedule of type: %s' % self.getId())
         for schedule_value in self._get_task_schedule_value_list(task):
-            msg = "task is scheduled: %s, %s" % (task.getId(), schedule_value)
+            msg = 'task is scheduled: %s, %s' % (task.getId(), schedule_value)
             self.env.log.debug(msg)
-            if self.compareTime(currentTime, schedule_value):
+            if schedule_value and self.compareTime(currentTime,
+                                                   schedule_value):
                 return True
-        self.env.log.debug("no matching schedule found")
+        self.env.log.debug('no matching schedule found')
         return False
 
     def compareTime(self, currentTime, schedule_value):
         """
-        Test is accordingly to this scheduler, given currentTime and schedule value,
-        is time to fire the task.
+        Test is accordingly to this scheduler, given currentTime and
+        schedule value, is time to fire the task.
         currentTime is a structure computed by time.localtime(time())
         scheduled_value is the value of the configuration in trac.ini
         """
         SchedulerType.__init__(self)
 
     def getId(self):
-        return "daily"
+        return 'daily'
 
     def getHint(self):
-        return "ex: 8h30 fire every day at 8h30"
+        return 'ex: 8h30 fire every day at 8h30'
 
     def compareTime(self, currentTime, schedule_value):
-        # compare value with current
-        if schedule_value:
-            self.env.log.debug(self.getId() + " compare currentTime=" + str(currentTime) + " with schedule_value " + schedule_value)
-        else:
-            self.env.log.debug(self.getId() + " compare currentTime=" + str(currentTime) + " with NO schedule_value ")
-        if schedule_value:
-            return schedule_value == str(currentTime.tm_hour) + "h" + str(currentTime.tm_min)
-        else:
-            return False
+        msg = self._comp_debug % (self.getId(), currentTime, schedule_value)
+        self.env.log.debug(msg)
+        return schedule_value == '%sh%s' % currentTime[3:5]
 
 
 class HourlyScheduler(Component, SchedulerType):
         return "monthly"
 
     def getHint(self):
-        return "ex: 15@12h00 fire every month on the 15th day of month at 12h00"
+        msg = "ex: 15@12h00 fire every month on the 15th day of month at 12h00"
+        return msg
 
     def compareTime(self, currentTime, schedule_value):
         # compare value with current
 
 class CronScheduler(Component, SchedulerType):
     """
-    Scheduler that used a cron-like syntax to specified when task must be triggered
+    Scheduler that used a cron-like syntax to specified when task must
+    be triggered
     """
     def __init__(self):
         SchedulerType.__init__(self)
                 return obj
 
         def matchtime(self, t):
-            """Return True if this event should trigger at the specified localtime"""
+            """
+            Return True if this event should trigger at the specified localtime
+            """
             return ((t.tm_min in self.mins) and
                     (t.tm_hour in self.hours) and
                     (t.tm_mday in self.days) and
         """
         utility method to parse value of a cron item.
         Support of *, range expression (ex 1-10)
-        adjust is used to translate value (ex: first day of week is 0 in python and 1 in Cron)
+        adjust is used to translate value
+        (ex: first day of week is 0 in python and 1 in Cron)
         """
         if value == "*":
             kwargs[event_param] = CronScheduler._allMatch
 
     def _parse_cron_expression(self, cron):
         '''
-        Parse cron expression and return dictionary of argument key/value suitable
-        for Event object
+        Parse cron expression and return dictionary of argument key/value
+        suitable for Event object
         '''
         self.env.log.debug("parsing cron expression %s" % cron)
         kwargs = {}

tests/test_daily_scheduler.py

+# -*- coding: utf-8 -*-
+from time import localtime, time
+
+import pytest
+
+time_now = time()
+local_now = localtime(time_now)
+local_now_plus_1min = localtime(time_now + 60)
+local_now_plus_1hour = localtime(time_now + 3600)
+
+
+def pytest_funcarg__daily_scheduler(request, component):
+    daily_scheduler = component['daily_scheduler']
+    return daily_scheduler
+
+
+def test_daily_scheduler_basic(daily_scheduler):
+    assert 'daily' == daily_scheduler.getId()
+
+
+@pytest.mark.parametrize(('cur_time', 'sch_value', 'expected'), [
+    (local_now, '%sh%s' % local_now[3:5], True),
+    (local_now, '%sh%s' % local_now_plus_1min[3:5], False),
+    (local_now, '%sh%s' % local_now_plus_1hour[3:5], False),
+    (local_now, None, False),
+    (local_now, '', False),
+])
+def test_daily_scheduler_compareTime(daily_scheduler,
+                                     cur_time, sch_value, expected):
+    assert expected is daily_scheduler.compareTime(cur_time, sch_value)

tests/test_scheduler_type.py

+# -*- coding: utf-8 -*-
+from time import localtime, time
+
+import pytest
+
+time_now = time()
+local_now = localtime(time_now)
+
+def pytest_funcarg__task(request, component):
+    task = component['heart_beat_task']
+    return task
+
+def pytest_funcarg__my_scheduler(request, env):
+    from traccron.scheduler import SchedulerType
+
+    class MyScheduler(SchedulerType):
+        def __init__(self, env):
+            self.env = env
+            self.schedule_task_list = []
+            SchedulerType.__init__(self)
+
+        def getId(self):
+            return 'myscheduler'
+
+        def compareTime(self, currentTime, schedule_value):
+            return currentTime == schedule_value
+
+        def _get_task_schedule_value_list(self, task):
+            return self.schedule_task_list
+
+    my_scheduler = MyScheduler(env)
+    return my_scheduler
+
+
+@pytest.mark.parametrize(('cur_time', 'value_list', 'expected'), [
+    (local_now, [local_now], True),
+    (local_now, ['', local_now], True),
+    (local_now, [None, local_now], True),
+    (local_now, [], False),
+    (local_now, [''], False),
+    (local_now, ['9h10'], False),
+    (local_now, [None], False),
+])
+def test_isTriggerTime(my_scheduler, task, cur_time, value_list, expected):
+    my_scheduler.schedule_task_list = value_list
+    assert expected is my_scheduler.isTriggerTime(task, cur_time)