← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/HTTPTestUtil.py

  • Committer: Aaron Bentley
  • Date: 2005-09-19 02:52:24 UTC
  • mto: (1185.1.29)
  • mto: This revision was merged to the branch mainline in revision 1390.
  • Revision ID: aaron.bentley@utoronto.ca-20050919025224-1cc3c70640086e09
TODO re 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 BaseHTTPServer, SimpleHTTPServer, socket, errno, time
18
 
from bzrlib.tests import TestCaseInTempDir
 
17
import BaseHTTPServer, SimpleHTTPServer
 
18
from bzrlib.selftest 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",
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()
 
30
        self.server.test_case.log("webserver - %s - - [%s] %s\n" %
 
31
                                  (self.address_string(),
 
32
                                   self.log_date_time_string(),
 
33
                                   format%args))
68
34
 
69
35
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
70
36
    def __init__(self, server_address, RequestHandlerClass, test_case):
72
38
                                                RequestHandlerClass)
73
39
        self.test_case = test_case
74
40
 
75
 
 
76
41
class TestCaseWithWebserver(TestCaseInTempDir):
77
42
    """Derived class that starts a localhost-only webserver
78
43
    (in addition to what TestCaseInTempDir does).
105
70
 
106
71
        self._http_base_url = 'http://localhost:%s/' % port
107
72
        self._http_starting.release()
108
 
        httpd.socket.settimeout(0.1)
 
73
        httpd.socket.settimeout(1)
109
74
 
110
75
        while self._http_running:
111
76
            try:
130
95
        return self._http_base_url + remote_path
131
96
 
132
97
    def setUp(self):
133
 
        TestCaseInTempDir.setUp(self)
 
98
        super(TestCaseWithWebserver, self).setUp()
134
99
        import threading, os
135
100
        self._local_path_parts = self.test_dir.split(os.path.sep)
136
101
        self._http_starting = threading.Lock()
140
105
        self._http_thread = threading.Thread(target=self._http_start)
141
106
        self._http_thread.setDaemon(True)
142
107
        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
108
 
147
109
    def tearDown(self):
148
110
        self._http_running = False
149
111
        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
 
 
 
112
        super(TestCaseWithWebserver, self).tearDown()