在单元测试中使用 wsgiref.simple_server

发布于 2024-07-14 19:35:59 字数 858 浏览 5 评论 0原文

我有一些像这样的功能:


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

我想用单元测试来测试它。

我做了这样的事情:


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”

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"

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。

评论(5

三月梨花 2024-07-21 19:35:59

如果您正在测试 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,它将是您可能不必启动的真实服务器-像这样停下来。

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.

时光倒影 2024-07-21 19:35:59

单独进程中启动服务器,

使用多处理在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()放在那里,则后续服务器实例可能会出现端口问题已经被使用了。

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.

黑色毁心梦 2024-07-21 19:35:59

您的服务器必须是一个单独的进程。

您需要使用 subprocess.Popen() 启动它。

如果您使用的是 Python 2.6,则可以在拆解期间终止子进程。

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

如果您不使用 Python 2.6,终止服务器可能会令人不快。

Your server must be a separate process.

You'll want to start it with subprocess.Popen()

If you're using Python 2.6, you can then kill the subprocess during tearDown.

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

If you're not using Python 2.6, killing the server can be unpleasant.

仙女 2024-07-21 19:35:59

您还可以提供一个并不实际运行服务器的 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()


这被称为“猴子修补”,有些人对此不以为然,但对于测试您的代码来说,它可以使生活变得有趣容易得多,因为您不需要更改原始代码即可使其可测试。

You could also provide a mock version of urlopen that doesn't actually run the server.

Assuming your original code was in mycode.py, in your test code you'd do something like:



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


This is known as "monkey patching" which some frown upon, but for testing your code it can make life a lot easier, as you don't need to change your original code to make it testable.

む无字情书 2024-07-21 19:35:59

我的解决方案:


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 析构函数中将其关闭。

My solution:


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


I start "server" in another thread and shutdown it at TestFunc destructor.

~没有更多了~
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文