/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_cethread.py

  • Committer: Breezy landing bot
  • Author(s): Colin Watson
  • Date: 2020-11-16 21:47:08 UTC
  • mfrom: (7521.1.1 remove-lp-workaround)
  • Revision ID: breezy.the.bot@gmail.com-20201116214708-jos209mgxi41oy15
Remove breezy.git workaround for bazaar.launchpad.net.

Merged from https://code.launchpad.net/~cjwatson/brz/remove-lp-workaround/+merge/393710

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_cethread.py
1
 
# Copyright (C) 2010, 2011 Canonical Ltd
 
1
# Copyright (C) 2011, 2016 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
16
16
 
17
17
import threading
18
18
 
19
 
from bzrlib import (
 
19
from .. import (
20
20
    cethread,
21
21
    tests,
22
22
    )
45
45
 
46
46
    def test_join_around_exception(self):
47
47
        resume = threading.Event()
 
48
 
48
49
        class MyException(Exception):
49
50
            pass
50
51
 
64
65
    def test_sync_event(self):
65
66
        control = threading.Event()
66
67
        in_thread = threading.Event()
 
68
 
67
69
        class MyException(Exception):
68
70
            pass
69
71
 
74
76
            raise MyException()
75
77
 
76
78
        tt = cethread.CatchingExceptionThread(target=raise_my_exception,
77
 
                                            sync_event=in_thread)
 
79
                                              sync_event=in_thread)
78
80
        tt.start()
79
81
        tt.join(timeout=0)
80
82
        self.assertIs(None, tt.exception)
81
83
        self.assertIs(in_thread, tt.sync_event)
82
84
        control.set()
83
85
        self.assertRaises(MyException, tt.join)
84
 
        self.assertEquals(True, tt.sync_event.isSet())
 
86
        self.assertEqual(True, tt.sync_event.isSet())
85
87
 
86
88
    def test_switch_and_set(self):
87
89
        """Caller can precisely control a thread."""
111
113
 
112
114
        tt = TestThread()
113
115
        tt.start()
114
 
        self.assertEquals('starting', tt.current_step)
 
116
        self.assertEqual('starting', tt.current_step)
115
117
        control1.set()
116
118
        tt.step1.wait()
117
 
        self.assertEquals('step1', tt.current_step)
 
119
        self.assertEqual('step1', tt.current_step)
118
120
        control2.set()
119
121
        tt.step2.wait()
120
 
        self.assertEquals('step2', tt.current_step)
 
122
        self.assertEqual('step2', tt.current_step)
121
123
        control3.set()
122
124
        # We don't wait on tt.final
123
125
        tt.join()
124
 
        self.assertEquals('done', tt.current_step)
 
126
        self.assertEqual('done', tt.current_step)
125
127
 
126
128
    def test_exception_while_switch_and_set(self):
127
129
        control1 = threading.Event()
152
154
 
153
155
        tt = TestThread()
154
156
        tt.start()
155
 
        self.assertEquals('starting', tt.current_step)
 
157
        self.assertEqual('starting', tt.current_step)
156
158
        control1.set()
157
159
        # We now wait on step1 which will be set when catching the exception
158
160
        tt.step1.wait()