python - pytest フィクスチャから asyncio サーバーを実行する

Question

pytest フィクスチャで asyncio を使用してサーバーを実行しようとしています

@pytest.fixture(autouse=True)
@pytest.mark.asyncio
async def start_endpoints(
    endpoint1: ServerEndpoint,
    endpoint2: ServerEndpoint
):
    pool = ThreadPoolExecutor(max_workers=2)
    loop = asyncio.get_running_loop()

    await loop.run_in_executor(pool, endpoint1.start)
    await loop.run_in_executor(pool, endpoint2.start)

start方法は以下のようなものです

async def start(self):
        try:

            server = await asyncio.start_server(self.handle_req, self.addr, self.port)
            addr = server.sockets[0].getsockname()
            print(f'{self.name}: serving on {addr}')

            async with server:
                await server.serve_forever()

サーバーとの接続を開こうとすると、テストはこのエラーを出力しますが、

self = <_WindowsSelectorEventLoop running=False closed=False debug=False>
fut = <Future finished exception=ConnectionRefusedError(10061, "Connect call failed ('127.0.0.1', 9000)")>
sock = <socket.socket [closed] fd=-1, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6>
address = ('127.0.0.1', 9000)

    def _sock_connect_cb(self, fut, sock, address):
        if fut.cancelled():
            return
    
        try:
            err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
            if err != 0:
                # Jump to any except clause below.
>               raise OSError(err, f'Connect call failed {address}')
E               ConnectionRefusedError: [Errno 10061] Connect call failed ('127.0.0.1', 9000)

編集:問題は、イベントループが直後に閉じられることです。そのため、すべてのフィクスチャをマークしようとしまし(scope="module")たが、今では

ScopeMismatch: You tried to access the 'function' scoped fixture 'event_loop' with a 'module' scoped request object, involved factories
test\e2e\test_peer.py:380:  def start_endpoints

EDIT2：

だから私はevent_loopフィクスチャを追加しました

@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

を使用して、各フィクスチャのデフォルト ループをオーバーライドする必要があります@pytest.mark.asyncio。

@pytest.fixture(autouse=True, scope="module")
@pytest.mark.asyncio
async def start_endpoints(
    event_loop,
    endpoint1: ServerEndpoint,
    endpoint2: ServerEndpoint
):
    pool = ThreadPoolExecutor(max_workers=2)

    await event_loop.run_in_executor(pool, endpoint1.start)
    await event_loop.run_in_executor(pool, endpoint2.start)

テスト内でデバッグすることにより、event_loopは内部に保存しているループと同じですServerEndpoint(つまりasyncio.get_running_loop()) ですが、まだ取得していますConnectionRefusedError