← 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-10-02 13:14:12 UTC
  • mto: This revision was merged to the branch mainline in revision 1414.
  • Revision ID: jelmer@samba.org-20051002131412-c44d6cd10336ac94
Add test for empty commit messages.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/HTTPTestUtil.py
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
import BaseHTTPServer, SimpleHTTPServer, socket, errno, time
18
 
from bzrlib.tests import TestCaseInTempDir
19
 
from bzrlib.osutils import relpath
 
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)
 
30
        self.server.test_case.log("webserver - %s - - [%s] %s" %
 
31
                                  (self.address_string(),
 
32
                                   self.log_date_time_string(),
 
33
                                   format%args))
31
34
 
32
35
    def handle_one_request(self):
33
36
        """Handle a single HTTP request.
41
44
            try:
42
45
                self.raw_requestline = self.rfile.readline()
43
46
            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)
 
47
                if e.args[0] == errno.EAGAIN:
 
48
                    self.log_message('EAGAIN (%d) while reading from raw_requestline' % i)
49
49
                    time.sleep(0.01)
50
50
                    continue
51
51
                raise
113
113
    def get_remote_url(self, path):
114
114
        import os
115
115
 
 
116
        path_parts = path.split(os.path.sep)
116
117
        if os.path.isabs(path):
117
 
            remote_path = relpath(self.test_dir, path)
 
118
            if path_parts[:len(self._local_path_parts)] != \
 
119
                   self._local_path_parts:
 
120
                raise BadWebserverPath(path, self.test_dir)
 
121
            remote_path = '/'.join(path_parts[len(self._local_path_parts):])
118
122
        else:
119
 
            remote_path = path
 
123
            remote_path = '/'.join(path_parts)
120
124
 
121
125
        self._http_starting.acquire()
122
126
        self._http_starting.release()
125
129
    def setUp(self):
126
130
        TestCaseInTempDir.setUp(self)
127
131
        import threading, os
 
132
        self._local_path_parts = self.test_dir.split(os.path.sep)
128
133
        self._http_starting = threading.Lock()
129
134
        self._http_starting.acquire()
130
135
        self._http_running = True
132
137
        self._http_thread = threading.Thread(target=self._http_start)
133
138
        self._http_thread.setDaemon(True)
134
139
        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
140
 
139
141
    def tearDown(self):
140
142
        self._http_running = False
141
143
        self._http_thread.join()
142
 
        if self._http_proxy is not None:
143
 
            import os
144
 
            os.environ["http_proxy"] = self._http_proxy
145
144
        TestCaseInTempDir.tearDown(self)
146
145