1. Tetsuya Morimoto
  2. trac.plugins.traccron

Commits

Tetsuya Morimoto  committed ce20571

added cronconf tests

  • Participants
  • Parent commits 6632752
  • Branches default

Comments (0)

Files changed (3)

File setup.cfg

View file
  • Ignore whitespace
 tag_svn_revision = true
 
 [pytest]
-pep8ignore = E302 E501 E701 W601
+pep8ignore = E128 E302 E501 E701 W601

File src/traccron/core.py

View file
  • Ignore whitespace
     Main class of the Trac Cron Plugin. This is the entry point
     for Trac plugin architecture
     """
-
     implements(IAdminPanelProvider, ITemplateProvider, IRequestHandler)
 
     cron_task_list = ExtensionPoint(ICronTask)
     """
     A Ticker is simply a simply timer that will repeatly wake up.
     """
-
     def __init__(self, env, interval, callback):
         """
         Create a new Ticker.
     """
     This class read and write configuration for TracCronPlugin
     """
+    TRACCRON_SECTION = 'traccron'
 
-    TRACCRON_SECTION = "traccron"
+    TICKER_ENABLED_KEY = 'ticker_enabled'
+    TICKER_ENABLED_DEFAULT = 'False'
 
-    TICKER_ENABLED_KEY = "ticker_enabled"
-    TICKER_ENABLED_DEFAULT = "False"
-
-    TICKER_INTERVAL_KEY = "ticker_interval"
+    TICKER_INTERVAL_KEY = 'ticker_interval'
     TICKER_INTERVAL_DEFAULT = 1  # minutes
 
-    TASK_ENABLED_KEY = "enabled"
-    TASK_ENABLED_DEFAULT = "True"
+    TASK_ENABLED_KEY = 'enabled'
+    TASK_ENABLED_DEFAULT = 'True'
 
-    SCHEDULE_ENABLED_KEY = "enabled"
-    SCHEDULE_ENABLED_DEFAULT = "True"
+    SCHEDULE_ENABLED_KEY = 'enabled'
+    SCHEDULE_ENABLED_DEFAULT = 'True'
 
-    SCHEDULE_ARGUMENT_KEY = "arg"
-    SCHEDULE_ARGUMENT_DEFAULT = ""
+    SCHEDULE_ARGUMENT_KEY = 'arg'
+    SCHEDULE_ARGUMENT_DEFAULT = ''
 
-    TASK_LISTENER_ENABLED_KEY = "enabled"
-    TASK_LISTENER_ENABLED_DEFAULT = "True"
+    TASK_LISTENER_ENABLED_KEY = 'enabled'
+    TASK_LISTENER_ENABLED_DEFAULT = 'True'
 
-    EMAIL_NOTIFIER_TASK_BASEKEY = "email_task_event"
+    EMAIL_NOTIFIER_TASK_BASEKEY = 'email_task_event'
 
-    EMAIL_NOTIFIER_TASK_LIMIT_KEY = "limit"
+    EMAIL_NOTIFIER_TASK_LIMIT_KEY = 'limit'
     EMAIL_NOTIFIER_TASK_LIMIT_DEFAULT = 1
 
-    EMAIL_NOTIFIER_TASK_RECIPIENT_KEY = "recipient"
-    EMAIL_NOTIFIER_TASK_RECIPIENT_DEFAULT = ""
+    EMAIL_NOTIFIER_TASK_RECIPIENT_KEY = 'recipient'
+    EMAIL_NOTIFIER_TASK_RECIPIENT_DEFAULT = ''
 
-    EMAIL_NOTIFIER_TASK_ONLY_ERROR_KEY = "only_error"
-    EMAIL_NOTIFIER_TASK_ONLY_ERROR_DEFAULT = "False"
+    EMAIL_NOTIFIER_TASK_ONLY_ERROR_KEY = 'only_error'
+    EMAIL_NOTIFIER_TASK_ONLY_ERROR_DEFAULT = 'False'
 
