Tetsuya Morimoto avatar Tetsuya Morimoto committed cc8d0f8

added NotificationEmailTaskEvent component tests

Comments (0)

Files changed (2)

src/traccron/listener.py

 # -*- encoding: UTF-8 -*-
-'''
+"""
 Created on 28 oct. 2010
 
 @author: thierry
-'''
+"""
 ###############################################################################
 ##
 ##        O U T    O F    T H E    B O X    T A S K    L I S T E N E R
 from traccron.core import CronConfig
 
 
+class NotifyEmailTaskEvent(NotifyEmail):
+
+    template_name = 'notify_task_event_template.txt'
+
+    def __init__(self, env):
+        NotifyEmail.__init__(self, env)
+        self.cronconf = CronConfig(self.env)
+
+    def get_recipients(self, resid):
+        """
+        Return the recipients as defined in trac.ini.
+        """
+        reclist = self.cronconf.get_email_notifier_task_recipient_list()
+        return (reclist, [])
+
+    def notifyTaskEvent(self, task_event_list):
+        """
+        Send task event by mail if recipients is defined in trac.ini
+        """
+        self.env.log.debug('notifying task event...')
+        if self.cronconf.get_email_notifier_task_recipient():
+            # prepare the data for the email content generation
+            mess = ''
+            start = True
+            for event in task_event_list:
+                event_time = 'at %d h %d' % event.time[3:5]  # hour, min
+                if start:
+                    mess += 'task[%s]' % event.task.getId()
+                    mess += '\nstarted %s' % event_time
+                    mess += '\n'
+                else:
+                    mess += 'ended %s' % event_time
+                    if event.success:
+                        mess += '\nsuccess'
+                    else:
+                        mess += '\nFAILURE'
+                    mess += '\n\n'
+                start = not start
+
+            self.data.update(notify_body=mess)
+            NotifyEmail.notify(self, None, 'task event notification')
+        else:
+            self.env.log.debug('no recipient for task event, aborting')
+
+    def send(self, torcpts, ccrcpts):
+        return NotifyEmail.send(self, torcpts, ccrcpts)
+
+
+class StartTaskEvent():
+    """
+    Store the event of a task start
+    """
+    def __init__(self, task):
+        self.task = task
+        self.time = localtime(time())
+
+
+class EndTaskEvent():
+    """
+    Store the event of a task end
+    """
+    def __init__(self, task, success):
+        self.task = task
+        self.time = localtime(time())
+        self.success = success
+
+
 class NotificationEmailTaskEvent(Component, ITaskEventListener, ITemplateProvider):
     """
     This task listener send notification mail about task event.
     """
-    implements(ITaskEventListener)
-
-    class NotifyEmailTaskEvent(NotifyEmail):
-
-            template_name = "notify_task_event_template.txt"
-
-            def __init__(self, env):
-                NotifyEmail.__init__(self, env)
-                self.cronconf = CronConfig(self.env)
-
-            def get_recipients(self, resid):
-                """
-                Return the recipients as defined in trac.ini.
-                """
-                reclist = self.cronconf.get_email_notifier_task_recipient_list()
-                return (reclist, [])
-
-            def notifyTaskEvent(self, task_event_list):
-                """
-                Send task event by mail if recipients is defined in trac.ini
-                """
-                self.env.log.debug("notifying task event...")
-                if self.cronconf.get_email_notifier_task_recipient():
-                    # prepare the data for the email content generation
-                    mess = ""
-                    start = True
-                    for event in task_event_list:
-                        if start:
-                            mess = mess + "task[%s]" % (event.task.getId(),)
-                            mess = mess + "\nstarted at %d h %d" % (event.time.tm_hour, event.time.tm_min)
-                            mess = mess + "\n"
-                        else:
-                            mess = mess + "ended at %d h %d" % (event.time.tm_hour, event.time.tm_min)
-                            if (event.success):
-                                mess = mess + "\nsuccess"
-                            else:
-                                mess = mess + "\nFAILURE"
-                            mess = mess + "\n\n"
-                        start = not start
-
-                    self.data.update({"notify_body": mess})
-                    NotifyEmail.notify(self, None, "task event notification")
-                else:
-                    self.env.log.debug("no recipient for task event, aborting")
-
-            def send(self, torcpts, ccrcpts):
-                return NotifyEmail.send(self, torcpts, ccrcpts)
+    implements(ITaskEventListener, ITemplateProvider)
 
     def __init__(self):
         self.cronconf = CronConfig(self.env)
         self.task_count = 0
         self.notifier = None
 
-    class StartTaskEvent():
-        """
-        Store the event of a task start
-        """
-        def __init__(self, task):
-            self.task = task
-            self.time = localtime(time())
-
-    class EndTaskEvent():
-        """
-        Store the event of a task end
-        """
-        def __init__(self, task, success):
-            self.task = task
-            self.time = localtime(time())
-            self.success = success
-
     def get_htdocs_dirs(self):
         return []
 
         called by the core system when the task is triggered,
         just before the waek_up method is called
         """
