Skip to content

Instantly share code, notes, and snippets.

@almet almet/gist:467129
Created Jul 7, 2010

What would you like to do?
"""Mocked PyPI Server implementation, to use in tests.
This module also provides a simple test case to extend if you need to use
the PyPIServer all along your test case. Be sure to read the documentation
before any use.
import Queue
import threading
import time
import urllib2
from BaseHTTPServer import HTTPServer
from SimpleHTTPServer import SimpleHTTPRequestHandler
import os.path
import select
from import unittest
PYPI_DEFAULT_STATIC_PATH = os.path.dirname(os.path.abspath(__file__)) + "/pypiserver"
def use_pypi_server(*server_args, **server_kwargs):
"""Decorator to make use of the PyPIServer for test methods,
just when needed, and not for the entire duration of the testcase.
def wrapper(func):
def wrapped(*args, **kwargs):
server = PyPIServer(*server_args, **server_kwargs)
func(server=server, *args, **kwargs)
return wrapped
return wrapper
class PyPIServerTestCase(unittest.TestCase):
def setUp(self):
super(PyPIServerTestCase, self).setUp()
self.pypi = PyPIServer()
def tearDown(self):
super(PyPIServerTestCase, self).tearDown()
class PyPIServer(threading.Thread):
"""PyPI Mocked server.
Provides a mocked version of the PyPI API's, to ease tests.
Support serving static content and serving previously given text.
def __init__(self, test_static_path=None,
static_filesystem_paths=["default"], static_uri_paths=["simple"]):
"""Initialize the server.
static_uri_paths and static_base_path are parameters used to provides
respectively the http_paths to serve statically, and where to find the
matching files on the filesystem.
self._run = True
self.httpd = HTTPServer(('', 0), PyPIRequestHandler)
self.httpd.RequestHandlerClass.log_request = lambda *_: None
self.httpd.RequestHandlerClass.pypi_server = self
self.address = (self.httpd.server_name, self.httpd.server_port)
self.request_queue = Queue.Queue()
self._requests = []
self.default_response_status = 200
self.default_response_headers = [('Content-type', 'text/plain')]
self.default_response_data = "hello"
# initialize static paths / filesystems
self.static_uri_paths = static_uri_paths
if test_static_path is not None:
self.static_filesystem_paths = [PYPI_DEFAULT_STATIC_PATH + "/" + path
for path in static_filesystem_paths]
def run(self):
# loop because we can't stop it otherwise, for python < 2.6
while self._run:
r, w, e =[self.httpd], [], [], 0.5)
if r:
def stop(self):
"""self shutdown is not supported for python < 2.6"""
self._run = False
def get_next_response(self):
return (self.default_response_status,
def requests(self):
"""Use this property to get all requests that have been made
to the server
while True:
except Queue.Empty:
return self._requests
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 'content-length' in self.headers.dict:
request_data =
request_data =
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 = []
relative_path = self.path
for fs_path in fs_paths:
if self.path.endswith("/"):
relative_path += "index.html"
file = open(fs_path + relative_path)
data =
if relative_path.endswith('.tar.gz'):
headers=[('Content-type', 'application/x-gtar')]
headers=[('Content-type', 'text/html')]
self.make_response(data, headers=headers)
except IOError:
if data is None:
self.make_response("Not found", 404)
# otherwise serve the content from get_next_response
# 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):
status = int(status)
except ValueError:
# we probably got something like YYY Codename.
# Just get the first 3 digits
status = int(status[:3])
for header, value in headers:
self.send_header(header, value)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.