← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_cethread.py

  • Committer: Vincent Ladeuil
  • Date: 2011-02-10 12:37:27 UTC
  • mto: This revision was merged to the branch mainline in revision 5661.
  • Revision ID: v.ladeuil+lp@free.fr-20110210123727-8e0pu4wtlt6fj7nf
thread is already a python module, avoid confusion and use cethread instead.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_cethread.py
17
17
import threading
18
18
 
19
19
from bzrlib import (
 
20
    cethread,
20
21
    tests,
21
 
    thread,
22
22
    )
23
23
 
24
24
 
28
28
        def do_nothing():
29
29
            pass
30
30
 
31
 
        tt = thread.CatchingExceptionThread(target=do_nothing)
 
31
        tt = cethread.CatchingExceptionThread(target=do_nothing)
32
32
        tt.start()
33
33
        tt.join()
34
34
 
39
39
        def raise_my_exception():
40
40
            raise MyException()
41
41
 
42
 
        tt = thread.CatchingExceptionThread(target=raise_my_exception)
 
42
        tt = cethread.CatchingExceptionThread(target=raise_my_exception)
43
43
        tt.start()
44
44
        self.assertRaises(MyException, tt.join)
45
45
 
54
54
            # Now we can raise
55
55
            raise MyException()
56
56
 
57
 
        tt = thread.CatchingExceptionThread(target=raise_my_exception)
 
57
        tt = cethread.CatchingExceptionThread(target=raise_my_exception)
58
58
        tt.start()
59
59
        tt.join(timeout=0)
60
60
        self.assertIs(None, tt.exception)
73
73
            # Now we can raise
74
74
            raise MyException()
75
75
 
76
 
        tt = thread.CatchingExceptionThread(target=raise_my_exception,
 
76
        tt = cethread.CatchingExceptionThread(target=raise_my_exception,
77
77
                                            sync_event=in_thread)
78
78
        tt.start()
79
79
        tt.join(timeout=0)
89
89
        control2 = threading.Event()
90
90
        control3 = threading.Event()
91
91
 
92
 
        class TestThread(thread.CatchingExceptionThread):
 
92
        class TestThread(cethread.CatchingExceptionThread):
93
93
 
94
94
            def __init__(self, *args, **kwargs):
95
95
                super(TestThread, self).__init__(*args,
131
131
        class MyException(Exception):
132
132
            pass
133
133
 
134
 
        class TestThread(thread.CatchingExceptionThread):
 
134
        class TestThread(cethread.CatchingExceptionThread):
135
135
 
136
136
            def __init__(self, *args, **kwargs):
137
137
                self.step1 = threading.Event()