← Back to branch summary

~ubuntu-branches/ubuntu/lucid/python2.6/lucid

« back to all changes in this revision

Viewing changes to Lib/test/test_thread.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2010-03-11 13:30:19 UTC
  • mto: (10.1.13 sid)
  • mto: This revision was merged to the branch mainline in revision 44.
  • Revision ID: james.westby@ubuntu.com-20100311133019-sblbooa3uqrkoe70
Tags: upstream-2.6.5~rc2
Import upstream version 2.6.5~rc2

Show diffs side-by-side

added added

removed removed

Lines of Context:
Lib/test/test_thread.py
4
4
from test import test_support
5
5
import thread
6
6
import time
 
7
import sys
7
8
 
 
9
from test import lock_tests
8
10
 
9
11
NUMTASKS = 10
10
12
NUMTRIPS = 3
26
28
        self.done_mutex.acquire()
27
29
        self.running_mutex = thread.allocate_lock()
28
30
        self.random_mutex = thread.allocate_lock()
 
31
        self.created = 0
29
32
        self.running = 0
30
33
        self.next_ident = 0
31
34
 
37
40
            self.next_ident += 1
38
41
            verbose_print("creating task %s" % self.next_ident)
39
42
            thread.start_new_thread(self.task, (self.next_ident,))
 
43
            self.created += 1
40
44
            self.running += 1
41
45
 
42
46
    def task(self, ident):
47
51
        verbose_print("task %s done" % ident)
48
52
        with self.running_mutex:
49
53
            self.running -= 1
50
 
            if self.running == 0:
 
54
            if self.created == NUMTASKS and self.running == 0:
51
55
                self.done_mutex.release()
52
56
 
53
57
    def test_starting_threads(self):
89
93
            for tss in (262144, 0x100000):
90
94
                verbose_print("trying stack_size = (%d)" % tss)
91
95
                self.next_ident = 0
 
96
                self.created = 0
92
97
                for i in range(NUMTASKS):
93
98
                    self.newtask()
94
99
 
161
166
            self.done_mutex.release()
162
167
 
163
168
 
 
169
class LockTests(lock_tests.LockTests):
 
170
    locktype = thread.allocate_lock
 
171
 
 
172
 
 
173
class TestForkInThread(unittest.TestCase):
 
174
    def setUp(self):
 
175
        self.read_fd, self.write_fd = os.pipe()
 
176
 
 
177
    def _test_forkinthread(self):
 
178
        def thread1():
 
179
            try:
 
180
                pid = os.fork() # fork in a thread
 
181
            except RuntimeError:
 
182
                sys.exit(0) # exit the child
 
183
 
 
184
            if pid == 0: # child
 
185
                os.close(self.read_fd)
 
186
                os.write(self.write_fd, "OK")
 
187
                sys.exit(0)
 
188
            else: # parent
 
189
                os.close(self.write_fd)
 
190
 
 
191
        thread.start_new_thread(thread1, ())
 
192
        self.assertEqual(os.read(self.read_fd, 2), "OK",
 
193
                         "Unable to fork() in thread")
 
194
 
 
195
    if not sys.platform.startswith('win'):
 
196
        test_forkinthread = _test_forkinthread
 
197
 
 
198
    def tearDown(self):
 
199
        try:
 
200
            os.close(self.read_fd)
 
201
        except OSError:
 
202
            pass
 
203
 
 
204
        try:
 
205
            os.close(self.write_fd)
 
206
        except OSError:
 
207
            pass
 
208
 
 
209
 
164
210
def test_main():
165
 
    test_support.run_unittest(ThreadRunningTests, BarrierTest)
 
211
    test_support.run_unittest(ThreadRunningTests, BarrierTest, LockTests,
 
212
                              TestForkInThread)
166
213
 
167
214
if __name__ == "__main__":
168
215
    test_main()