-    UNREACHABLE_MILESTONE_TASK_BASEKEY = "unreachable_milestone"
-    UNREACHABLE_MILESTONE_TASK_RECIPIENT_KEY = "recipient"
-    UNREACHABLE_MILESTONE_TASK_RECIPIENT_DEFAULT = ""
+    UNREACHABLE_MILESTONE_TASK_BASEKEY = 'unreachable_milestone'
+    UNREACHABLE_MILESTONE_TASK_RECIPIENT_KEY = 'recipient'
+    UNREACHABLE_MILESTONE_TASK_RECIPIENT_DEFAULT = ''
 
     def __init__(self, env):
         self.env = env
 
+    def _make_task_and_schedule_key(self, task, schedule_type):
+        return '%s.%s' % (task.getId(), schedule_type.getId())
+
+    def _make_task_enabled_key(self, task):
+        return '%s.%s' % (task.getId(), self.TASK_ENABLED_KEY)
+
+    def _make_schedule_enabled_key(self, task, schedule_type):
+        return '%s.%s.%s' % (task.getId(), schedule_type.getId(),
+                             self.SCHEDULE_ENABLED_KEY)
+
+    def _make_schedule_argument_key(self, task, schedule_type):
+        return '%s.%s.%s' % (task.getId(), schedule_type.getId(),
+                             self.SCHEDULE_ARGUMENT_KEY)
+
+    def _make_listener_enabled_key(self, listener):
+        return '%s.%s' % (listener.getId(), self.TASK_LISTENER_ENABLED_KEY)
+
+    @property
+    def email_notifier_task_limit_key(self):
+        return '%s.%s' % (self.EMAIL_NOTIFIER_TASK_BASEKEY,
+                          self.EMAIL_NOTIFIER_TASK_LIMIT_KEY)
+
+    @property
+    def email_notifier_task_recipient_key(self):
+        return '%s.%s' % (self.EMAIL_NOTIFIER_TASK_BASEKEY,
+                          self.EMAIL_NOTIFIER_TASK_RECIPIENT_KEY)
+
+    @property
+    def email_notifier_task_only_error_key(self):
+        return '%s.%s' % (self.EMAIL_NOTIFIER_TASK_BASEKEY,
+                          self.EMAIL_NOTIFIER_TASK_ONLY_ERROR_KEY)
+
+    @property
+    def unreachable_milestone_task_recipient_key(self):
+        return '%s.%s' % (self.UNREACHABLE_MILESTONE_TASK_BASEKEY,
+                          self.UNREACHABLE_MILESTONE_TASK_RECIPIENT_KEY)
+
     def get_ticker_enabled(self):
-        return self.env.config.getbool(self.TRACCRON_SECTION, self.TICKER_ENABLED_KEY, self.TICKER_ENABLED_DEFAULT)
+        return self.env.config.getbool(self.TRACCRON_SECTION,
+                                       self.TICKER_ENABLED_KEY,
+                                       self.TICKER_ENABLED_DEFAULT)
 
     def set_ticker_enabled(self, value):
-        self.env.config.set(self.TRACCRON_SECTION, self.TICKER_ENABLED_KEY, value)
+        self.env.config.set(self.TRACCRON_SECTION,
+                            self.TICKER_ENABLED_KEY,
+                            value)
 
     def get_ticker_interval(self):
-        return self.env.config.getint(self.TRACCRON_SECTION, self.TICKER_INTERVAL_KEY, self.TICKER_INTERVAL_DEFAULT)
+        return self.env.config.getint(self.TRACCRON_SECTION,
+                                      self.TICKER_INTERVAL_KEY,
+                                      self.TICKER_INTERVAL_DEFAULT)
 
     def set_ticker_interval(self, value):
