Index: third_party/twisted_8_1/twisted/test/test_threadable.py |
diff --git a/third_party/twisted_8_1/twisted/test/test_threadable.py b/third_party/twisted_8_1/twisted/test/test_threadable.py |
deleted file mode 100644 |
index 394541e357384a37e3e3c45787e8207b7f9e3785..0000000000000000000000000000000000000000 |
--- a/third_party/twisted_8_1/twisted/test/test_threadable.py |
+++ /dev/null |
@@ -1,104 +0,0 @@ |
-# Copyright (c) 2001-2004 Twisted Matrix Laboratories. |
-# See LICENSE for details. |
- |
-import sys, pickle |
- |
-try: |
- import threading |
-except ImportError: |
- threading = None |
- |
-from twisted.trial import unittest |
-from twisted.python import threadable |
-from twisted.internet import defer, reactor |
- |
-class TestObject: |
- synchronized = ['aMethod'] |
- |
- x = -1 |
- y = 1 |
- |
- def aMethod(self): |
- for i in xrange(10): |
- self.x, self.y = self.y, self.x |
- self.z = self.x + self.y |
- assert self.z == 0, "z == %d, not 0 as expected" % (self.z,) |
- |
-threadable.synchronize(TestObject) |
- |
-class SynchronizationTestCase(unittest.TestCase): |
- if hasattr(sys, 'getcheckinterval'): |
- def setUpClass(self): |
- self.checkInterval = sys.getcheckinterval() |
- sys.setcheckinterval(7) |
- |
- def tearDownClass(self): |
- sys.setcheckinterval(self.checkInterval) |
- |
- |
- def setUp(self): |
- # XXX This is a trial hack. We need to make sure the reactor |
- # actually *starts* for isInIOThread() to have a meaningful result. |
- # Returning a Deferred here should force that to happen, if it has |
- # not happened already. In the future, this should not be |
- # necessary. |
- d = defer.Deferred() |
- reactor.callLater(0, d.callback, None) |
- return d |
- |
- |
- def testIsInIOThread(self): |
- foreignResult = [] |
- t = threading.Thread(target=lambda: foreignResult.append(threadable.isInIOThread())) |
- t.start() |
- t.join() |
- self.failIf(foreignResult[0], "Non-IO thread reported as IO thread") |
- self.failUnless(threadable.isInIOThread(), "IO thread reported as not IO thread") |
- |
- |
- def testThreadedSynchronization(self): |
- o = TestObject() |
- |
- errors = [] |
- |
- def callMethodLots(): |
- try: |
- for i in xrange(1000): |
- o.aMethod() |
- except AssertionError, e: |
- errors.append(str(e)) |
- |
- threads = [] |
- for x in range(5): |
- t = threading.Thread(target=callMethodLots) |
- threads.append(t) |
- t.start() |
- |
- for t in threads: |
- t.join() |
- |
- if errors: |
- raise unittest.FailTest(errors) |
- |
- def testUnthreadedSynchronization(self): |
- o = TestObject() |
- for i in xrange(1000): |
- o.aMethod() |
- |
-class SerializationTestCase(unittest.TestCase): |
- def testPickling(self): |
- lock = threadable.XLock() |
- lockType = type(lock) |
- lockPickle = pickle.dumps(lock) |
- newLock = pickle.loads(lockPickle) |
- self.failUnless(isinstance(newLock, lockType)) |
- |
- def testUnpickling(self): |
- lockPickle = 'ctwisted.python.threadable\nunpickle_lock\np0\n(tp1\nRp2\n.' |
- lock = pickle.loads(lockPickle) |
- newPickle = pickle.dumps(lock, 2) |
- newLock = pickle.loads(newPickle) |
- |
-if threading is None: |
- SynchronizationTestCase.testThreadedSynchronization.skip = "Platform lacks thread support" |
- SerializationTestCase.testPickling.skip = "Platform lacks thread support" |