/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 bzrlib/selftest/HTTPTestUtil.py

  • Committer: John Arbash Meinel
  • Date: 2005-09-15 21:35:53 UTC
  • mfrom: (907.1.57)
  • mto: (1393.2.1)
  • mto: This revision was merged to the branch mainline in revision 1396.
  • Revision ID: john@arbash-meinel.com-20050915213552-a6c83a5ef1e20897
(broken) Transport work is merged in. Tests do not pass yet.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/HTTPTestUtil.py
1
1
# Copyright (C) 2005 by 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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
#
 
7
 
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
12
 
#
 
12
 
13
13
# You should have received a copy of the GNU General Public License
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 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.
 
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.
29
46
    """
30
47
 
 
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
 
31
97
    def setUp(self):
32
98
        super(TestCaseWithWebserver, self).setUp()
33
 
        self.transport_readonly_server = bzrlib.transport.http.HttpServer
 
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()