Index: third_party/twisted_8_1/twisted/words/test/test_xishutil.py |
diff --git a/third_party/twisted_8_1/twisted/words/test/test_xishutil.py b/third_party/twisted_8_1/twisted/words/test/test_xishutil.py |
deleted file mode 100644 |
index d9e55d0a06138552402e6822afc6b4dd5fd64056..0000000000000000000000000000000000000000 |
--- a/third_party/twisted_8_1/twisted/words/test/test_xishutil.py |
+++ /dev/null |
@@ -1,307 +0,0 @@ |
-# Copyright (c) 2001-2008 Twisted Matrix Laboratories. |
-# See LICENSE for details. |
- |
-""" |
-Test cases for twisted.words.xish.utility |
-""" |
- |
-from twisted.trial import unittest |
- |
-from twisted.python.util import OrderedDict |
-from twisted.words.xish import utility |
-from twisted.words.xish.domish import Element |
-from twisted.words.xish.utility import EventDispatcher |
- |
-class CallbackTracker: |
- """ |
- Test helper for tracking callbacks. |
- |
- Increases a counter on each call to L{call} and stores the object |
- passed in the call. |
- """ |
- |
- def __init__(self): |
- self.called = 0 |
- self.obj = None |
- |
- |
- def call(self, obj): |
- self.called = self.called + 1 |
- self.obj = obj |
- |
- |
- |
-class OrderedCallbackTracker: |
- """ |
- Test helper for tracking callbacks and their order. |
- """ |
- |
- def __init__(self): |
- self.callList = [] |
- |
- |
- def call1(self, object): |
- self.callList.append(self.call1) |
- |
- |
- def call2(self, object): |
- self.callList.append(self.call2) |
- |
- |
- def call3(self, object): |
- self.callList.append(self.call3) |
- |
- |
- |
-class EventDispatcherTest(unittest.TestCase): |
- """ |
- Tests for L{EventDispatcher}. |
- """ |
- |
- def testStuff(self): |
- d = EventDispatcher() |
- cb1 = CallbackTracker() |
- cb2 = CallbackTracker() |
- cb3 = CallbackTracker() |
- |
- d.addObserver("/message/body", cb1.call) |
- d.addObserver("/message", cb1.call) |
- d.addObserver("/presence", cb2.call) |
- d.addObserver("//event/testevent", cb3.call) |
- |
- msg = Element(("ns", "message")) |
- msg.addElement("body") |
- |
- pres = Element(("ns", "presence")) |
- pres.addElement("presence") |
- |
- d.dispatch(msg) |
- self.assertEquals(cb1.called, 2) |
- self.assertEquals(cb1.obj, msg) |
- self.assertEquals(cb2.called, 0) |
- |
- d.dispatch(pres) |
- self.assertEquals(cb1.called, 2) |
- self.assertEquals(cb2.called, 1) |
- self.assertEquals(cb2.obj, pres) |
- self.assertEquals(cb3.called, 0) |
- |
- d.dispatch(d, "//event/testevent") |
- self.assertEquals(cb3.called, 1) |
- self.assertEquals(cb3.obj, d) |
- |
- d.removeObserver("/presence", cb2.call) |
- d.dispatch(pres) |
- self.assertEquals(cb2.called, 1) |
- |
- |
- def test_addObserverTwice(self): |
- """ |
- Test adding two observers for the same query. |
- |
- When the event is dispath both of the observers need to be called. |
- """ |
- d = EventDispatcher() |
- cb1 = CallbackTracker() |
- cb2 = CallbackTracker() |
- |
- d.addObserver("//event/testevent", cb1.call) |
- d.addObserver("//event/testevent", cb2.call) |
- d.dispatch(d, "//event/testevent") |
- |
- self.assertEquals(cb1.called, 1) |
- self.assertEquals(cb1.obj, d) |
- self.assertEquals(cb2.called, 1) |
- self.assertEquals(cb2.obj, d) |
- |
- |
- def test_addObserverInDispatch(self): |
- """ |
- Test for registration of an observer during dispatch. |
- """ |
- d = EventDispatcher() |
- msg = Element(("ns", "message")) |
- cb = CallbackTracker() |
- |
- def onMessage(_): |
- d.addObserver("/message", cb.call) |
- |
- d.addOnetimeObserver("/message", onMessage) |
- |
- d.dispatch(msg) |
- self.assertEquals(cb.called, 0) |
- |
- d.dispatch(msg) |
- self.assertEquals(cb.called, 1) |
- |
- d.dispatch(msg) |
- self.assertEquals(cb.called, 2) |
- |
- |
- def test_addOnetimeObserverInDispatch(self): |
- """ |
- Test for registration of a onetime observer during dispatch. |
- """ |
- d = EventDispatcher() |
- msg = Element(("ns", "message")) |
- cb = CallbackTracker() |
- |
- def onMessage(msg): |
- d.addOnetimeObserver("/message", cb.call) |
- |
- d.addOnetimeObserver("/message", onMessage) |
- |
- d.dispatch(msg) |
- self.assertEquals(cb.called, 0) |
- |
- d.dispatch(msg) |
- self.assertEquals(cb.called, 1) |
- |
- d.dispatch(msg) |
- self.assertEquals(cb.called, 1) |
- |
- |
- def testOnetimeDispatch(self): |
- d = EventDispatcher() |
- msg = Element(("ns", "message")) |
- cb = CallbackTracker() |
- |
- d.addOnetimeObserver("/message", cb.call) |
- d.dispatch(msg) |
- self.assertEquals(cb.called, 1) |
- d.dispatch(msg) |
- self.assertEquals(cb.called, 1) |
- |
- |
- def testDispatcherResult(self): |
- d = EventDispatcher() |
- msg = Element(("ns", "message")) |
- pres = Element(("ns", "presence")) |
- cb = CallbackTracker() |
- |
- d.addObserver("/presence", cb.call) |
- result = d.dispatch(msg) |
- self.assertEquals(False, result) |
- |
- result = d.dispatch(pres) |
- self.assertEquals(True, result) |
- |
- |
- def testOrderedXPathDispatch(self): |
- d = EventDispatcher() |
- cb = OrderedCallbackTracker() |
- d.addObserver("/message/body", cb.call2) |
- d.addObserver("/message", cb.call3, -1) |
- d.addObserver("/message/body", cb.call1, 1) |
- |
- msg = Element(("ns", "message")) |
- msg.addElement("body") |
- d.dispatch(msg) |
- self.assertEquals(cb.callList, [cb.call1, cb.call2, cb.call3], |
- "Calls out of order: %s" % |
- repr([c.__name__ for c in cb.callList])) |
- |
- |
- # Observers are put into CallbackLists that are then put into dictionaries |
- # keyed by the event trigger. Upon removal of the last observer for a |
- # particular event trigger, the (now empty) CallbackList and corresponding |
- # event trigger should be removed from those dictionaries to prevent |
- # slowdown and memory leakage. |
- |
- def test_cleanUpRemoveEventObserver(self): |
- """ |
- Test observer clean-up after removeObserver for named events. |
- """ |
- |
- d = EventDispatcher() |
- cb = CallbackTracker() |
- |
- d.addObserver('//event/test', cb.call) |
- d.dispatch(None, '//event/test') |
- self.assertEqual(1, cb.called) |
- d.removeObserver('//event/test', cb.call) |
- self.assertEqual(0, len(d._eventObservers.pop(0))) |
- |
- |
- def test_cleanUpRemoveXPathObserver(self): |
- """ |
- Test observer clean-up after removeObserver for XPath events. |
- """ |
- |
- d = EventDispatcher() |
- cb = CallbackTracker() |
- msg = Element((None, "message")) |
- |
- d.addObserver('/message', cb.call) |
- d.dispatch(msg) |
- self.assertEqual(1, cb.called) |
- d.removeObserver('/message', cb.call) |
- self.assertEqual(0, len(d._xpathObservers.pop(0))) |
- |
- |
- def test_cleanUpOnetimeEventObserver(self): |
- """ |
- Test observer clean-up after onetime named events. |
- """ |
- |
- d = EventDispatcher() |
- cb = CallbackTracker() |
- |
- d.addOnetimeObserver('//event/test', cb.call) |
- d.dispatch(None, '//event/test') |
- self.assertEqual(1, cb.called) |
- self.assertEqual(0, len(d._eventObservers.pop(0))) |
- |
- |
- def test_cleanUpOnetimeXPathObserver(self): |
- """ |
- Test observer clean-up after onetime XPath events. |
- """ |
- |
- d = EventDispatcher() |
- cb = CallbackTracker() |
- msg = Element((None, "message")) |
- |
- d.addOnetimeObserver('/message', cb.call) |
- d.dispatch(msg) |
- self.assertEqual(1, cb.called) |
- self.assertEqual(0, len(d._xpathObservers.pop(0))) |
- |
- |
- def test_observerRaisingException(self): |
- """ |
- Test that exceptions in observers do not bubble up to dispatch. |
- |
- The exceptions raised in observers should be logged and other |
- observers should be called as if nothing happened. |
- """ |
- |
- class OrderedCallbackList(utility.CallbackList): |
- def __init__(self): |
- self.callbacks = OrderedDict() |
- |
- class TestError(Exception): |
- pass |
- |
- def raiseError(_): |
- raise TestError() |
- |
- d = EventDispatcher() |
- cb = CallbackTracker() |
- |
- originalCallbackList = utility.CallbackList |
- |
- try: |
- utility.CallbackList = OrderedCallbackList |
- |
- d.addObserver('//event/test', raiseError) |
- d.addObserver('//event/test', cb.call) |
- try: |
- d.dispatch(None, '//event/test') |
- except TestError: |
- self.fail("TestError raised. Should have been logged instead.") |
- |
- self.assertEqual(1, len(self.flushLoggedErrors(TestError))) |
- self.assertEqual(1, cb.called) |
- finally: |
- utility.CallbackList = originalCallbackList |