использование wsgiref.simple_server в юнит-тестах

Question

У меня есть несколько функций, подобных этой:

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 () в другом потоке? или может быть есть другой решение?

EDIT: Я хочу протестировать функцию "func", а не "app_200_hello"

Answer 1

Если вы тестируете приложение 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, это будет реальный сервер, который вам, вероятно, не нужно запускать -стоп, как это.

Answer 2

Использование многопроцессорной обработки для запуска сервера в отдельном процессе

в настройке сделать что-то вроде:

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 (), то в последующих экземплярах сервера может возникнуть проблема с уже используемым портом.

Answer 3

Ваш сервер должен быть отдельным процессом.

Вы захотите начать с subprocess.Popen()

Если вы используете Python 2.6, вы можете убить подпроцесс во время tearDown.

def setUp( self ):
    self.server= subprocess.Popen( "python","myserver","etc." )
def tearDown( self ):
    self.server.kill()

Если вы не используете Python 2.6, уничтожение сервера может быть неприятным.

Answer 4

Вы также можете предоставить фиктивную версию urlopen, которая фактически не запускает сервер.

Если исходный код был в mycode.py, в тестовом коде вы бы сделали что-то вроде:

import mycode

class TestFunc(unittest.TestCase):
    def setUp(self):
        # patch mycode, to use mock version of urlopen
        self._original_urlopen = mycode.urlopen
        mycode.urlopen=self.mock_urlopen

    def tearDown(self):
        # unpatch urlopen
        mycode.urlopen=self._original_urlopen

    def mock_urlopen(self,url):
        # return whatever data you need urlopen to return

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

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

Это называется «исправлением обезьян», которое некоторые недовольны, но для тестирования вашего кода это может значительно облегчить жизнь, так как вам не нужно менять исходный код, чтобы сделать его тестируемым.

Answer 5

Мое решение:

URL = 'http://localhost:8085'
def func():
    response = urlopen(URL)
    return response.read()

import unittest
from wsgiref.simple_server import WSGIServer, WSGIRequestHandler
import threading
from urllib2 import urlopen
from cStringIO import StringIO

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

server = WSGIServer(('localhost', 8085), WSGIRequestHandler)
server.set_app(app_200_hello)

t = threading.Thread(target=server.serve_forever)
t.start()

class TestFunc(unittest.TestCase):
    def setUp(self):
        pass

    def test1(self):
        r = func()
        self.assertEqual(r, 'Hello world')

    def __del__(self):
        server.shutdown()

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

Я запускаю «сервер» в другом потоке и выключаю его в деструкторе TestFunc.