← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/HTTPTestUtil.py

remove diff.py shebang

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
19
 
from bzrlib.osutils import relpath
 
17
import BaseHTTPServer, SimpleHTTPServer
 
18
from bzrlib.selftest import TestCaseInTempDir
20
19
 
21
20
 
22
21
class WebserverNotAvailable(Exception):
23
22
    pass
24
23
 
 
24
class BadWebserverPath(ValueError):
 
25
    def __str__(self):
 
26
        return 'path %s is not in %s' % self.args
 
27
 
25
28
class TestingHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
26
29
    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()
 
30
        self.server.test_case.log("webserver - %s - - [%s] %s\n" %
 
31
                                  (self.address_string(),
 
32
                                   self.log_date_time_string(),
 
33
                                   format%args))
65
34
 
66
35
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
67
36
    def __init__(self, server_address, RequestHandlerClass, test_case):
69
38
                                                RequestHandlerClass)
70
39
        self.test_case = test_case
71
40
 
72
 
 
73
41
class TestCaseWithWebserver(TestCaseInTempDir):
74
42
    """Derived class that starts a localhost-only webserver
75
43
    (in addition to what TestCaseInTempDir does).
102
70
 
103
71
        self._http_base_url = 'http://localhost:%s/' % port
104
72
        self._http_starting.release()
105
 
        httpd.socket.settimeout(0.1)
 
73
        httpd.socket.settimeout(1)
106
74
 
107
75
        while self._http_running:
108
76
            try:
113
81
    def get_remote_url(self, path):
114
82
        import os
115
83
 
 
84
        path_parts = path.split(os.path.sep)
116
85
        if os.path.isabs(path):
117
 
            remote_path = relpath(self.test_dir, path)
 
86
            if path_parts[:len(self._local_path_parts)] != \
 
87
                   self._local_path_parts:
 
88
                raise BadWebserverPath(path, self.test_dir)
 
89
            remote_path = '/'.join(path_parts[len(self._local_path_parts):])
118
90
        else:
119
 
            remote_path = path
 
91
            remote_path = '/'.join(path_parts)
120
92
 
121
93
        self._http_starting.acquire()
122
94
        self._http_starting.release()
123
95
        return self._http_base_url + remote_path
124
96
 
125
97
    def setUp(self):
126
 
        TestCaseInTempDir.setUp(self)
 
98
        super(TestCaseWithWebserver, self).setUp()
127
99
        import threading, os
 
100
        self._local_path_parts = self.test_dir.split(os.path.sep)
128
101
        self._http_starting = threading.Lock()
129
102
        self._http_starting.acquire()
130
103
        self._http_running = True
132
105
        self._http_thread = threading.Thread(target=self._http_start)
133
106
        self._http_thread.setDaemon(True)
134
107
        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
108
 
139
109
    def tearDown(self):
140
110
        self._http_running = False
141
111
        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
 
 
 
112
        super(TestCaseWithWebserver, self).tearDown()