pytestでsocketサーバへの接続が完了しない

# はじめに
タイトルのとおりですが、pytestに限った話ではないかもしれません。

# 経緯
tcpipクライアントをpytestするために、tcpipサーバーをテストコードで用意して実行したら、いつまでもテストが終了しない事象が発生した。


```python:test_server
"""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秒でタイムアウトしテスト失敗するようにしています。

実行すると以下の通り失敗します。

```sh
% 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を入れればよいです。

```python:test_server_fixed.py
        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`など非同期用のメソッドを使うのが王道だった気がします（[参考](https://qiita.com/sand/items/4f27e62c7491b8e85d12)）