-        self.task_event_buffer.append(NotificationEmailTaskEvent.StartTaskEvent(task))
+        self.task_event_buffer.append(StartTaskEvent(task))
         self.task_count = self.task_count + 1
 
     def onEndTask(self, task, success):
             self.task_event_buffer.pop()
             self.task_count -= 1
             return
-        self.task_event_buffer.append(NotificationEmailTaskEvent.EndTaskEvent(task, success))
+
+        self.task_event_buffer.append(EndTaskEvent(task, success))
         # if the buffer reach the count then we notify
         if (self.task_count >= self.cronconf.get_email_notifier_task_limit()):
             # send the mail
             if not self.notifier:
-                self.notifier = NotificationEmailTaskEvent.NotifyEmailTaskEvent(self.env)
+                self.notifier = NotifyEmailTaskEvent(self.env)
             self.notifier.notifyTaskEvent(self.task_event_buffer)
 
             # reset task event buffer
     def getDescription(self):
         return self.__doc__
 
+
 class HistoryTaskEvent(Component, ITaskEventListener):
     """
-    This task event listener catch task execution to fill all History store in its environment
+    This task event listener catch task execution to fill all History store
+    in its environment
     """
-
     implements(ITaskEventListener)
 
     history_store_list = ExtensionPoint(IHistoryTaskExecutionStore)
         """
         return the id of the listener. It is used in trac.ini
         """
-        return "history_task_event"
+        return 'history_task_event'
 
     def getDescription(self):
         return self.__doc__
 
     def _notify_history(self):
-        for historyStore in self.history_store_list:
-            historyStore.addExecution(self.task, self.start, self.end, self.success)
+        for hist in self.history_store_list:
+            hist.addExecution(self.task, self.start, self.end, self.success)

tests/test_notification_email_task_event.py

+# -*- coding: utf-8 -*-
+import textwrap
+
+import pytest
+
+from traccron.listener import NotifyEmailTaskEvent
+
+
+def pytest_funcarg__task(request, component):
+    task = component['heart_beat_task']
+    return task
+
+def pytest_funcarg__notification_email_task_event(request, component):
+    notification_email_task_event = component['notification_email_task_event']
+    return notification_email_task_event
+
+
+def test_notify_email_task_event_basic(env, task):
+    from traccron.listener import StartTaskEvent, EndTaskEvent
+    # task event settings
+    event = NotifyEmailTaskEvent(env)
+    key = event.cronconf.email_notifier_task_recipient_key
+    event.cronconf.set_value(key, 'admin')
+
+    # notify recipients
+    assert ([u'admin'], []) == event.get_recipients(1)
+
+    # notify message body
+    start_task_event = StartTaskEvent(task)
+    end_task_event = EndTaskEvent(task, True)
+    event.notifyTaskEvent([start_task_event, end_task_event])
+    msg_tmpl = """\
+        task[%s]
+        started at %s h %s
+        ended at %s h %s
+        success\n\n"""
+    expected = textwrap.dedent(msg_tmpl % (
+        task.getId(),
+        start_task_event.time.tm_hour,
+        start_task_event.time.tm_min,
+        end_task_event.time.tm_hour,
+        end_task_event.time.tm_min
+    ))
+    assert expected == event.data['notify_body']
+
+
+def test_notification_email_task_event_basic(notification_email_task_event,
+                                             task):
+    event = notification_email_task_event  # just an alias
+    id_ = event.cronconf.EMAIL_NOTIFIER_TASK_BASEKEY
+    assert id_ == event.getId()
+
+    # notify recipients settings
+    key = event.cronconf.email_notifier_task_recipient_key
+    event.cronconf.set_value(key, 'myuser')
+    recipients = event.cronconf.get_email_notifier_task_recipient_list()
+    assert [u'myuser'] == recipients
+
+    # operate tasks
+    event.onStartTask(task)
+    assert 1 == event.task_count
+    event.onEndTask(task, False)
+    assert 0 == event.task_count
+    assert 0 == len(event.task_event_buffer)
+
+    # notify message body
+    assert isinstance(event.notifier, NotifyEmailTaskEvent)
+    notify_body = event.notifier.data.get('notify_body') or ''
+    assert notify_body.startswith('task[%s]' % task.getId())
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.