在单元测试中使用wsgiref.simple_server

我有一些像这样的功能:

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

我想用unittest测试它.

我做了这样的事情:

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()

在setUp()我的测试停止,因为s.handle_request()等待请求.我怎么能绕过那个？在另一个线程中运行s.handle_request()？或者还有另一种解决方案？

编辑:我想测试"func"函数,而不是"app_200_hello"

如果您正在测试WSGI应用程序,我强烈建议werkzeug.test通过在没有服务器的情况下测试应用程序本身来解决这些问题:

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'

这种方法完全消除了对WSGI服务器的需求.

当然,如果您确实想要启动服务器,则必须使用单独的线程或进程,但之后必须确保停止它.然而,令我印象深刻的是,您唯一想要使用真实服务器进行测试的是生产集成测试,在这种情况下您的服务器将不是wsgiref,它将是一个真正的服务器,您可能不必启动 - 像这样停下来.

使用多处理在单独的进程中启动服务器

在setUp做类似的事情:

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()

然后在tearDown做:

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

我发现如果你没有明确地将del()放在那里,那么后续的服务器实例可能会遇到已经使用的端口的问题.