-        self.env.config.set(self.TRACCRON_SECTION, self.TICKER_INTERVAL_KEY, value)
+        self.env.config.set(self.TRACCRON_SECTION,
+                            self.TICKER_INTERVAL_KEY,
+                            value)
 
     def get_schedule_value(self, task, schedule_type):
         """
         Return the raw value of the schedule
         """
-        return self.env.config.get(self.TRACCRON_SECTION, task.getId() + "." + schedule_type.getId(), None)
+        key = self._make_task_and_schedule_key(task, schedule_type)
+        return self.env.config.get(self.TRACCRON_SECTION, key, None)
 
     def get_schedule_value_list(self, task, schedule_type):
         """
         Return the list of value for the schedule type and task
         """
-        return self.env.config.getlist(self.TRACCRON_SECTION, task.getId() + "." + schedule_type.getId())
+        key = self._make_task_and_schedule_key(task, schedule_type)
+        return self.env.config.getlist(self.TRACCRON_SECTION, key)
 
     def set_schedule_value(self, task, schedule_type, value):
-        self.env.config.set(self.TRACCRON_SECTION, task.getId() + "." + schedule_type.getId(), value)
+        key = self._make_task_and_schedule_key(task, schedule_type)
+        self.env.config.set(self.TRACCRON_SECTION, key, value)
 
     def is_task_enabled(self, task):
         """
         Return the value that indicate if the task is enabled
         """
-        return self.env.config.getbool(self.TRACCRON_SECTION, task.getId() + "." + self.TASK_ENABLED_KEY, self.TASK_ENABLED_DEFAULT)
+        return self.env.config.getbool(self.TRACCRON_SECTION,
+                                       self._make_task_enabled_key(task),
+                                       self.TASK_ENABLED_DEFAULT)
 
     def set_task_enabled(self, task, value):
-        self.env.config.set(self.TRACCRON_SECTION, task.getId() + "." + self.TASK_ENABLED_KEY, value)
+        key = self._make_task_enabled_key(task)
+        self.env.config.set(self.TRACCRON_SECTION, key, value)
 
     def is_schedule_enabled(self, task, schedule):
         """
-        Return the value that indicate if the schedule for a given task is enabled
+        Return the value that indicate if the schedule for a given task
+        is enabled
         """
-        return self.env.config.getbool(self.TRACCRON_SECTION, task.getId() + "." + schedule.getId() + "." + self.SCHEDULE_ENABLED_KEY, self.SCHEDULE_ENABLED_DEFAULT)
+        key = self._make_schedule_enabled_key(task, schedule)
+        return self.env.config.getbool(self.TRACCRON_SECTION,
+                                       key,
+                                       self.SCHEDULE_ENABLED_DEFAULT)
 
     def set_schedule_enabled(self, task, schedule, value):
-        self.env.config.set(self.TRACCRON_SECTION, task.getId() + "." + schedule.getId() + "." + self.SCHEDULE_ENABLED_KEY, value)
+        key = self._make_schedule_enabled_key(task, schedule)
+        self.env.config.set(self.TRACCRON_SECTION, key, value)
 
     def get_schedule_arg(self, task, schedule):
         """
         Return the raw value of argument for a given schedule of a task
         """
-        return self.env.config.get(self.TRACCRON_SECTION, task.getId() + "." + schedule.getId() + "." + self.SCHEDULE_ARGUMENT_KEY, None)
+        key = self._make_schedule_argument_key(task, schedule)
+        return self.env.config.get(self.TRACCRON_SECTION, key, None)
 
     def get_schedule_arg_list(self, task, schedule):
         """
         Return the list of argument for a given schedule of a task
         """
-        return self.env.config.getlist(self.TRACCRON_SECTION, task.getId() + "." + schedule.getId() + "." + self.SCHEDULE_ARGUMENT_KEY)
+        key = self._make_schedule_argument_key(task, schedule)
+        return self.env.config.getlist(self.TRACCRON_SECTION, key)
 
     def set_schedule_arg(self, task, schedule, value):
