[Python-checkins] distutils2: Replace wsgiref by BaseHTTPServer to be python 2.4 compatible.
tarek.ziade
python-checkins at python.org
Sun Jul 4 11:48:39 CEST 2010
tarek.ziade pushed d4fe9695bc65 to distutils2:
http://hg.python.org/distutils2/rev/d4fe9695bc65
changeset: 297:d4fe9695bc65
user: Alexis Metaireau <ametaireau at gmail.com>
date: Thu Jun 03 17:44:59 2010 +0200
summary: Replace wsgiref by BaseHTTPServer to be python 2.4 compatible.
files: src/distutils2/tests/pypi_server.py, src/distutils2/tests/test_pypi_server.py
diff --git a/src/distutils2/tests/pypi_server.py b/src/distutils2/tests/pypi_server.py
--- a/src/distutils2/tests/pypi_server.py
+++ b/src/distutils2/tests/pypi_server.py
@@ -10,6 +10,8 @@
import time
import unittest2
import urllib2
+from BaseHTTPServer import HTTPServer
+from SimpleHTTPServer import SimpleHTTPRequestHandler
from wsgiref.simple_server import make_server
import os.path
@@ -39,7 +41,6 @@
super(PyPIServerTestCase, self).tearDown()
self.pypi.stop()
-
class PyPIServer(threading.Thread):
"""PyPI Mocked server.
Provides a mocked version of the PyPI API's, to ease tests.
@@ -57,14 +58,15 @@
"""
threading.Thread.__init__(self)
self._run = True
- self.httpd = make_server('', 0, self.pypi_app)
+ self.httpd = HTTPServer(('', 0), PyPIRequestHandler)
self.httpd.RequestHandlerClass.log_request = lambda *_: None
+ self.httpd.RequestHandlerClass.pypi_server = self
self.address = self.httpd.server_address
self.request_queue = Queue.Queue()
self._requests = []
- self.default_response_status = "200 OK"
+ self.default_response_status = 200
self.default_response_headers = [('Content-type', 'text/plain')]
- self.default_response_data = ["hello"]
+ self.default_response_data = "hello"
self.static_uri_paths = static_uri_paths
if test_static_path is not None:
static_filesystem_paths.append(test_static_path)
@@ -82,59 +84,6 @@
"""self shutdown is not supported for python < 2.6"""
self._run = False
- def pypi_app(self, environ, start_response):
- """Serve the content.
-
- Also record the requests to be accessed later. If trying to access an
- url matching a static uri, serve static content, otherwise serve
- what is provided by the `get_next_response` method.
- """
- # record the request. Read the input only on PUT or POST requests
- if environ["REQUEST_METHOD"] in ("PUT", "POST"):
- if environ.get("CONTENT_LENGTH"):
- request_data = environ.pop('wsgi.input').read(
- int(environ['CONTENT_LENGTH']))
- else:
- request_data = environ.pop('wsgi.input').read()
- elif environ["REQUEST_METHOD"] in ("GET", "DELETE"):
- request_data = ''
-
- self.request_queue.put((environ, request_data))
-
- # serve the content from local disc if we request an URL beginning
- # by a pattern defined in `static_paths`
- relative_path = environ["PATH_INFO"].replace(self.full_address, '')
- url_parts = relative_path.split("/")
- if len(url_parts) > 1 and url_parts[1] in self.static_uri_paths:
- data = None
- # always take the last first.
- fs_paths = []
- fs_paths.extend(self.static_filesystem_paths)
- fs_paths.reverse()
- for fs_path in fs_paths:
- try:
- if relative_path.endswith("/"):
- relative_path += "index.html"
- print fs_path + relative_path
- file = open(fs_path + relative_path)
- data = file.read()
- start_response("200 OK", [('Content-type', 'text/html')])
- except IOError:
- pass
-
- if data is None:
- start_response("404 NOT FOUND",
- [('Content-type', 'text/html')])
- data = "Not Found"
- return data
-
- # otherwise serve the content from get_next_response
- else:
- # send back a response
- status, headers, data = self.get_next_response()
- start_response(status, headers)
- return data
-
def get_next_response(self):
return (self.default_response_status,
self.default_response_headers,
@@ -155,3 +104,83 @@
@property
def full_address(self):
return "http://%s:%s" % self.address
+
+
+class PyPIRequestHandler(SimpleHTTPRequestHandler):
+ # we need to access the pypi server while serving the content
+ pypi_server = None
+
+ def do_POST(self):
+ return self.serve_request()
+ def do_GET(self):
+ return self.serve_request()
+ def do_DELETE(self):
+ return self.serve_request()
+ def do_PUT(self):
+ return self.serve_request()
+
+ def serve_request(self):
+ """Serve the content.
+
+ Also record the requests to be accessed later. If trying to access an
+ url matching a static uri, serve static content, otherwise serve
+ what is provided by the `get_next_response` method.
+ """
+ # record the request. Read the input only on PUT or POST requests
+ if self.command in ("PUT", "POST"):
+ if self.headers.dict.has_key("content-length"):
+ request_data = self.rfile.read(
+ int(self.headers['content-length']))
+ else:
+ request_data = self.rfile.read()
+ elif self.command in ("GET", "DELETE"):
+ request_data = ''
+
+ self.pypi_server.request_queue.put((self, request_data))
+
+ # serve the content from local disc if we request an URL beginning
+ # by a pattern defined in `static_paths`
+ url_parts = self.path.split("/")
+ if (len(url_parts) > 1 and
+ url_parts[1] in self.pypi_server.static_uri_paths):
+ data = None
+ # always take the last first.
+ fs_paths = []
+ fs_paths.extend(self.pypi_server.static_filesystem_paths)
+ fs_paths.reverse()
+ relative_path = self.path
+ for fs_path in fs_paths:
+ try:
+ if self.path.endswith("/"):
+ relative_path += "index.html"
+ file = open(fs_path + relative_path)
+ data = file.read()
+ self.make_response(data)
+ except IOError:
+ pass
+
+ if data is None:
+ self.make_response("Not found", 404)
+
+ # otherwise serve the content from get_next_response
+ else:
+ # send back a response
+ status, headers, data = self.pypi_server.get_next_response()
+ self.make_response(data, status, headers)
+
+ def make_response(self, data, status=200,
+ headers=[('Content-type', 'text/html')]):
+ """Send the response to the HTTP client"""
+ if not isinstance(status, int):
+ try:
+ status = int(status)
+ except ValueError:
+ # we probably got something like YYY Codename.
+ # Just get the first 3 digits
+ status = int(status[:3])
+
+ self.send_response(status)
+ for header, value in headers:
+ self.send_header(header, value)
+ self.end_headers()
+ self.wfile.write(data)
diff --git a/src/distutils2/tests/test_pypi_server.py b/src/distutils2/tests/test_pypi_server.py
--- a/src/distutils2/tests/test_pypi_server.py
+++ b/src/distutils2/tests/test_pypi_server.py
@@ -21,9 +21,11 @@
request = urllib2.Request(server.full_address, data, headers)
urllib2.urlopen(request)
self.assertEqual(len(server.requests), 1)
- environ, request_data = server.requests[-1]
+ handler, request_data = server.requests[-1]
self.assertIn("Rock Around The Bunker", request_data)
- self.assertEqual(environ["HTTP_X_TEST_HEADER"], "Mister Iceberg")
+ self.assertTrue(handler.headers.dict.has_key("x-test-header"))
+ self.assertEqual(handler.headers.dict["x-test-header"],
+ "Mister Iceberg")
server.stop()
def test_serve_static_content(self):
--
Repository URL: http://hg.python.org/distutils2
More information about the Python-checkins
mailing list