~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/HTTPTestUtil.py

  • Committer: Aaron Bentley
  • Date: 2006-02-22 14:39:42 UTC
  • mto: (2027.1.2 revert-subpath-56549)
  • mto: This revision was merged to the branch mainline in revision 1570.
  • Revision ID: abentley@panoramicfeedback.com-20060222143942-ae72299f2de66767
Fixed build_tree with symlinks

Show diffs side-by-side

added added

removed removed

Lines of Context:
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, socket, errno, time
18
 
from bzrlib.tests import TestCaseInTempDir
19
 
from bzrlib.osutils import relpath
20
 
 
21
 
 
22
 
class WebserverNotAvailable(Exception):
23
 
    pass
24
 
 
25
 
class TestingHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
26
 
    def log_message(self, format, *args):
27
 
        self.server.test_case.log("webserver - %s - - [%s] %s",
28
 
                                  self.address_string(),
29
 
                                  self.log_date_time_string(),
30
 
                                  format%args)
31
 
 
32
 
    def handle_one_request(self):
33
 
        """Handle a single HTTP request.
34
 
 
35
 
        You normally don't need to override this method; see the class
36
 
        __doc__ string for information on how to handle specific HTTP
37
 
        commands such as GET and POST.
38
 
 
39
 
        """
40
 
        for i in xrange(1,11): # Don't try more than 10 times
41
 
            try:
42
 
                self.raw_requestline = self.rfile.readline()
43
 
            except socket.error, e:
44
 
                if e.args[0] in (errno.EAGAIN, errno.EWOULDBLOCK):
45
 
                    # omitted for now because some tests look at the log of
46
 
                    # the server and expect to see no errors.  see recent
47
 
                    # email thread. -- mbp 20051021. 
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.
 
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.
78
29
    """
79
30
 
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
 
        if os.path.isabs(path):
117
 
            remote_path = relpath(self.test_dir, path)
118
 
        else:
119
 
            remote_path = path
120
 
 
121
 
        self._http_starting.acquire()
122
 
        self._http_starting.release()
123
 
        return self._http_base_url + remote_path
124
 
 
125
31
    def setUp(self):
126
 
        TestCaseInTempDir.setUp(self)
127
 
        import threading, os
128
 
        self._http_starting = threading.Lock()
129
 
        self._http_starting.acquire()
130
 
        self._http_running = True
131
 
        self._http_base_url = None
132
 
        self._http_thread = threading.Thread(target=self._http_start)
133
 
        self._http_thread.setDaemon(True)
134
 
        self._http_thread.start()
135
 
        self._http_proxy = os.environ.get("http_proxy")
136
 
        if self._http_proxy is not None:
137
 
            del os.environ["http_proxy"]
138
 
 
139
 
    def tearDown(self):
140
 
        self._http_running = False
141
 
        self._http_thread.join()
142
 
        if self._http_proxy is not None:
143
 
            import os
144
 
            os.environ["http_proxy"] = self._http_proxy
145
 
        TestCaseInTempDir.tearDown(self)
146
 
 
 
32
        super(TestCaseWithWebserver, self).setUp()
 
33
        self.transport_readonly_server = bzrlib.transport.http.HttpServer