はじめに
タイトルのとおりですが、pytestに限った話ではないかもしれません。
経緯
tcpipクライアントをpytestするために、tcpipサーバーをテストコードで用意して実行したら、いつまでもテストが終了しない事象が発生した。
"""pytest-asyncioの動作検証用の最小限のテストコード"""
import asyncio
import socket
from typing import AsyncGenerator, Tuple
import pytest
import pytest_asyncio
@pytest_asyncio.fixture
async def simple_server() -> AsyncGenerator[Tuple[socket.socket, bool], None]:
"""最小限の機能を持つTCPサーバーフィクスチャ"""
print("Server: フィクスチャ開始")
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("localhost", 12346))
server_socket.listen(1)
server_socket.settimeout(0.1)
print("Server: ソケット設定完了")
async def handle_client():
print("Server: handle_client開始")
while True:
try:
print("Server: 接続待ち")
client, _ = server_socket.accept()
print("Server: クライアント接続")
client.close()
except socket.timeout:
print("Server: タイムアウト")
continue
print("Server: タスク作成前")
server_task = asyncio.create_task(handle_client())
print("Server: タスク作成完了")
try:
yield server_socket, True
finally:
print("Server: クリーンアップ開始")
server_task.cancel()
server_socket.close()
print("Server: クリーンアップ完了")
@pytest.mark.asyncio
@pytest.mark.timeout(1, method="thread")
async def test_simple_server(simple_server):
"""サーバーフィクスチャのテスト"""
print("Test: テスト開始")
server_socket, is_ready = simple_server
print("Test: フィクスチャ取得完了")
# 簡単な接続テスト
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
await asyncio.sleep(0.1)
print("Test: クライアント接続試行")
client.connect(("localhost", 12346))
print("Test: クライアント接続成功")
client.close()
print("Test: テスト完了")
serverにclientが接続したら成功というテストです。
期待としては一瞬で終わる処理です。pytest-timeoutを用いて、1秒でタイムアウトしテスト失敗するようにしています。
実行すると以下の通り失敗します。
% uv run pytest test_server.py
============================================ test session starts =============================================
platform darwin -- Python 3.11.2, pytest-8.3.4, pluggy-1.5.0
rootdir:
configfile: pyproject.toml
plugins: asyncio-0.25.3, anyio-4.8.0, timeout-2.3.1
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=function
collected 1 item
test_server.py ++++++++++++++++++++++++++++++++++++++++++++++++++ Timeout +++++++++++++++++++++++++++++++++++++++++++++++++++
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Captured stdout ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Server: フィクスチャ開始
Server: ソケット設定完了
Server: タスク作成前
Server: タスク作成完了
Server: handle_client開始
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
Server: タイムアウト
Server: 接続待ち
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Stack of MainThread (8591880768) ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Server: タイムアウト
Server: 接続待ち
...
エラーの出力を見ると「接続待ち」と「タイムアウト」が繰り返されていますので、server_socket.accept()の失敗を繰り返しているようです。また、フィクスチャのprintのみ実行されており、テストのprintが実行されていないことから、テストにはいる手前で処理が止まっているようです。
対策
フィクスチャの接続待ちのループでsleepを入れればよいです。
while True:
try:
print("Server: 接続待ち")
client, _ = server_socket.accept()
print("Server: クライアント接続")
client.close()
except socket.timeout:
print("Server: タイムアウト")
await asyncio.sleep(0.01)
continue
実行してみます。
% uv run pytest test_server.py
============================================ test session starts =============================================
platform darwin -- Python 3.11.2, pytest-8.3.4, pluggy-1.5.0
rootdir:
configfile: pyproject.toml
plugins: asyncio-0.25.3, anyio-4.8.0, timeout-2.3.1
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=function
collected 1 item
test_server.py . [100%]
============================================= 1 passed in 0.13s ==============================================
% uv run pytest test_server_fixed.py
============================================ test session starts =============================================
platform darwin -- Python 3.11.2, pytest-8.3.4, pluggy-1.5.0
rootdir:
configfile: pyproject.toml
plugins: asyncio-0.25.3, anyio-4.8.0, timeout-2.3.1
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=function
collected 1 item
test_server_fixed.py . [100%]
============================================= 1 passed in 0.24s ==============================================
成功しました。
考察
sleepなしだとsocket.acceptが非常に高頻度に実行されるためcpuを占有しテストが開始できなかったのだと思われます。
非同期で実行しているので、同期的に処理をストップしているわけではないのですが、非同期と言っても１つのcpuを複数プロセスで交互に使うようなイメージなので、非常にcpu負荷の高いプロセスが一つあると、止まってしまうことが起こるのだと思われます。
もう一つ、accept自体が待機処理であり、server_socket.settimeoutの指定で0.1秒は待機するような設定にしていたので、別途sleepを入れる必要があることも気づきにくかったです。深く調べてないですが、おそらくacceptの待機は同期の待機なのだと思われます。同期のsleepはcpuを占有するため、cpu負荷を下げるためのsleepとしてはasyncio.sleepが必要ということなのかもしれません。
while Trueで待機するときはintervalのsleepを含めることを忘れないようにしましょう。
メインプログラムのコードだと習慣づいているのですが、テスト用に生成AIの助けも借りながら作成したコードだったため、原因に気づくのに遅れてしまいました。
意識付けるためにメモとして残しておきます。
ここloop.sock_acceptなど非同期用のメソッドを使うのが王道だった気がします（参考）