<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import <EOL> import sys , pickle <EOL> try : <EOL> import threading <EOL> except ImportError : <EOL> threadingSkip = "<STR_LIT>" <EOL> else : <EOL> threadingSkip = None <EOL> from twisted . python . compat import _PY3 <EOL> from twisted . trial import unittest <EOL> from twisted . python import threadable <EOL> class TestObject : <EOL> synchronized = [ '<STR_LIT>' ] <EOL> x = - <NUM_LIT:1> <EOL> y = <NUM_LIT:1> <EOL> def aMethod ( self ) : <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> self . x , self . y = self . y , self . x <EOL> self . z = self . x + self . y <EOL> assert self . z == <NUM_LIT:0> , "<STR_LIT>" % ( self . z , ) <EOL> threadable . synchronize ( TestObject ) <EOL> class SynchronizationTests ( unittest . SynchronousTestCase ) : <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> if _PY3 : <EOL> if getattr ( sys , '<STR_LIT>' , None ) is not None : <EOL> self . addCleanup ( sys . setswitchinterval , sys . getswitchinterval ( ) ) <EOL> sys . setswitchinterval ( <NUM_LIT> ) <EOL> else : <EOL> if getattr ( sys , '<STR_LIT>' , None ) is not None : <EOL> self . addCleanup ( sys . setcheckinterval , sys . getcheckinterval ( ) ) <EOL> sys . setcheckinterval ( <NUM_LIT:7> ) <EOL> def test_synchronizedName ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( "<STR_LIT>" , TestObject . aMethod . __name__ ) <EOL> def test_isInIOThread ( self ) : <EOL> """<STR_LIT>""" <EOL> threadable . registerAsIOThread ( ) <EOL> foreignResult = [ ] <EOL> t = threading . Thread ( <EOL> target = lambda : foreignResult . append ( threadable . isInIOThread ( ) ) ) <EOL> t . start ( ) <EOL> t . join ( ) <EOL> self . assertFalse ( <EOL> foreignResult [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . assertTrue ( <EOL> threadable . isInIOThread ( ) , "<STR_LIT>" ) <EOL> def testThreadedSynchronization ( self ) : <EOL> o = TestObject ( ) <EOL> errors = [ ] <EOL> def callMethodLots ( ) : <EOL> try : <EOL> for i in range ( <NUM_LIT:1000> ) : <EOL> o . aMethod ( ) <EOL> except AssertionError as e : <EOL> errors . append ( str ( e ) ) <EOL> threads = [ ] <EOL> for x in range ( <NUM_LIT:5> ) : <EOL> t = threading . Thread ( target = callMethodLots ) <EOL> threads . append ( t ) <EOL> t . start ( ) <EOL> for t in threads : <EOL> t . join ( ) <EOL> if errors : <EOL> raise unittest . FailTest ( errors ) <EOL> if threadingSkip is not None : <EOL> testThreadedSynchronization . skip = threadingSkip <EOL> test_isInIOThread . skip = threadingSkip <EOL> def testUnthreadedSynchronization ( self ) : <EOL> o = TestObject ( ) <EOL> for i in range ( <NUM_LIT:1000> ) : <EOL> o . aMethod ( ) <EOL> class SerializationTests ( unittest . SynchronousTestCase ) : <EOL> def testPickling ( self ) : <EOL> lock = threadable . XLock ( ) <EOL> lockType = type ( lock ) <EOL> lockPickle = pickle . dumps ( lock ) <EOL> newLock = pickle . loads ( lockPickle ) <EOL> self . assertTrue ( isinstance ( newLock , lockType ) ) <EOL> if threadingSkip is not None : <EOL> testPickling . skip = threadingSkip <EOL> def testUnpickling ( self ) : <EOL> lockPickle = b'<STR_LIT>' <EOL> lock = pickle . loads ( lockPickle ) <EOL> newPickle = pickle . dumps ( lock , <NUM_LIT:2> ) <EOL> pickle . loads ( newPickle ) </s>
