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

Merge from mbp.

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
 
17
import BaseHTTPServer, SimpleHTTPServer, socket, errno, time
 
18
from bzrlib.tests import TestCaseInTempDir
19
19
 
20
20
 
21
21
class WebserverNotAvailable(Exception):
27
27
 
28
28
class TestingHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
29
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))
 
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] in (errno.EAGAIN, errno.EWOULDBLOCK):
 
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()
34
68
 
35
69
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
36
70
    def __init__(self, server_address, RequestHandlerClass, test_case):
38
72
                                                RequestHandlerClass)
39
73
        self.test_case = test_case
40
74
 
 
75
 
41
76
class TestCaseWithWebserver(TestCaseInTempDir):
42
77
    """Derived class that starts a localhost-only webserver
43
78
    (in addition to what TestCaseInTempDir does).
70
105
 
71
106
        self._http_base_url = 'http://localhost:%s/' % port
72
107
        self._http_starting.release()
73
 
        httpd.socket.settimeout(1)
 
108
        httpd.socket.settimeout(0.1)
74
109
 
75
110
        while self._http_running:
76
111
            try:
95
130
        return self._http_base_url + remote_path
96
131
 
97
132
    def setUp(self):
98
 
        super(TestCaseWithWebserver, self).setUp()
 
133
        TestCaseInTempDir.setUp(self)
99
134
        import threading, os
100
135
        self._local_path_parts = self.test_dir.split(os.path.sep)
101
136
        self._http_starting = threading.Lock()
105
140
        self._http_thread = threading.Thread(target=self._http_start)
106
141
        self._http_thread.setDaemon(True)
107
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"]
108
146
 
109
147
    def tearDown(self):
110
148
        self._http_running = False
111
149
        self._http_thread.join()
112
 
        super(TestCaseWithWebserver, self).tearDown()
 
150
        if self._http_proxy is not None:
 
151
            import os
 
152
            os.environ["http_proxy"] = self._http_proxy
 
153
        TestCaseInTempDir.tearDown(self)
 
154