test_notification.py
161 lines
| 4.9 KiB
| text/x-python
|
PythonLexer
Barry Wark
|
r1410 | # encoding: utf-8 | ||
"""This file contains unittests for the notification.py module.""" | ||||
#----------------------------------------------------------------------------- | ||||
Brian Granger
|
r1963 | # Copyright (C) 2008-2009 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. | ||||
Barry Wark
|
r1410 | #----------------------------------------------------------------------------- | ||
Brian Granger
|
r1963 | |||
Barry Wark
|
r1410 | #----------------------------------------------------------------------------- | ||
Brian Granger
|
r1963 | # Imports | ||
Barry Wark
|
r1410 | #----------------------------------------------------------------------------- | ||
Brian Granger
|
r1960 | # Tell nose to skip this module | ||
__test__ = {} | ||||
Brian Granger
|
r1963 | from twisted.trial import unittest | ||
Barry Wark
|
r1411 | import IPython.kernel.core.notification as notification | ||
Barry Wark
|
r1410 | |||
Brian Granger
|
r1963 | #----------------------------------------------------------------------------- | ||
# Support Classes | ||||
#----------------------------------------------------------------------------- | ||||
Barry Wark
|
r1410 | |||
class Observer(object): | ||||
"""docstring for Observer""" | ||||
Barry Wark
|
r1411 | def __init__(self, expectedType, expectedSender, | ||
center=notification.sharedCenter, **kwargs): | ||||
Barry Wark
|
r1410 | super(Observer, self).__init__() | ||
self.expectedType = expectedType | ||||
self.expectedSender = expectedSender | ||||
self.expectedKwArgs = kwargs | ||||
self.recieved = False | ||||
Barry Wark
|
r1411 | center.add_observer(self.callback, | ||
self.expectedType, | ||||
self.expectedSender) | ||||
Barry Wark
|
r1410 | |||
def callback(self, theType, sender, args={}): | ||||
"""callback""" | ||||
Barry Wark
|
r1411 | assert(theType == self.expectedType or | ||
self.expectedType == None) | ||||
assert(sender == self.expectedSender or | ||||
self.expectedSender == None) | ||||
Barry Wark
|
r1410 | assert(args == self.expectedKwArgs) | ||
self.recieved = True | ||||
def verify(self): | ||||
"""verify""" | ||||
assert(self.recieved) | ||||
Barry Wark
|
r1411 | def reset(self): | ||
"""reset""" | ||||
self.recieved = False | ||||
Barry Wark
|
r1410 | |||
class Notifier(object): | ||||
"""docstring for Notifier""" | ||||
def __init__(self, theType, **kwargs): | ||||
super(Notifier, self).__init__() | ||||
self.theType = theType | ||||
self.kwargs = kwargs | ||||
Barry Wark
|
r1411 | def post(self, center=notification.sharedCenter): | ||
Barry Wark
|
r1410 | """fire""" | ||
center.post_notification(self.theType, self, | ||||
**self.kwargs) | ||||
Brian Granger
|
r1963 | #----------------------------------------------------------------------------- | ||
# Tests | ||||
#----------------------------------------------------------------------------- | ||||
Barry Wark
|
r1410 | |||
Barry Wark
|
r1411 | class NotificationTests(unittest.TestCase): | ||
"""docstring for NotificationTests""" | ||||
Barry Wark
|
r1410 | |||
Barry Wark
|
r1411 | def tearDown(self): | ||
notification.sharedCenter.remove_all_observers() | ||||
Barry Wark
|
r1410 | |||
Barry Wark
|
r1411 | def test_notification_delivered(self): | ||
"""Test that notifications are delivered""" | ||||
expectedType = 'EXPECTED_TYPE' | ||||
sender = Notifier(expectedType) | ||||
observer = Observer(expectedType, sender) | ||||
sender.post() | ||||
observer.verify() | ||||
Barry Wark
|
r1410 | |||
Barry Wark
|
r1411 | def test_type_specificity(self): | ||
"""Test that observers are registered by type""" | ||||
expectedType = 1 | ||||
unexpectedType = "UNEXPECTED_TYPE" | ||||
sender = Notifier(expectedType) | ||||
unexpectedSender = Notifier(unexpectedType) | ||||
observer = Observer(expectedType, sender) | ||||
Barry Wark
|
r1410 | |||
Barry Wark
|
r1411 | sender.post() | ||
unexpectedSender.post() | ||||
Barry Wark
|
r1410 | |||
Barry Wark
|
r1411 | observer.verify() | ||
Barry Wark
|
r1410 | |||
Barry Wark
|
r1411 | def test_sender_specificity(self): | ||
"""Test that observers are registered by sender""" | ||||
expectedType = "EXPECTED_TYPE" | ||||
sender1 = Notifier(expectedType) | ||||
sender2 = Notifier(expectedType) | ||||
observer = Observer(expectedType, 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=notification.sharedCenter) | ||||
self.assert_(len(notification.sharedCenter.observers[('TYPE',None)]) >= 10, | ||||
"observers registered") | ||||
notification.sharedCenter.remove_all_observers() | ||||
self.assert_(len(notification.sharedCenter.observers) == 0, "observers removed") | ||||
Brian Granger
|
r1963 | |||
Barry Wark
|
r1411 | def test_any_sender(self): | ||
"""test_any_sender""" | ||||
expectedType = "EXPECTED_TYPE" | ||||
sender1 = Notifier(expectedType) | ||||
sender2 = Notifier(expectedType) | ||||
observer = Observer(expectedType, None) | ||||
sender1.post() | ||||
observer.verify() | ||||
observer.reset() | ||||
sender2.post() | ||||
observer.verify() | ||||
Brian Granger
|
r1963 | |||
Barry Wark
|
r1411 | 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) | ||||
notification.sharedCenter.post_notification('EXPECTED_TYPE', self) | ||||
o.verify() | ||||
Brian Granger
|
r1963 | |||