-        self.env.config.set(self.TRACCRON_SECTION, task.getId() + "." + schedule.getId() + "." + self.SCHEDULE_ARGUMENT_KEY, value)
+        key = self._make_schedule_argument_key(task, schedule)
+        self.env.config.set(self.TRACCRON_SECTION, key, value)
 
     def get_email_notifier_task_limit(self):
         """
         Return the number of task event to notify.
         """
         return self.env.config.getint(self.TRACCRON_SECTION,
-                                      self.EMAIL_NOTIFIER_TASK_BASEKEY + "." + self.EMAIL_NOTIFIER_TASK_LIMIT_KEY, self.EMAIL_NOTIFIER_TASK_LIMIT_DEFAULT)
+                                      self.email_notifier_task_limit_key,
+                                      self.EMAIL_NOTIFIER_TASK_LIMIT_DEFAULT)
 
     def get_email_notifier_task_recipient(self):
         """
         Return the recipients for task listener as raw value
         """
         return self.env.config.get(self.TRACCRON_SECTION,
-                                   self.EMAIL_NOTIFIER_TASK_BASEKEY + "." + self.EMAIL_NOTIFIER_TASK_RECIPIENT_KEY, self.EMAIL_NOTIFIER_TASK_RECIPIENT_DEFAULT)
+                                   self.email_notifier_task_recipient_key,
+                                   self.EMAIL_NOTIFIER_TASK_RECIPIENT_DEFAULT)
 
     def get_email_notifier_task_recipient_list(self):
         """
         Return the recipients for task listener
         """
-        return self.env.config.getlist(self.TRACCRON_SECTION,
-                                       self.EMAIL_NOTIFIER_TASK_BASEKEY + "." + self.EMAIL_NOTIFIER_TASK_RECIPIENT_KEY)
+        key = self.email_notifier_task_recipient_key
+        return self.env.config.getlist(self.TRACCRON_SECTION, key)
 
     def is_email_notifier_only_error(self):
         """
         for task on error
         """
         return self.env.config.getbool(self.TRACCRON_SECTION,
-                                       self.EMAIL_NOTIFIER_TASK_BASEKEY + "." + self.EMAIL_NOTIFIER_TASK_ONLY_ERROR_KEY,
-                                       self.EMAIL_NOTIFIER_TASK_ONLY_ERROR_DEFAULT)
+                                self.email_notifier_task_only_error_key,
+                                self.EMAIL_NOTIFIER_TASK_ONLY_ERROR_DEFAULT)
 
     def set_email_notifier_only_error(self, value):
-        self.env.config.set(self.TRACCRON_SECTION, self.EMAIL_NOTIFIER_TASK_BASEKEY + "." + self.EMAIL_NOTIFIER_TASK_ONLY_ERROR_KEY, value)
+        key = self.email_notifier_task_only_error_key
+        self.env.config.set(self.TRACCRON_SECTION, key, value)
 
     def is_task_listener_enabled(self, listener):
-        return self.env.config.getbool(self.TRACCRON_SECTION, listener.getId() + "." + self.TASK_LISTENER_ENABLED_KEY, self.TASK_LISTENER_ENABLED_DEFAULT)
+        return self.env.config.getbool(self.TRACCRON_SECTION,
+                                self._make_listener_enabled_key(listener),
+                                self.TASK_LISTENER_ENABLED_DEFAULT)
 
     def set_task_listener_enabled(self, listener, value):
-        self.env.config.set(self.TRACCRON_SECTION, listener.getId() + "." + self.TASK_LISTENER_ENABLED_KEY, value)
+        key = self._make_listener_enabled_key(listener)
+        self.env.config.set(self.TRACCRON_SECTION, key, value)
 
     def get_unreachable_milestone_task_recipient_list(self):
         """
         Return recipient list for unreachable milestone
         """
