Remove third_party/twisted_8_1

third_party/twisted_8_1/twisted/words/test/test_xishutil.py

-# 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