← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/HTTPTestUtil.py

  • Committer: Robert Collins
  • Date: 2005-10-18 05:26:22 UTC
  • mto: This revision was merged to the branch mainline in revision 1463.
  • Revision ID: robertc@robertcollins.net-20051018052622-653d638c9e26fde4
fix broken tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/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 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, socket, errno, time
 
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" %
 
31
                                  (self.address_string(),
 
32
                                   self.log_date_time_string(),
 
33
                                   format%args))
 
34
 
 
35
    def handle_one_request(self):
 
36
        """Handle a single HTTP request.
 
37
 
 
38
        You normally don't need to override this method; see the class
 
39
        __doc__ string for information on how to handle specific HTTP
 
40
        commands such as GET and POST.
 
41
 
 
42
        """
 
43
        for i in xrange(1,11): # Don't try more than 10 times
 
44
            try:
 
45
                self.raw_requestline = self.rfile.readline()
 
46
            except socket.error, e:
 
47
                if e.args[0] == errno.EAGAIN:
 
48
                    self.log_message('EAGAIN (%d) while reading from raw_requestline' % i)
 
49
                    time.sleep(0.01)
 
50
                    continue
 
51
                raise
 
52
            else:
 
53
                break
 
54
        if not self.raw_requestline:
 
55
            self.close_connection = 1
 
56
            return
 
57
        if not self.parse_request(): # An error code has been sent, just exit
 
58
            return
 
59
        mname = 'do_' + self.command
 
60
        if not hasattr(self, mname):
 
61
            self.send_error(501, "Unsupported method (%r)" % self.command)
 
62
            return
 
63
        method = getattr(self, mname)
 
64
        method()
 
65
 
 
66
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
 
67
    def __init__(self, server_address, RequestHandlerClass, test_case):
 
68
        BaseHTTPServer.HTTPServer.__init__(self, server_address,
 
69
                                                RequestHandlerClass)
 
70
        self.test_case = test_case
 
71
 
 
72
 
 
73
class TestCaseWithWebserver(TestCaseInTempDir):
 
74
    """Derived class that starts a localhost-only webserver
 
75
    (in addition to what TestCaseInTempDir does).
 
76
 
 
77
    This is useful for testing RemoteBranch.
29
78
    """
30
79
 
 
80
    _HTTP_PORTS = range(13000, 0x8000)
 
81
 
 
82
    def _http_start(self):
 
83
        import SimpleHTTPServer, BaseHTTPServer, socket, errno
 
84
        httpd = None
 
85
        for port in self._HTTP_PORTS:
 
86
            try:
 
87
                httpd = TestingHTTPServer(('localhost', port),
 
88
                                          TestingHTTPRequestHandler,
 
89
                                          self)
 
90
            except socket.error, e:
 
91
                if e.args[0] == errno.EADDRINUSE:
 
92
                    continue
 
93
                print >>sys.stderr, "Cannot run webserver :-("
 
94
                raise
 
95
            else:
 
96
                break
 
97
 
 
98
        if httpd is None:
 
99
            raise WebserverNotAvailable("Cannot run webserver :-( "
 
100
                                        "no free ports in range %s..%s" %
 
101
                                        (_HTTP_PORTS[0], _HTTP_PORTS[-1]))
 
102
 
 
103
        self._http_base_url = 'http://localhost:%s/' % port
 
104
        self._http_starting.release()
 
105
        httpd.socket.settimeout(0.1)
 
106
 
 
107
        while self._http_running:
 
108
            try:
 
109
                httpd.handle_request()
 
110
            except socket.timeout:
 
111
                pass
 
112
 
 
113
    def get_remote_url(self, path):
 
114
        import os
 
115
 
 
116
        path_parts = path.split(os.path.sep)
 
117
        if os.path.isabs(path):
 
118
            if path_parts[:len(self._local_path_parts)] != \
 
119
                   self._local_path_parts:
 
120
                raise BadWebserverPath(path, self.test_dir)
 
121
            remote_path = '/'.join(path_parts[len(self._local_path_parts):])
 
122
        else:
 
123
            remote_path = '/'.join(path_parts)
 
124
 
 
125
        self._http_starting.acquire()
 
126
        self._http_starting.release()
 
127
        return self._http_base_url + remote_path
 
128
 
31
129
    def setUp(self):
32
 
        super(TestCaseWithWebserver, self).setUp()
33
 
        self.transport_readonly_server = bzrlib.transport.http.HttpServer
 
130
        TestCaseInTempDir.setUp(self)
 
131
        import threading, os
 
132
        self._local_path_parts = self.test_dir.split(os.path.sep)
 
133
        self._http_starting = threading.Lock()
 
134
        self._http_starting.acquire()
 
135
        self._http_running = True
 
136
        self._http_base_url = None
 
137
        self._http_thread = threading.Thread(target=self._http_start)
 
138
        self._http_thread.setDaemon(True)
 
139
        self._http_thread.start()
 
140
        self._http_proxy = os.environ.get("http_proxy")
 
141
        if self._http_proxy is not None:
 
142
            del os.environ["http_proxy"]
 
143
 
 
144
    def tearDown(self):
 
145
        self._http_running = False
 
146
        self._http_thread.join()
 
147
        if self._http_proxy is not None:
 
148
            import os
 
149
            os.environ["http_proxy"] = self._http_proxy
 
150
        TestCaseInTempDir.tearDown(self)
 
151