-        return self.env.config.getlist(self.TRACCRON_SECTION,
-                                       self.UNREACHABLE_MILESTONE_TASK_BASEKEY + "." + self.UNREACHABLE_MILESTONE_TASK_RECIPIENT_KEY)
+        key = self.unreachable_milestone_task_recipient_key
+        return self.env.config.getlist(self.TRACCRON_SECTION, key)
 
     def get_unreachable_milestone_task_recipient(self):
         """
         Return raw value of unreachable milestone recipient
         """
         return self.env.config.get(self.TRACCRON_SECTION,
-                                   self.UNREACHABLE_MILESTONE_TASK_BASEKEY + "." + self.UNREACHABLE_MILESTONE_TASK_RECIPIENT_KEY,
-                                   self.UNREACHABLE_MILESTONE_TASK_RECIPIENT_DEFAULT)
+                            self.unreachable_milestone_task_recipient_key,
+                            self.UNREACHABLE_MILESTONE_TASK_RECIPIENT_DEFAULT)
 
     def set_value(self, key, value):
         self.env.config.set(self.TRACCRON_SECTION, key, value)
     def save(self):
         self.env.config.save()
 
+
 class WebUi(IAdminPanelProvider, ITemplateProvider, IRequestHandler):
     """
     Class that deal with Web stuff. It is the both the controller and the page builder.
     """
-
     def __init__(self, core):
         self.env = core.env
         self.cron_task_list = core.getTaskList()

File tests/test_cronconf.py

View file
  • Ignore whitespace
