← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/HTTPTestUtil.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2006-06-16 14:33:42 UTC
  • mfrom: (1770.2.1 config)
  • Revision ID: pqm@pqm.ubuntu.com-20060616143342-8f7f4a4f77c1e4c8
Use create_signature for signing policy, deprecate check_signatures for this

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/HTTPTestUtil.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
 
import BaseHTTPServer, SimpleHTTPServer
18
 
from bzrlib.selftest import TestCaseInTempDir
19
 
 
20
 
 
21
 
class WebserverNotAvailable(Exception):
22
 
    pass
23
 
 
24
 
class BadWebserverPath(ValueError):
25
 
    def __str__(self):
26
 
        return 'path %s is not in %s' % self.args
27
 
 
28
 
class TestingHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
29
 
    def log_message(self, format, *args):
30
 
        self.server.test_case.log("webserver - %s - - [%s] %s\n" %
31
 
                                  (self.address_string(),
32
 
                                   self.log_date_time_string(),
33
 
                                   format%args))
34
 
 
35
 
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
36
 
    def __init__(self, server_address, RequestHandlerClass, test_case):
37
 
        BaseHTTPServer.HTTPServer.__init__(self, server_address,
38
 
                                                RequestHandlerClass)
39
 
        self.test_case = test_case
40
 
 
41
 
class TestCaseWithWebserver(TestCaseInTempDir):
42
 
    """Derived class that starts a localhost-only webserver
43
 
    (in addition to what TestCaseInTempDir does).
44
 
 
45
 
    This is useful for testing RemoteBranch.
 
17
import os
 
18
 
 
19
import bzrlib
 
20
from bzrlib.tests import TestCaseWithTransport
 
21
 
 
22
 
 
23
class TestCaseWithWebserver(TestCaseWithTransport):
 
24
    """A support class that provides readonly urls that are http://.
 
25
 
 
26
    This is done by forcing the readonly server to be an http one. This 
 
27
    will current fail if the primary transport is not backed by regular disk
 
28
    files.
46
29
    """
47
30
 
48
 
    _HTTP_PORTS = range(13000, 0x8000)
49
 
 
50
 
    def _http_start(self):
51
 
        import SimpleHTTPServer, BaseHTTPServer, socket, errno
52
 
        httpd = None
53
 
        for port in self._HTTP_PORTS:
54
 
            try:
55
 
                httpd = TestingHTTPServer(('localhost', port),
56
 
                                          TestingHTTPRequestHandler,
57
 
                                          self)
58
 
            except socket.error, e:
59
 
                if e.args[0] == errno.EADDRINUSE:
60
 
                    continue
61
 
                print >>sys.stderr, "Cannot run webserver :-("
62
 
                raise
63
 
            else:
64
 
                break
65
 
 
66
 
        if httpd is None:
67
 
            raise WebserverNotAvailable("Cannot run webserver :-( "
68
 
                                        "no free ports in range %s..%s" %
69
 
                                        (_HTTP_PORTS[0], _HTTP_PORTS[-1]))
70
 
 
71
 
        self._http_base_url = 'http://localhost:%s/' % port
72
 
        self._http_starting.release()
73
 
        httpd.socket.settimeout(1)
74
 
 
75
 
        while self._http_running:
76
 
            try:
77
 
                httpd.handle_request()
78
 
            except socket.timeout:
79
 
                pass
80
 
 
81
 
    def get_remote_url(self, path):
82
 
        import os
83
 
 
84
 
        path_parts = path.split(os.path.sep)
85
 
        if os.path.isabs(path):
86
 
            if path_parts[:len(self._local_path_parts)] != \
87
 
                   self._local_path_parts:
88
 
                raise BadWebserverPath(path, self.test_dir)
89
 
            remote_path = '/'.join(path_parts[len(self._local_path_parts):])
90
 
        else:
91
 
            remote_path = '/'.join(path_parts)
92
 
 
93
 
        self._http_starting.acquire()
94
 
        self._http_starting.release()
95
 
        return self._http_base_url + remote_path
96
 
 
97
31
    def setUp(self):
98
32
        super(TestCaseWithWebserver, self).setUp()
99
 
        import threading, os
100
 
        self._local_path_parts = self.test_dir.split(os.path.sep)
101
 
        self._http_starting = threading.Lock()
102
 
        self._http_starting.acquire()
103
 
        self._http_running = True
104
 
        self._http_base_url = None
105
 
        self._http_thread = threading.Thread(target=self._http_start)
106
 
        self._http_thread.setDaemon(True)
107
 
        self._http_thread.start()
108
 
 
109
 
    def tearDown(self):
110
 
        self._http_running = False
111
 
        self._http_thread.join()
112
 
        super(TestCaseWithWebserver, self).tearDown()
 
33
        self.transport_readonly_server = bzrlib.transport.http.HttpServer