← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/importd/tests/helpers.py

merge from RF

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/importd/tests/helpers.py
5
5
import shutil
6
6
import unittest
7
7
 
8
 
from SimpleHTTPServer import SimpleHTTPRequestHandler
9
 
from BaseHTTPServer import HTTPServer
10
 
import socket
11
 
 
12
8
from canonical.launchpad.ftests import harness
13
9
from canonical.ftests import pgsql
14
10
 
32
28
    'ZopelessUtilitiesHelper',
33
29
    'ZopelessTestCase',
34
30
    'JobTestCase',
35
 
    'WebserverTestCase',
36
31
    ]
37
32
 
38
33
 
85
80
        return job
86
81
 
87
82
 
88
 
def test_path(name):
89
 
    test_dir = os.path.dirname(__file__)
90
 
    relpath = os.path.join(test_dir, name)
91
 
    return os.path.abspath(relpath)
92
 
 
93
 
 
94
83
class ZopelessHelper(harness.LaunchpadZopelessTestSetup):
95
84
    dbuser = 'importd'
96
85
 
163
152
        self.zopeless_helper.tearDown()
164
153
 
165
154
 
166
 
# Webserver helper was based on the bzr code for doing http tests.
167
 
 
168
 
class BadWebserverPath(ValueError):
169
 
    def __str__(self):
170
 
        return 'path %s is not in %s' % self.args
171
 
 
172
 
class TestHTTPRequestHandler(SimpleHTTPRequestHandler):
173
 
    def log_message(self, *args):
174
 
        pass
175
 
 
176
 
 
177
 
class WebserverHelper(SandboxHelper):
178
 
 
179
 
    def _http_start(self):
180
 
        httpd = HTTPServer(('localhost', 0), TestHTTPRequestHandler)
181
 
        host, port = httpd.socket.getsockname()
182
 
 
183
 
        self._http_base_url = 'http://localhost:%s/' % port
184
 
        self._http_starting.release()
185
 
        httpd.socket.settimeout(0.1)
186
 
 
187
 
        while self._http_running:
188
 
            try:
189
 
                httpd.handle_request()
190
 
            except socket.timeout:
191
 
                pass
192
 
 
193
 
    def get_remote_url(self, path):
194
 
        import os
195
 
 
196
 
        path_parts = path.split(os.path.sep)
197
 
        if os.path.isabs(path):
198
 
            if path_parts[:len(self._local_path_parts)] != \
199
 
                   self._local_path_parts:
200
 
                raise BadWebserverPath(path, self.path)
201
 
            remote_path = '/'.join(path_parts[len(self._local_path_parts):])
202
 
        else:
203
 
            remote_path = '/'.join(path_parts)
204
 
 
205
 
        self._http_starting.acquire()
206
 
        self._http_starting.release()
207
 
        return self._http_base_url + remote_path
208
 
 
209
 
    def setUp(self):
210
 
        SandboxHelper.setUp(self)
211
 
        import threading, os
212
 
        self._local_path_parts = self.path.split(os.path.sep)
213
 
        self._http_starting = threading.Lock()
214
 
        self._http_starting.acquire()
215
 
        self._http_running = True
216
 
        self._http_base_url = None
217
 
        self._http_thread = threading.Thread(target=self._http_start)
218
 
        self._http_thread.setDaemon(True)
219
 
        self._http_thread.start()
220
 
        self._http_proxy = os.environ.get("http_proxy")
221
 
        if self._http_proxy is not None:
222
 
            del os.environ["http_proxy"]
223
 
 
224
 
    def tearDown(self):
225
 
        self._http_running = False
226
 
        self._http_thread.join()
227
 
        if self._http_proxy is not None:
228
 
            import os
229
 
            os.environ["http_proxy"] = self._http_proxy
230
 
        SandboxHelper.tearDown(self)