← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/lp/bugs/tests/test_bugnotification.py

  • Committer: Colin Watson
  • Date: 2011-08-19 00:25:11 UTC
  • mfrom: (7675.1045.728 db-devel)
  • mto: This revision was merged to the branch mainline in revision 13909.
  • Revision ID: cjwatson@canonical.com-20110819002511-0x8hrqs1ckiqk53g
merge db-devel

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/lp/bugs/tests/test_bugnotification.py
6
6
__metaclass__ = type
7
7
 
8
8
from itertools import chain
 
9
from datetime import datetime
 
10
import pytz
9
11
import transaction
10
12
import unittest
11
13
 
29
31
from lp.answers.tests.test_question_notifications import pop_questionemailjobs
30
32
from lp.bugs.interfaces.bugtask import (
31
33
    BugTaskStatus,
32
 
    IUpstreamBugTask,
 
34
    IBugTask,
33
35
    )
 
36
from lp.bugs.mail.bugnotificationrecipients import BugNotificationRecipients
34
37
from lp.bugs.model.bugnotification import (
35
38
    BugNotification,
36
39
    BugNotificationFilter,
37
40
    BugNotificationSet,
38
41
    )
39
42
from lp.bugs.model.bugsubscriptionfilter import BugSubscriptionFilterMute
 
43
from lp.services.messages.interfaces.message import IMessageSet
40
44
from lp.testing import (
41
45
    TestCaseWithFactory,
42
46
    person_logged_in,
134
138
        # Ensure that notifications are sent to subscribers of a
135
139
        # question linked to the expired bug.
136
140
        bugtask = self.bug.default_bugtask
137
 
        bugtask_before_modification = Snapshot(
138
 
            bugtask, providing=IUpstreamBugTask)
 
141
        bugtask_before_modification = Snapshot(bugtask, providing=IBugTask)
139
142
        bugtask.transitionToStatus(BugTaskStatus.EXPIRED, self.product.owner)
140
143
        bug_modified = ObjectModifiedEvent(
141
144
            bugtask, bugtask_before_modification, ["status"])
642
645
            {team.teamowner: [notification.recipients[0]],
643
646
             team: [notification.recipients[1]]},
644
647
            [notification]))
 
648
 
 
649
 
 
650
class TestGetDeferredNotifications(TestCaseWithFactory):
 
651
    """Test the getDeferredNotifications method."""
 
652
 
 
653
    layer = DatabaseFunctionalLayer
 
654
 
 
655
    def setUp(self):
 
656
        super(TestGetDeferredNotifications, self).setUp()
 
657
        self.bns = BugNotificationSet()
 
658
 
 
659
    def test_no_deferred_notifications(self):
 
660
        results = self.bns.getDeferredNotifications()
 
661
        self.assertEqual(0, results.count())
 
662
 
 
663
    def _make_deferred_notification(self):
 
664
        bug = self.factory.makeBug()
 
665
        empty_recipients = BugNotificationRecipients()
 
666
        message = getUtility(IMessageSet).fromText(
 
667
            'subject', 'a comment.', bug.owner,
 
668
            datecreated=datetime.now(pytz.UTC))
 
669
        self.bns.addNotification(
 
670
            bug, False, message, empty_recipients, None, deferred=True)
 
671
 
 
672
    def test_one_deferred_notification(self):
 
673
        self._make_deferred_notification()
 
674
        results = self.bns.getDeferredNotifications()
 
675
        self.assertEqual(1, results.count())
 
676
 
 
677
    def test_many_deferred_notification(self):
 
678
        num = 5
 
679
        for i in xrange(num):
 
680
            self._make_deferred_notification()
 
681
        results = self.bns.getDeferredNotifications()
 
682
        self.assertEqual(num, results.count())
 
683
 
 
684
    def test_destroy_notifications(self):
 
685
        self._make_deferred_notification()
 
686
        results = self.bns.getDeferredNotifications()
 
687
        self.assertEqual(1, results.count())
 
688
        notification = results[0]
 
689
        notification.destroySelf()
 
690
        results = self.bns.getDeferredNotifications()
 
691
        self.assertEqual(0, results.count())