代码之家  ›  专栏  ›  技术社区  ›  moin moin

如何在pytest中为异步IO代码编写fixture

  •  3
  • moin moin  · 技术社区  · 8 年前

    我想将asyncio与pytest一起使用。

    以下是我想做的:

    • 在测试运行时运行服务器-在测试运行时停止服务器 完成
    • 在一个完美的世界中,我会将服务器实现为固定设备(使用yield)

    我喜欢这样编写测试代码:

    def test_add(svr_fixture):
        await asyncio.sleep(100)
        assert m.add(1, 2) == 3   # I like the readability of this and want to restore it
    

    我试着用 pytest异步 ( https://pypi.python.org/pypi/pytest-asyncio )但我不知道怎么做。

    我在这个测试中想到了什么(有效,但它看起来很笨拙,掩盖了测试的意图):

    def test_add():
        async def do_it():
            await asyncio.sleep(100)
            return m.add(1, 2)
    
        loop = asyncio.get_event_loop()
        coro = loop.create_server(server.ServerProtocol, '127.0.0.1', 8023)
        asyncio.async(coro)
        res = loop.run_until_complete(do_it())
        assert res == 3
    

    如果您有任何关于如何将服务器代码提取到夹具(如文档链接或示例)中的帮助,我们将不胜感激。

    我认为不需要完整的服务器代码(但这里有: https://stackoverflow.com/a/48277838/570293 )

    1 回复  |  直到 8 年前
        1
  •  0
  •   moin moin    8 年前

    正如我在问题中指出的那样,我不希望异步的东西膨胀我的测试用例。到目前为止,我能找到的唯一简单可行的解决方案是使用多处理。我理解这个过程。terminate()不是结束异步IO循环的“最佳方式”,但至少它工作可靠。

    # -*- coding: utf-8 -*-
    import time
    from multiprocessing import Process
    
    import pytest
    from my_server import server
    
    
    @pytest.fixture
    def fake_server():
        p = Process(target=server.run, args=())
        p.start()
    
        yield
        p.terminate()
    
    
    def test_add2(fake_server):
        time.sleep(30)
        assert m.add(1, 2) == 3