websocket pytest：如何在每次测试后终止FastAPI TestClient？

我正在测试FastAPI应用程序，该应用程序将永远保持开放的WebSocket连接。根据设计，只有客户端应该关闭WebSocket。在测试执行之后，pytest并没有完成，它永远挂起。
1.测试这样一个应用程序的正确方法是什么？
1.在测试执行之后，有没有什么方法可以终止fixture？
范例：

$ pytest -k test_comedy_tom_hanks_async
======================================================================= test session starts ========================================================================
platform linux -- Python 3.11.5, pytest-7.4.2, pluggy-1.3.0 -- /home/gbajson/.cache/pypoetry/virtualenvs/vox-fdgLK-f1-py3.11/bin/python
cachedir: .pytest_cache
rootdir: /storage/amoje/Sync/area22/vox-async
configfile: pytest.ini
plugins: timeout-2.1.0, anyio-3.7.1, asyncio-0.21.1
asyncio: mode=Mode.STRICT
collected 13 items / 9 deselected / 4 selected                                                                                                                     
run-last-failure: no previously failed tests, not deselecting items.

src/tests/test_main.py::test_comedy_tom_hanks_async[asyncio+uvloop-3200.0-0.1] PASSED 

<<< HANGING FOREVER >>>

以下是我用于测试的夹具：

@pytest.fixture(scope="function", name="websocket")
async def fixture_ws_audio():
    client = TestClient(app)
    with client.websocket_connect("/ws/audio") as websocket:
        yield websocket

我已经试过了：

@pytest.mark.timeout(20)
我添加了以下fixture，以便在每次测试之后执行一些操作，但是pytest永远不会到达logger.info("Test finished.")

@pytest.fixture(autouse=True)
def run_before_and_after_tests():
    """Fixture to execute asserts before and after a test is run"""
    # Setup: fill with any logic you want
    logger.info("Starting test.")
    yield  # this is where the testing happens
    logger.info("Test finished.")

编辑2023-10-09

后端和测试代码都使用asyncio.to_thread调用，例如：

try:
            coro = asyncio.to_thread(read_from_ws, websocket)
            coro_waited = asyncio.wait_for(coro, timeout)
            results = await asyncio.gather(coro_waited)
            text = results[0]

        except asyncio.TimeoutError:
            logger.info("asyncio.TimeoutError: text: %s", text)
            break

当我在执行所有测试后终止pytest会话时，我看到：

======== 4 passed, 5 deselected in 77.85s (0:01:17) ========
^CException ignored in: <module 'threading' from '/usr/lib/python3.11/threading.py'>
Traceback (most recent call last):
  File "/usr/lib/python3.11/threading.py", line 1553, in _shutdown
    atexit_call()
  File "/usr/lib/python3.11/concurrent/futures/thread.py", line 31, in _python_exit
    t.join()
  File "/usr/lib/python3.11/threading.py", line 1112, in join
    self._wait_for_tstate_lock()
  File "/usr/lib/python3.11/threading.py", line 1132, in _wait_for_tstate_lock
    if lock.acquire(block, timeout):
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
KeyboardInterrupt:

这看起来像是在测试功能完成后websocket连接打开的问题，pytest在退出之前等待所有fixture完成。
TestClient的WebSocket_connect方法返回一个上下文管理器，当与with语句一起使用时，它将关闭WebSocket connection upon exiting. However, since you are using it in a fixture with yield, the websocket is not getting closed untill all the tests that use the fixture have completed.解决这个问题的一种方法是使用FastAPI的依赖注入系统来确保WebSocket在测试完成后关闭。举例来说：

@pytest.fixture(scope="function", name="websocket")
async def fixture_ws_audio():
    client = TestClient(app)
    websocket = client.websocket_connect("/ws/audio")
    yield websocket
    await websocket.close()

通过在yield之后添加await WebSocket.close（）行，您可以确保测试完成后WebSocket连接关闭。
你也可以使用request.addfinalizer，一个为fixtures添加清理代码的方法。

@pytest.fixture(scope="function", name="websocket")
def fixture_ws_audio(request):
    client = TestClient(app)
    websocket = client.websocket_connect("/ws/audio")
    def fin():
        await websocket.close()
    request.addfinalizer(fin)  # adding a finalizer to close the websocket
    return websocket

websocket pytest：如何在每次测试后终止FastAPI TestClient？

1条答案

相关问题

热门标签

最新问答