Tetsuya Morimoto avatar Tetsuya Morimoto committed cfaea4d

added UnreachableMilestoneTask component tests and refactoring
dropped W601 from pep8ignore list

Comments (0)

Files changed (5)

 tag_svn_revision = true
 
 [pytest]
-pep8ignore = E128 E302 E501 E701 W601
+pep8ignore = E128 E302 E501 E701

src/traccron/task.py

 from trac.ticket.model import Ticket
 from trac.core import Component, implements
 from trac.notification import NotifyEmail
-from trac.util.datefmt import utc, to_utimestamp
+from trac.util.datefmt import utc, from_utimestamp, to_utimestamp
 from trac.web.chrome import ITemplateProvider
 from traccron.api import ICronTask
 from traccron.core import CronConfig
         Send a digest mail to ticket owner to remind him of those
         sleeping tickets
         """
-        for owner in tiketsByOwner.keys():
+        for owner in tiketsByOwner:
             # prepare the data for the email content generation
             self.data.update({
                 'ticket_count': len(tiketsByOwner[owner]),
         Send a digest mail to the reporter to remind them
         of those orphaned tickets
         """
-        for reporter in tiketsByReporter.keys():
+        for reporter in tiketsByReporter:
             # prepare the data for the email content generation
             self.data.update({
                 'ticket_count': len(tiketsByReporter[reporter]),
     """
     Remind user about sleeping ticket they are assigned to.
     """
-
     implements(ICronTask, ITemplateProvider)
 
     select_assigned_ticket = """
         return self.__doc__
 
 
+class BaseTicketNotification(NotifyEmail):
+
+    def __init__(self, env, milestone):
+        NotifyEmail.__init__(self, env)
+        self.milestone = milestone
+
+    def get_recipients(self, recipient):
+        return ([recipient], [])
+
+    def send(self, torcpts, ccrcpts):
+        return NotifyEmail.send(self, torcpts, ccrcpts)
+
+    def populate_unreachable_tickets_data(self, tickets):
+        # we are not called if there is no tickets
+        self.data['milestone'] = tickets[0]['milestone']
+        due_date = from_utimestamp(tickets[0]['due'])
+        self.data['due_date'] = due_date.strftime('%Y-%m-%d')
+
+        tickets_list = ''
+        for ticket in tickets:
+            tickets_list += ticket['summary'] + '\n'
+            tickets_list += self.env.abs_href.ticket(ticket['ticket']) + '\n'
+            tickets_list += '\n'
+
+        self.data['tickets_list'] = tickets_list
+
+    def notify_opened_ticket(self, recipient, tickets):
+        """
+        Send a digest mail to recipients (e.g. ticket owner and reporter)
+        about ticket still opened
+        """
+        self.populate_unreachable_tickets_data(tickets)
+        subject = 'Milestone %s with still opened ticket' % self.milestone
+        NotifyEmail.notify(self, recipient, subject)
+        self.env.log.debug('notify opened ticket: %s, %s, %s' % (
+                           recipient, self.milestone, self.__class__.__name__))
+
+
+class ReporterOpenedTicketNotification(BaseTicketNotification):
+    """
+    Notify reporter about an opened ticket in a near milestone
+    """
+    template_name = "opened_ticket_for_reporter_template.txt"
+
+    def __init__(self, env, milestone):
+        BaseTicketNotification.__init__(self, env, milestone)
+
+
+class OwnerOpenedTicketNotification(BaseTicketNotification):
+    """
+    Notify owner about an opened ticket in a near milestone
+    """
+    template_name = "opened_ticket_for_owner_template.txt"
+
+    def __init__(self, env, milestone):
+        BaseTicketNotification.__init__(self, env, milestone)
+
+
+class UnreachableMilestoneNotification(BaseTicketNotification):
+    """
+    Notify the specified person (ex: admin, release manager) that a milestone
+    is about to closed but there still are opened ticket
+    """
+    template_name = 'unreachable_milestone_template.txt'
+
+    def __init__(self, env, milestone):
+        BaseTicketNotification.__init__(self, env, milestone)
+        self.cronconf = CronConfig(self.env)
+
+    def get_recipients(self, milestone):
+        reclist = self.cronconf.get_unreachable_milestone_task_recipient_list()
+        return (reclist, [])
+
+    def notify_unreachable_milestone(self, tickets):
+        """
+        Send a digest mail listing all tickets still opened in the milestone
+        """
+        self.populate_unreachable_tickets_data(tickets)
+        subject = "Milestone %s still has opened ticket" % self.milestone
+        NotifyEmail.notify(self, self.milestone, subject)
+        self.env.log.debug('notify unreachable milestone: %s, %s' % (
+                           self.milestone, self.__class__.__name__))
+
+
 class UnreachableMilestoneTask(Component, ICronTask, ITemplateProvider):
     """
     Send notification about near milestone with opened ticked
     """
+    implements(ICronTask, ITemplateProvider)
 
-    implements(ICronTask, ITemplateProvider)
+    select_near_milestone_ticket = """
+        SELECT t.id , t.owner, t.reporter, t.milestone, t.summary, m.due
+        FROM ticket t, milestone m
+        WHERE t.milestone = m.name
+        AND m.due < %s
+    """
 
     def __init__(self):
         self.cronconf = CronConfig(self.env)
         from pkg_resources import resource_filename
         return [resource_filename(__name__, 'templates')]
 
+    def _set_recipient_data(self, dico, recipient, milestone, ticket_data):
+        dico.setdefault(recipient, {}).__setitem__(milestone, [ticket_data])
+        dico[recipient].setdefault(milestone, []).append(ticket_data)
+
+    def remind_unreachable_recipients(self, dico):
+        """
+        Send notification for each milestone
+        """
+        for milestone in dico:
+            notifier = UnreachableMilestoneNotification(self.env, milestone)
+            notifier.notify_unreachable_milestone(dico[milestone])
+
+    def remind_recipient(self, dico, klass):
+        """
+        Send notification for owner or reporter
+        """
+        for recipient in dico:
+            _dico = dico[recipient]
+            for milestone in _dico:
+                notifier = klass(self.env, milestone)
+                notifier.notify_opened_ticket(recipient, _dico[milestone])
+
     def wake_up(self, *args):
         delay = 3
         if len(args) > 0:
             delay = int(args[0])
 
-        class BaseTicketNotification(NotifyEmail):
-
-            def __init__(self, env, milestone):
-                NotifyEmail.__init__(self, env)
-                self.milestone = milestone
-
-            def populate_unreachable_tickets_data(self, tickets):
-                self.data['milestone'] = tickets[0]['milestone']  # we are not called if there is no tickets
-                due_date = localtime(tickets[0]['due'])
-                self.data['due_date'] = "%d-%d-%d" % (due_date.tm_mon, due_date.tm_mday, due_date.tm_year)
-                tickets_list = ""
-                for ticket in tickets:
-                    tickets_list += ticket['summary'] + "\n"
-                    tickets_list += self.env.abs_href.ticket(ticket['ticket']) + "\n"
-                    tickets_list += "\n"
-
-                self.data['tickets_list'] = tickets_list
-
-        class ReporterOpenedTicketNotification(BaseTicketNotification):
-            """
-            Notify reporter about an opened ticket in
-            a near milestone
-            """
-            template_name = "opened_ticket_for_reporter_template.txt"
-
-            def __init__(self, env, milestone):
-                BaseTicketNotification.__init__(self, env, milestone)
-
-            def get_recipients(self, reporter):
-                return ([reporter], [])
-
-            def notify_opened_ticket(self, reporter, tickets):
-                """
-                Send a digest mail to ticket owner and reporter
-                about ticket still opened
-                """
-
-                self.populate_unreachable_tickets_data(tickets)
-                NotifyEmail.notify(self, reporter, "Milestone %s with still opened ticket" % self.milestone)
-
-            def send(self, torcpts, ccrcpts):
-                return NotifyEmail.send(self, torcpts, ccrcpts)
-
-        class OwnerOpenedTicketNotification(BaseTicketNotification):
-            """
-            Notify owner about an opened ticket in
-            a near milestone
-            """
-            template_name = "opened_ticket_for_owner_template.txt"
-
-            def __init__(self, env, milestone):
-                BaseTicketNotification.__init__(self, env, milestone)
-
-            def get_recipients(self, owner):
-                return ([owner], [])
-
-            def notify_opened_ticket(self, owner, tickets):
-                """
-                Send a digest mail to ticket owner
-                about ticket still opened
-                """
-
-                self.populate_unreachable_tickets_data(tickets)
-                NotifyEmail.notify(self, owner, "Milestone %s still has opened ticket" % self.milestone)
-
-            def send(self, torcpts, ccrcpts):
-                return NotifyEmail.send(self, torcpts, ccrcpts)
-
-        class UnreachableMilestoneNotification(BaseTicketNotification):
-            """
-            Notify the specified person (ex: admin, release manager) that a milestone
-            is about to closed but there still are opened ticket
-            """
-
-            template_name = "unreachable_milestone_template.txt"
-
-            def __init__(self, env, milestone):
-                BaseTicketNotification.__init__(self, env, milestone)
-                self.cronconf = CronConfig(self.env)
-
-            def get_recipients(self, milestone):
-                reclist = self.cronconf.get_unreachable_milestone_task_recipient_list()
-                return (reclist, [])
-
-            def notify_unreachable_milestone(self, tickets):
-                """
-                Send a digest mail listing all tickets still opened in the milestone
-                """
-                self.populate_unreachable_tickets_data(tickets)
-
-                NotifyEmail.notify(self, self.milestone, "Milestone %s still has opened ticket" % self.milestone)
-
-            def send(self, torcpts, ccrcpts):
-                return NotifyEmail.send(self, torcpts, ccrcpts)
-
         # look opened ticket in near milestone
         db = self.env.get_db_cnx()
         cursor = db.cursor()
         # select ticket whom milestone are due in less than specified delay
-        cursor.execute("""
-                SELECT t.id , t.owner, t.reporter, t.milestone, t.summary, m.due  FROM ticket t, milestone m
-                WHERE  t.milestone = m.name
-                AND    m.due < %s
-            """, (time() + delay * 24 * 60 * 60,))
-        dico = {}
-        dico_reporter = {}
-        dico_owner = {}
+        delay_time = datetime.now(utc) + timedelta(days=delay)
+        cursor.execute(self.select_near_milestone_ticket,
+                       (to_utimestamp(delay_time),))
+
+        dico, dico_owner, dico_reporter = {}, {}, {}
         for ticket, owner, reporter, milestone, summary, due in cursor:
-            self.env.log.info("warning ticket %d will probably miss its milestone %s" % (ticket, milestone))
+            msg = 'warning ticket %d will probably miss its milestone %s' % (
+                  ticket, milestone)
+            self.env.log.info(msg)
             ticket_data = {
                 'ticket': ticket,
                 'owner': owner,
                 'summary': summary,
                 'due': due
             }
-            if dico.has_key(milestone):
-                dico[milestone].append(ticket_data)
-            else:
-                dico[milestone] = [ticket_data]
+            dico.setdefault(milestone, []).append(ticket_data)
+            self._set_recipient_data(dico_owner, owner, milestone, ticket_data)
+            self._set_recipient_data(dico_reporter, reporter, milestone,
+                                     ticket_data)
 
-            if dico_owner.has_key(owner):
-                if dico_owner[owner].has_key(milestone):
-                    dico_owner[owner][milestone].append(ticket_data)
-                else:
-                    dico_owner[owner][milestone] = [ticket_data]
-            else:
-                dico_owner[owner] = {milestone: [ticket_data]}
-
-            if dico_reporter.has_key(reporter):
-                if dico_reporter[reporter].has_key(milestone):
-                    dico_reporter[reporter][milestone].append(ticket_data)
-                else:
-                    dico_reporter[reporter][milestone] = [ticket_data]
-            else:
-                dico_reporter[reporter] = {milestone: [ticket_data]}
-
-        for milestone in dico.keys():
-            # send notification for each milestone
-            UnreachableMilestoneNotification(self.env, milestone).notify_unreachable_milestone(dico[milestone])
-        for owner in dico_owner.keys():
-            # for this owner
-            _dico_owner = dico_owner[owner]
-            for milestone in _dico_owner.keys():
-                OwnerOpenedTicketNotification(self.env, milestone).notify_opened_ticket(owner, _dico_owner[milestone])
-        for reporter in dico_reporter.keys():
-            _dico_reporter = dico_reporter[reporter]
-            for milestone in _dico_reporter.keys():
-                ReporterOpenedTicketNotification(self.env, milestone).notify_opened_ticket(reporter, _dico_reporter[milestone])
+        self.remind_unreachable_recipients(dico)
+        self.remind_recipient(dico_owner, OwnerOpenedTicketNotification)
+        self.remind_recipient(dico_reporter, ReporterOpenedTicketNotification)
 
     def getId(self):
         return self.cronconf.UNREACHABLE_MILESTONE_TASK_BASEKEY

tests/test_sleeping_ticket_reminder_task.py

 from datetime import datetime, timedelta
 
 import pytest
-from utils import create_orphaned_ticket, create_ticket, create_tickets
+from utils import create_orphaned_ticket, create_new_tickets
 from utils import has_log_message
 
 from trac.util.datefmt import utc
 
     # create tickets
     when = datetime.now(utc) - timedelta(days=5)
-    t1 = create_tickets(env, 3, owner='sleeper1', when=when)
+    t1 = create_new_tickets(env, 3, owner='sleeper1', when=when)
     o1 = create_orphaned_ticket(env, when=when)
 
     sleeping_ticket_reminder_task.wake_up()
     # create tickets
     ticket_subtract_days = 5
     when = datetime.now(utc) - timedelta(days=ticket_subtract_days)
-    t1 = create_tickets(env, 3, owner='sleeper1', when=when)
+    t1 = create_new_tickets(env, 3, owner='sleeper1', when=when)
     o1 = create_orphaned_ticket(env, when=when)
 
     sleeping_ticket_reminder_task.wake_up(*args)

tests/test_unreachable_milestone_task.py

+# -*- coding: utf-8 -*-
+import logging
+from datetime import datetime, timedelta
+
+import pytest
+from utils import create_milestone_ticket
+from utils import has_log_message
+
+from trac.util.datefmt import utc
+
+
+def pytest_funcarg__unreachable_milestone_task(request, component):
+    unreachable_milestone_task = component['unreachable_milestone_task']
+    return unreachable_milestone_task
+
+
+def test_unreachable_milestone_task_getId(unreachable_milestone_task):
+    from traccron.core import CronConfig
+    task = unreachable_milestone_task
+    assert CronConfig.UNREACHABLE_MILESTONE_TASK_BASEKEY == task.getId()
+
+
+def test_unreachable_milestone_task_wake_up(unreachable_milestone_task,
+                                            caplog):
+    env = unreachable_milestone_task.env
+    caplog.setLevel(logging.DEBUG, logger=env.log.name)
+
+    # TODO: should be more simple
+    # create tickets
+    duedate = datetime.now(utc)
+    duedate_plus_5days = duedate + timedelta(days=5)
+    create_milestone_ticket(env, 'ms1', duedate, 'owner1', 'repo1')
+    create_milestone_ticket(env, 'ms1', duedate, 'owner1', 'repo1'),
+    create_milestone_ticket(env, 'ms1', duedate, 'owner2', 'repo2'),
+    create_milestone_ticket(env, 'ms2', duedate_plus_5days, 'owner1', 'repo1')
+    create_milestone_ticket(env, 'ms3', duedate, 'owner1', 'repo1')
+
+    unreachable_milestone_task.wake_up()
+    expected_messages = [
+        'applying config',
+        'stop existing ticker',
+        'ticker is disabled',
+        'action controllers for ticket workflow',
+        "Creating new milestone 'ms1'",
+        "Creating new milestone 'ms2'",
+        "Creating new milestone 'ms3'",
+        'warning ticket 1 will probably miss its milestone ms1',
+        'warning ticket 2 will probably miss its milestone ms1',
+        'warning ticket 3 will probably miss its milestone ms1',
+        'warning ticket 5 will probably miss its milestone ms3',
+        'notify unreachable milestone: ms3, UnreachableMilestoneNotification',
+        'notify unreachable milestone: ms1, UnreachableMilestoneNotification',
+        'notify opened ticket: owner1, ms3, OwnerOpenedTicketNotification',
+        'notify opened ticket: owner1, ms1, OwnerOpenedTicketNotification',
+        'notify opened ticket: owner2, ms1, OwnerOpenedTicketNotification',
+        'notify opened ticket: repo1, ms3, ReporterOpenedTicketNotification',
+        'notify opened ticket: repo1, ms1, ReporterOpenedTicketNotification',
+        'notify opened ticket: repo2, ms1, ReporterOpenedTicketNotification',
+    ]
+    assert has_log_message(caplog, expected_messages)
 # -*- coding: utf-8 -*-
 import sys
-import time
+from datetime import datetime, timedelta
 
-from trac.ticket.model import Ticket
+from trac.ticket.api import TicketSystem
+from trac.ticket.model import Milestone, Ticket
 
 _IGNORE_UNLOGGED_MESSAGES = [
     'stop existing ticker',
         return attrgetter(*args)
 
 
-def create_ticket(env, owner='user1', when=None):
+def compat_itemgetter(*args):
+    from operator import itemgetter
+    if sys.version_info[:2] == (2, 4):
+        return lambda obj: tuple(itemgetter(item)(obj) for item in args)
+    else:  # version >= 2.5
+        return itemgetter(*args)
+
+
+def update_value(field, value):
+    def _update_value(f):
+        def __update_value(*args, **kwargs):
+            ticket = f(*args, **kwargs)
+            ticket[field] = value
+            when = kwargs.get('when')
+            if when is not None:
+                changetime = ticket['changetime']
+                if changetime is not None:
+                    when = changetime
+                when += timedelta(seconds=1)
+            assert ticket.save_changes(when=when)
+            return ticket
+        return __update_value
+    return _update_value
+
+
+def create_ticket(env, **kwargs):
+    ticket_fields = map(compat_itemgetter('name'),
+                        TicketSystem(env).get_ticket_fields())
     t = Ticket(env)
-    t['summary'] = 'test'
-    t['description'] = 'ticket for test'
-    t['owner'] = owner
-    t.insert(when=when)
+    t['summary'] = '%s - test ticket' % datetime.now().strftime('%H:%M:%S.%f')
+    t['description'] = 'this ticket is for testing'
+    for field in kwargs:
+        if field in ticket_fields:
+            t[field] = kwargs.get(field)
+    t.insert(when=kwargs.get('when'))
     assert t.exists
-    t['status'] = 'new'
-    assert t.save_changes(when=when)
     return t
 
-def create_tickets(env, num, owner='user1', when=None):
-    return [create_ticket(env, owner, when) for _ in range(num)]
+
+@update_value('status', 'new')
+def create_new_ticket(env, **kwargs):
+    return create_ticket(env, **kwargs)
+
+
+def create_new_tickets(env, num, owner='user1', when=None):
+    return [create_new_ticket(env, owner=owner, when=when) for _ in range(num)]
+
 
 def create_orphaned_ticket(env, reporter='user1', when=None):
     t = Ticket(env)
     t['reporter'] = reporter
     t.insert(when=when)
     assert t.exists
+    # update manually since 'orphaned' means it is never changed
     db = env.get_db_cnx()
     cursor = db.cursor()
     cursor.execute("update ticket set status='new' where id=%s", (t.id,))
     db.commit()
     return t
+
+
+def get_or_create_milestone(env, name='ms1', duedate=0):
+    from trac.resource import ResourceNotFound
+    try:
+        m = Milestone(env, name=name)
+    except ResourceNotFound:
+        m = Milestone(env)
+        m.name = name
+        m.due = duedate
+        m.insert()
+        assert m.exists
+    return m
+
+
+def create_milestone_ticket(env, name, duedate, owner, reporter):
+    @update_value('milestone', name)
+    def create_ticket_with_milestone(env, **kwargs):
+        return create_ticket(env, **kwargs)
+
+    m = get_or_create_milestone(env, name=name, duedate=duedate)
+    t = create_ticket_with_milestone(env, owner=owner, reporter=reporter)
+    return m, t
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.