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