1from PyObjCTools.TestSupport import * 2import objc 3import sys 4import time 5 6from PyObjCTest.testhelper import PyObjC_TestClass4 7 8from Foundation import * 9 10class ThreadingTest (TestCase): 11 def setUp(self): 12 # Set a very small check interval, this will make it more likely 13 # that the interpreter crashes when threading is done incorrectly. 14 if sys.version_info[:2] >= (3, 2): 15 self._int = sys.getswitchinterval() 16 sys.setswitchinterval(0.0000001) 17 else: 18 self._int = sys.getcheckinterval() 19 sys.setcheckinterval(1) 20 21 def tearDown(self): 22 if sys.version_info[:2] >= (3, 2): 23 sys.setswitchinterval(self._int) 24 else: 25 sys.setcheckinterval(self._int) 26 27 def testNSObjectString(self): 28 29 class PyObjCTestThreadRunnerString (NSObject): 30 def init(self): 31 self = super(PyObjCTestThreadRunnerString, self).init() 32 if self is None: return None 33 34 self.storage = [] 35 return self 36 37 def run_(self, argument): 38 NSAutoreleasePool.alloc().init() 39 self.storage.append(argument) 40 41 myObj = PyObjCTestThreadRunnerString.alloc().init() 42 43 NSThread.detachNewThreadSelector_toTarget_withObject_( 44 'run:', myObj, b"hello world".decode('ascii')) 45 46 time.sleep(2) 47 self.assertEqual(myObj.storage[0], b"hello world".decode('ascii')) 48 49 def testNSObject(self): 50 51 class PyObjCTestThreadRunner (NSObject): 52 def run_(self, argument): 53 NSAutoreleasePool.alloc().init() 54 for i in range(100): 55 argument.append(i) 56 57 myObj = PyObjCTestThreadRunner.alloc().init() 58 lst = [] 59 60 NSThread.detachNewThreadSelector_toTarget_withObject_( 61 'run:', myObj, lst) 62 63 lst2 = [] 64 for i in range(100): 65 lst2.append(i*2) 66 67 time.sleep(2) 68 self.assertEqual(lst, list(range(100))) 69 70 def testPyObject(self): 71 import os 72 73 class TestThreadRunner : 74 def run_(self, argument): 75 for i in range(100): 76 argument.append(i) 77 78 myObj = TestThreadRunner() 79 lst = [] 80 81 # Redirect stderr to avoid spurious messages when running the 82 # tests. 83 dupped = os.dup(2) 84 fp = os.open('/dev/null', os.O_RDWR) 85 os.dup2(fp, 2) 86 os.close(fp) 87 88 try: 89 NSThread.detachNewThreadSelector_toTarget_withObject_( 90 'run:', myObj, lst) 91 92 lst2 = [] 93 for i in range(100): 94 lst2.append(i*2) 95 96 time.sleep(2) 97 self.assertEqual(lst, list(range(100))) 98 99 finally: 100 os.dup2(dupped, 2) 101 102 def testCalling(self): 103 class Dummy: 104 pass 105 class PyObjCTestCalling (NSObject) : 106 def call(self): 107 return Dummy() 108 109 my = PyObjC_TestClass4.alloc().init() 110 cb = PyObjCTestCalling.alloc().init() 111 112 NSThread.detachNewThreadSelector_toTarget_withObject_( 113 'runThread:', my, cb) 114 115 time.sleep(2) 116 117 retval = my.returnObject() 118 self.assert_(isinstance(retval, Dummy)) 119 120if __name__ == "__main__": 121 main() 122