How to create a temporary WSGI app suitable to unit tests mocking

testwsgimock.py

# -*- coding: utf-8 -*-

# If you need to test a REST client, this is a Python 2 recipe that runs a
# simple WSGI app for your tests. Any improvement suggestion is welcome.

# Run this with "python -m unittest testingwsgi"

# Put this in a testing resources module, say tests/resources.py

import os
import select
import socket
import threading
import unittest
import wsgiref.simple_server


class ThreadedServerControl(object):
    """Will provide a temporary test server in another thread for your
    application.
    """
    __stop_marker = 'stop'

    def __init__(self, app, host='localhost', port=8888):
        """Instance initialization
        Args:
            app (function): A WSGI application.
            host (str): Listening hostname or IP. 'localhost' and '0.0.0.0' do the job.
            port (int): Listening port preferably >= 1024 unless you're root
        """
        self.app = app
        self.host = host
        self.port = port

        # Communication pipe with the thread
        self.stop_read, self.stop_write = os.pipe()
        self.started = False
        return

    def __run(self):
        httpd = wsgiref.simple_server.make_server(self.host, self.port, self.app)

        # We don't want logs in the console
        log_request = httpd.RequestHandlerClass.log_request
        no_logging = lambda *args, **kwargs: None
        httpd.RequestHandlerClass.log_request = no_logging

        # Notify / unlock self.start()
        self.ready.set()
        while True:
            ready, dummy, dummy = select.select(
                [httpd, self.stop_read], [self.stop_write], []
            )
            # HTTP client request detected ?
            if httpd in ready:
                httpd.handle_request()

            # self.stop() synch called ?
            if self.stop_read in ready:
                os.read(self.stop_read, len(self.__stop_marker))
                # Re-enable console logging and exit
                httpd.RequestHandlerClass.log_request = log_request
                break

    def start(self):
        """Launches the server in a thread
        """
        # Bounce protection
        if self.started:
            return

        # Threaded server and synch setup
        self.ready = threading.Event()
        self.server_thread = threading.Thread(target=self.__run)
        self.server_thread.start()

        # Wait server readyness (if a client runs before -> raise URLError)
        self.ready.wait()
        self.started = True
        return

    def stop(self):
        """Stops and kills the server and thread
        """
        # Bounce protection
        if not self.started:
            return

        # Notify thread's hara kiri
        os.write(self.stop_write, self.__stop_marker)

        # Cleanup after thread's hara kiri
        self.server_thread.join()
        os.close(self.stop_write)
        os.close(self.stop_read)
        self.started = False
        return

def get_free_port():
    """Finds an available TCP/IP port"""
    s = socket.socket(socket.AF_INET, type=socket.SOCK_STREAM)
    s.bind(('localhost', 0))
    address, port = s.getsockname()
    s.close()
    return port

# Now this is in place, here is a sample test, say in tests/test_something.py
# ---------------------------------------------------------------------------

from urllib2 import urlopen
# from .testing import get_free_port, ThreadedServerControl

# Prepare your mocking WSGI application

def wsgi_application(environ, start_response):
    """Mock a real application"""
    start_response("200 OK", [('Content-Type', 'text/plain')])
    return "Hello World"


# And finally here's a simple test

class TestSomething(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.port = get_free_port()  # Or choose an arbitrary port litteral
        cls.base_url = 'http://localhost:{}'.format(cls.port)
        cls.server = ThreadedServerControl(wsgi_application, port=cls.port)
        cls.server.start()

    @classmethod
    def tearDownClass(cls):
        cls.server.stop()

    def test_basic(self):
        url = self.base_url + '/some/path'
        response  = urlopen(url)
        self.assertEqual(response.read(), "Hello World")

Thanks! Works great