← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_test_server.py

  • Committer: Andrew Bennetts
  • Date: 2010-10-13 00:26:41 UTC
  • mto: This revision was merged to the branch mainline in revision 5498.
  • Revision ID: andrew.bennetts@canonical.com-20101013002641-9tlh9k89mlj1666m
Keep docs-plain working.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_test_server.py
1
 
# Copyright (C) 2010, 2011 Canonical Ltd
 
1
# Copyright (C) 2010 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
24
24
    tests,
25
25
    )
26
26
from bzrlib.tests import test_server
27
 
from bzrlib.tests.scenarios import load_tests_apply_scenarios
28
 
 
29
 
 
30
 
load_tests = load_tests_apply_scenarios
 
27
 
 
28
 
 
29
def load_tests(basic_tests, module, loader):
 
30
    suite = loader.suiteClass()
 
31
    server_tests, remaining_tests = tests.split_suite_by_condition(
 
32
        basic_tests, tests.condition_isinstance(TestTCPServerInAThread))
 
33
    server_scenarios = [ (name, {'server_class': getattr(test_server, name)})
 
34
                         for name in
 
35
                         ('TestingTCPServer', 'TestingThreadingTCPServer')]
 
36
    tests.multiply_tests(server_tests, server_scenarios, suite)
 
37
    suite.addTest(remaining_tests)
 
38
    return suite
 
39
 
 
40
 
 
41
class TestThreadWithException(tests.TestCase):
 
42
 
 
43
    def test_start_and_join_smoke_test(self):
 
44
        def do_nothing():
 
45
            pass
 
46
 
 
47
        tt = test_server.ThreadWithException(target=do_nothing)
 
48
        tt.start()
 
49
        tt.join()
 
50
 
 
51
    def test_exception_is_re_raised(self):
 
52
        class MyException(Exception):
 
53
            pass
 
54
 
 
55
        def raise_my_exception():
 
56
            raise MyException()
 
57
 
 
58
        tt = test_server.ThreadWithException(target=raise_my_exception)
 
59
        tt.start()
 
60
        self.assertRaises(MyException, tt.join)
 
61
 
 
62
    def test_join_when_no_exception(self):
 
63
        resume = threading.Event()
 
64
        class MyException(Exception):
 
65
            pass
 
66
 
 
67
        def raise_my_exception():
 
68
            # Wait for the test to tell us to resume
 
69
            resume.wait()
 
70
            # Now we can raise
 
71
            raise MyException()
 
72
 
 
73
        tt = test_server.ThreadWithException(target=raise_my_exception)
 
74
        tt.start()
 
75
        tt.join(timeout=0)
 
76
        self.assertIs(None, tt.exception)
 
77
        resume.set()
 
78
        self.assertRaises(MyException, tt.join)
31
79
 
32
80
 
33
81
class TCPClient(object):
81
129
 
82
130
class TestTCPServerInAThread(tests.TestCase):
83
131
 
84
 
    scenarios = [ 
85
 
        (name, {'server_class': getattr(test_server, name)})
86
 
        for name in
87
 
        ('TestingTCPServer', 'TestingThreadingTCPServer')]
88
 
 
89
132
    # Set by load_tests()
90
133
    server_class = None
91
134
 
181
224
 
182
225
            def handle_connection(self):
183
226
                req = self.rfile.readline()
184
 
                threading.currentThread().set_sync_event(sync)
 
227
                threading.currentThread().set_ready_event(sync)
185
228
                raise FailToRespond()
186
229
 
187
230
        server = self.get_server(
203
246
            def handle(self):
204
247
                # We want to sync with the thread that is serving the
205
248
                # connection.
206
 
                threading.currentThread().set_sync_event(sync)
 
249
                threading.currentThread().set_ready_event(sync)
207
250
                raise CantServe()
208
251
 
209
252
        server = self.get_server(