-
-
Notifications
You must be signed in to change notification settings - Fork 4.4k
/
test_notification.py
150 lines (106 loc) · 4.44 KB
/
test_notification.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
# encoding: utf-8
"""This file contains unittests for the notification.py module."""
#-----------------------------------------------------------------------------
# Copyright (C) 2008-2011 The IPython Development Team
#
# Distributed under the terms of the BSD License. The full license is
# in the file COPYING, distributed as part of this software.
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
import unittest
from IPython.utils.notification import shared_center
#-----------------------------------------------------------------------------
# Support Classes
#-----------------------------------------------------------------------------
class Observer(object):
def __init__(self, expected_ntype, expected_sender,
center=shared_center, *args, **kwargs):
super(Observer, self).__init__()
self.expected_ntype = expected_ntype
self.expected_sender = expected_sender
self.expected_args = args
self.expected_kwargs = kwargs
self.recieved = False
center.add_observer(self.callback,
self.expected_ntype,
self.expected_sender)
def callback(self, ntype, sender, *args, **kwargs):
assert(ntype == self.expected_ntype or
self.expected_ntype == None)
assert(sender == self.expected_sender or
self.expected_sender == None)
assert(args == self.expected_args)
assert(kwargs == self.expected_kwargs)
self.recieved = True
def verify(self):
assert(self.recieved)
def reset(self):
self.recieved = False
class Notifier(object):
def __init__(self, ntype, **kwargs):
super(Notifier, self).__init__()
self.ntype = ntype
self.kwargs = kwargs
def post(self, center=shared_center):
center.post_notification(self.ntype, self,
**self.kwargs)
#-----------------------------------------------------------------------------
# Tests
#-----------------------------------------------------------------------------
class NotificationTests(unittest.TestCase):
def tearDown(self):
shared_center.remove_all_observers()
def test_notification_delivered(self):
"""Test that notifications are delivered"""
expected_ntype = 'EXPECTED_TYPE'
sender = Notifier(expected_ntype)
observer = Observer(expected_ntype, sender)
sender.post()
observer.verify()
def test_type_specificity(self):
"""Test that observers are registered by type"""
expected_ntype = 1
unexpected_ntype = "UNEXPECTED_TYPE"
sender = Notifier(expected_ntype)
unexpected_sender = Notifier(unexpected_ntype)
observer = Observer(expected_ntype, sender)
sender.post()
unexpected_sender.post()
observer.verify()
def test_sender_specificity(self):
"""Test that observers are registered by sender"""
expected_ntype = "EXPECTED_TYPE"
sender1 = Notifier(expected_ntype)
sender2 = Notifier(expected_ntype)
observer = Observer(expected_ntype, sender1)
sender1.post()
sender2.post()
observer.verify()
def test_remove_all_observers(self):
"""White-box test for remove_all_observers"""
for i in xrange(10):
Observer('TYPE', None, center=shared_center)
self.assert_(len(shared_center.observers[('TYPE',None)]) >= 10,
"observers registered")
shared_center.remove_all_observers()
self.assert_(len(shared_center.observers) == 0, "observers removed")
def test_any_sender(self):
expected_ntype = "EXPECTED_TYPE"
sender1 = Notifier(expected_ntype)
sender2 = Notifier(expected_ntype)
observer = Observer(expected_ntype, None)
sender1.post()
observer.verify()
observer.reset()
sender2.post()
observer.verify()
def test_post_performance(self):
"""Test that post_notification, even with many registered irrelevant
observers is fast"""
for i in xrange(10):
Observer("UNRELATED_TYPE", None)
o = Observer('EXPECTED_TYPE', None)
shared_center.post_notification('EXPECTED_TYPE', self)
o.verify()