+# -*- encoding: UTF-8 -*-
+import pytest
+
+from traccron.core import CronConfig
+
+
+def pytest_funcarg__cronconf(request, env):
+    cronconf = CronConfig(env)
+    return cronconf
+
+def pytest_funcarg__task(request, component):
+    task = component['heart_beat_task']
+    return task
+
+def pytest_funcarg__cron_scheduler(request, component):
+    cron_scheduler = component['cron_scheduler']
+    return cron_scheduler
+
+def pytest_funcarg__history_task_event(request, component):
+    history_task_event = component['history_task_event']
+    return history_task_event
+
+
+def test_get_ticker_enabled(cronconf):
+    default_value = eval(CronConfig.TICKER_ENABLED_DEFAULT)
+    assert default_value is cronconf.get_ticker_enabled()
+
+def test_set_ticker_enabled(cronconf):
+    cronconf.set_ticker_enabled(True)
+    assert True is cronconf.get_ticker_enabled()
+    cronconf.set_ticker_enabled(False)
+    assert False is cronconf.get_ticker_enabled()
+
+def test_get_ticker_interval(cronconf):
+    assert CronConfig.TICKER_INTERVAL_DEFAULT == cronconf.get_ticker_interval()
+
+def test_set_ticker_interval(cronconf):
+    cronconf.set_ticker_interval(10)
+    assert 10 == cronconf.get_ticker_interval()
+
+
+def test_get_schedule_value(cronconf, task, cron_scheduler):
+    assert None is cronconf.get_schedule_value(task, cron_scheduler)
+
+def test_get_schedule_value_list(cronconf, task, cron_scheduler):
+    assert 0 == len(cronconf.get_schedule_value_list(task, cron_scheduler))
+
+def test_set_schedule_value(cronconf, task, cron_scheduler):
+    value = '*/1 * * * * *'
+    cronconf.set_schedule_value(task, cron_scheduler, value)
+    assert value == cronconf.get_schedule_value(task, cron_scheduler)
+    assert 1 == len(cronconf.get_schedule_value_list(task, cron_scheduler))
+
+
+def test_is_task_enabled(cronconf, task):
+    default_value = eval(CronConfig.TASK_ENABLED_DEFAULT)
+    assert default_value is cronconf.is_task_enabled(task)
+
+def test_set_task_enabled(cronconf, task):
+    cronconf.set_task_enabled(task, True)
+    assert True is cronconf.is_task_enabled(task)
+    cronconf.set_task_enabled(task, False)
+    assert False is cronconf.is_task_enabled(task)
+
+
+def test_is_schedule_enabled(cronconf, task, cron_scheduler):
+    default_value = eval(CronConfig.SCHEDULE_ENABLED_DEFAULT)
+    assert default_value is cronconf.is_schedule_enabled(task, cron_scheduler)
+
+def test_set_schedule_enabled(cronconf, task, cron_scheduler):
+    cronconf.set_schedule_enabled(task, cron_scheduler, True)
+    assert True is cronconf.is_schedule_enabled(task, cron_scheduler)
+    cronconf.set_schedule_enabled(task, cron_scheduler, False)
+    assert False is cronconf.is_schedule_enabled(task, cron_scheduler)
+
+
+def test_get_schedule_arg(cronconf, task, cron_scheduler):
+    assert None is cronconf.get_schedule_arg(task, cron_scheduler)
+
+def test_get_schedule_arg_list(cronconf, task, cron_scheduler):
+    assert 0 == len(cronconf.get_schedule_arg_list(task, cron_scheduler))
+
+def test_set_schedule_arg(cronconf, task, cron_scheduler):
+    value = 'testtest'
+    cronconf.set_schedule_arg(task, cron_scheduler, value)
+    assert value == cronconf.get_schedule_arg(task, cron_scheduler)
+    assert 1 == len(cronconf.get_schedule_arg_list(task, cron_scheduler))
+
+
+def test_get_email_notifier_task_limit(cronconf):
+    default_value = CronConfig.EMAIL_NOTIFIER_TASK_LIMIT_DEFAULT
+    assert default_value == cronconf.get_email_notifier_task_limit()
+
+def test_get_email_notifier_task_recipient(cronconf):
+    default_value = CronConfig.EMAIL_NOTIFIER_TASK_RECIPIENT_DEFAULT
+    assert default_value == cronconf.get_email_notifier_task_recipient()
+
+def test_get_email_notifier_task_recipient_list(cronconf):
+    assert 0 == len(cronconf.get_email_notifier_task_recipient_list())
+
+
+def test_is_email_notifier_only_error(cronconf):
+    default_value = eval(CronConfig.EMAIL_NOTIFIER_TASK_ONLY_ERROR_DEFAULT)
+    assert default_value is cronconf.is_email_notifier_only_error()
+
+def test_set_email_notifier_only_error(cronconf):
+    cronconf.set_email_notifier_only_error(True)
+    assert True is cronconf.is_email_notifier_only_error()
+    cronconf.set_email_notifier_only_error(False)
+    assert False is cronconf.is_email_notifier_only_error()
+
+
+def test_is_task_listener_enabled(cronconf, history_task_event):
+    default_value = eval(CronConfig.TASK_LISTENER_ENABLED_DEFAULT)
+    assert default_value is cronconf.is_task_listener_enabled(history_task_event)
+
+def test_set_task_listener_enabled(cronconf, history_task_event):
+    cronconf.set_task_listener_enabled(history_task_event, True)
+    assert True is cronconf.is_task_listener_enabled(history_task_event)
+    cronconf.set_task_listener_enabled(history_task_event, False)
+    assert False is cronconf.is_task_listener_enabled(history_task_event)
+
+
+def test_get_unreachable_milestone_task_recipient_list(cronconf):
+    assert 0 == len(cronconf.get_unreachable_milestone_task_recipient_list())
+
+def test_get_unreachable_milestone_task_recipient(cronconf):
+    default_value = CronConfig.UNREACHABLE_MILESTONE_TASK_RECIPIENT_DEFAULT
+    assert default_value == cronconf.get_unreachable_milestone_task_recipient()
+
+
+def test_set_and_remove_value(cronconf):
+    section = CronConfig.TRACCRON_SECTION
+    key, value = 'cronconf-test', 'cronconf-test-value'
+    cronconf.set_value(key, value)
+    assert value == cronconf.env.config.get(section, key, None)
+    cronconf.remove_value(key)
+    assert None is cronconf.env.config.get(section, key, None)