← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/HTTPTestUtil.py

  • Committer: Jelmer Vernooij
  • Date: 2005-11-21 20:10:56 UTC
  • mto: (1185.33.40 bzr.dev)
  • mto: This revision was merged to the branch mainline in revision 1512.
  • Revision ID: jelmer@samba.org-20051121201056-b5b364b20cb58115
Don't clear Revision.parent_ids after it has been set from the arguments

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
                    # omitted for now because some tests look at the log of
 
49
                    # the server and expect to see no errors.  see recent
 
50
                    # email thread. -- mbp 20051021. 
 
51
                    ## self.log_message('EAGAIN (%d) while reading from raw_requestline' % i)
 
52
                    time.sleep(0.01)
 
53
                    continue
 
54
                raise
 
55
            else:
 
56
                break
 
57
        if not self.raw_requestline:
 
58
            self.close_connection = 1
 
59
            return
 
60
        if not self.parse_request(): # An error code has been sent, just exit
 
61
            return
 
62
        mname = 'do_' + self.command
 
63
        if not hasattr(self, mname):
 
64
            self.send_error(501, "Unsupported method (%r)" % self.command)
 
65
            return
 
66
        method = getattr(self, mname)
 
67
        method()
 
68
 
 
69
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
 
70
    def __init__(self, server_address, RequestHandlerClass, test_case):
 
71
        BaseHTTPServer.HTTPServer.__init__(self, server_address,
 
72
                                                RequestHandlerClass)
 
73
        self.test_case = test_case
 
74
 
 
75
 
 
76
class TestCaseWithWebserver(TestCaseInTempDir):
 
77
    """Derived class that starts a localhost-only webserver
 
78
    (in addition to what TestCaseInTempDir does).
 
79
 
 
80
    This is useful for testing RemoteBranch.
29
81
    """
30
82
 
 
83
    _HTTP_PORTS = range(13000, 0x8000)
 
84
 
 
85
    def _http_start(self):
 
86
        import SimpleHTTPServer, BaseHTTPServer, socket, errno
 
87
        httpd = None
 
88
        for port in self._HTTP_PORTS:
 
89
            try:
 
90
                httpd = TestingHTTPServer(('localhost', port),
 
91
                                          TestingHTTPRequestHandler,
 
92
                                          self)
 
93
            except socket.error, e:
 
94
                if e.args[0] == errno.EADDRINUSE:
 
95
                    continue
 
96
                print >>sys.stderr, "Cannot run webserver :-("
 
97
                raise
 
98
            else:
 
99
                break
 
100
 
 
101
        if httpd is None:
 
102
            raise WebserverNotAvailable("Cannot run webserver :-( "
 
103
                                        "no free ports in range %s..%s" %
 
104
                                        (_HTTP_PORTS[0], _HTTP_PORTS[-1]))
 
105
 
 
106
        self._http_base_url = 'http://localhost:%s/' % port
 
107
        self._http_starting.release()
 
108
        httpd.socket.settimeout(0.1)
 
109
 
 
110
        while self._http_running:
 
111
            try:
 
112
                httpd.handle_request()
 
113
            except socket.timeout:
 
114
                pass
 
115
 
 
116
    def get_remote_url(self, path):
 
117
        import os
 
118
 
 
119
        path_parts = path.split(os.path.sep)
 
120
        if os.path.isabs(path):
 
121
            if path_parts[:len(self._local_path_parts)] != \
 
122
                   self._local_path_parts:
 
123
                raise BadWebserverPath(path, self.test_dir)
 
124
            remote_path = '/'.join(path_parts[len(self._local_path_parts):])
 
125
        else:
 
126
            remote_path = '/'.join(path_parts)
 
127
 
 
128
        self._http_starting.acquire()
 
129
        self._http_starting.release()
 
130
        return self._http_base_url + remote_path
 
131
 
31
132
    def setUp(self):
32
 
        super(TestCaseWithWebserver, self).setUp()
33
 
        self.transport_readonly_server = bzrlib.transport.http.HttpServer
 
133
        TestCaseInTempDir.setUp(self)
 
134
        import threading, os
 
135
        self._local_path_parts = self.test_dir.split(os.path.sep)
 
136
        self._http_starting = threading.Lock()
 
137
        self._http_starting.acquire()
 
138
        self._http_running = True
 
139
        self._http_base_url = None
 
140
        self._http_thread = threading.Thread(target=self._http_start)
 
141
        self._http_thread.setDaemon(True)
 
142
        self._http_thread.start()
 
143
        self._http_proxy = os.environ.get("http_proxy")
 
144
        if self._http_proxy is not None:
 
145
            del os.environ["http_proxy"]
 
146
 
 
147
    def tearDown(self):
 
148
        self._http_running = False
 
149
        self._http_thread.join()
 
150
        if self._http_proxy is not None:
 
151
            import os
 
152
            os.environ["http_proxy"] = self._http_proxy
 
153
        TestCaseInTempDir.tearDown(self)
 
154