using wsgiref.simple_server in unittests

Question

I have some functions like this one:

URL = 'http://localhost:8080'
def func():
    response = urlopen(URL)
    return process(response)

And i want to test it with unittest.

I did something like this:

from wsgiref.simple_server import make_server
def app_200_hello(environ,start_response):
    stdout = StringIO('Hello world')
    start_response("200 OK", [('Content-Type','text/plain')])
    return [stdout.getvalue()]

s = make_server('localhost', 8080, app_200_hello)

class TestFunc(unittest.TestCase):
    def setUp(self):
        s.handle_request()

    def test1(self):
        r = func()
        assert r, something

if __name__ == '__main__':
    unittest.main()

At setUp() my tests stop because s.handle_request() waits for request. How i can get around that? Run s.handle_request() in another thread? or maybe there is another solution?

EDIT: I want to test "func" function, not "app_200_hello"

Answer 1

If you are testing a WSGI application, I can strongly recommend werkzeug.test which gets around these issues by testing the application itself without a server:

from werkzeug.test import Client

# then in your test case
def test1(self):
    client = Client(app_200_hello)
    appiter, status, headers = client.open()
    assert ''.join(appiter) == 'Hello World'
    assert status == '200 OK'

This approach just removes the need for a WSGI server altogether.

Of course if you did want to start a server, you would have to use a separate thread or a process, but then you have to make sure you stop it afterwards. However, it strikes me that the only time you would want to test with a real server is for production integration testing, in which case your server won't be wsgiref, it will be a real server that you probably won't have to start-stop like this.

Answer 2

Use multiprocessing to start the server in a separate process

in setUp do something like:

self.port = 8000
server = make_server('', self.port, make_my_wsgi_ap())
self.server_process = multiprocessing.Process(target=server.serve_forever)
self.server_process.start()

then in tearDown do:

self.server_process.terminate()
self.server_process.join()
del(self.server_process)

I've found that if you don't explicitly put the del() in there then subsequent server instances may have a problem with the port already being used.