Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(230)

Unified Diff: third_party/twisted_8_1/twisted/test/test_threadable.py

Issue 12261012: Remove third_party/twisted_8_1 (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/tools/build
Patch Set: Created 7 years, 10 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
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"
« no previous file with comments | « third_party/twisted_8_1/twisted/test/test_text.py ('k') | third_party/twisted_8_1/twisted/test/test_threadpool.py » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698