Asyncio PyTest RuntimeError：无运行事件循环问题求助

为基于Python websockets库开发的游戏编写测试用例，运行test_game_started测试时抛出RuntimeError: no running event loop，但测试本身却能通过。推测listen1()和listen2()任务在事件循环关闭后仍持续运行，尝试手动终止任务后报错依旧，寻求解决方案。

HelpTest类代码 #

class HelpTest:
    def __init__(self, loop):
        self.loop = loop
        self.ws1 = None
        self.ws2 = None
        self.msg1 = None
        self.msg2 = None
        self.tasks = []

    async def listen1(self):
        async with websockets.connect(f"ws://localhost:27000") as websocket:
            self.ws1 = websocket
            while True:
                try:
                    self.msg1 = await websocket.recv()
                    await asyncio.sleep(0.1)
                except RuntimeError:
                    break

    async def listen2(self):
        async with websockets.connect(f"ws://localhost:27000") as websocket:
            self.ws2 = websocket
            while True:
                try:
                    self.msg2 = await websocket.recv()
                    await asyncio.sleep(0.1)
                except RuntimeError:
                    break

    async def in_game_helper(self, mapData1, mapData2):
        self.tasks.extend([self.loop.create_task(self.listen1()), self.loop.create_task(self.listen2())])
        await asyncio.wait(self.tasks)
        await asyncio.sleep(1)
        await self.ws1.send(generate_payload("knock-knock", {"nick": 'a'}))
        await asyncio.sleep(0.4)
        await self.ws2.send(generate_payload("knock-knock", {"nick": 'b'}))
        await asyncio.sleep(0.4)
        await self.ws1.send(json.dumps({"header": "send_map", 'data': mapData1}))
        await asyncio.sleep(0.4)
        await self.ws1.send(json.dumps({"header": "send_map", 'data': mapData2}))
        await asyncio.sleep(3)
        #for task in self.tasks:
          #  print(task)
         #   task.cancel()
        #self.loop.stop()
        return json.loads(self.msg1)

    async def kill_helper(self, coords):
        pass

    async def miss_helper(self, coords):
        pass

TestServer测试类代码 #

class TestServer:
    # main()
    def setup_class(self):
        self.loop = asyncio.new_event_loop()
        self.ws1 = None
        self.ws2 = None
        self.msg1 = None
        self.msg2 = None
        self.tasks = []

    @pytest.fixture
    def mapData1(self):
        f = open("map1.json", 'r').read()
        data = json.loads(f)
        return data

    @pytest.fixture
    def mapData2(self):
        f = open("map2.json", 'r').read()
        data = json.loads(f)
        return data

    @pytest.fixture
    def event_loop(self):
        loop = asyncio.get_event_loop()
        yield loop
        loop.close()

    @pytest.mark.asyncio
    async def test_knock(self):
        async with websockets.connect(f"ws://localhost:27000") as websocket:
            await websocket.send(generate_payload("knock-knock", {"nick": 'a'}))
            for i in range(3):
                msg = json.loads(await websocket.recv())
                await asyncio.sleep(0.5)
            assert msg['header'] == 'registered'

    @pytest.mark.asyncio
    async def test_send1(self, mapData1):
        async with websockets.connect(f"ws://localhost:27000") as websocket:
            await websocket.send(generate_payload("knock-knock", {"nick": 'a'}))
            for i in range(3):
                await websocket.recv()
                await asyncio.sleep(0.5)
            await websocket.send(json.dumps({"header": "send_map", 'data': mapData1}))
            msg = json.loads(await websocket.recv())
            assert msg['header'] == 'ready'
            await websocket.close()

    @pytest.mark.asyncio
    async def test_send2(self, mapData2):
        async with websockets.connect(f"ws://localhost:27000") as websocket:
            await websocket.send(generate_payload("knock-knock", {"nick": 'b'}))
            for i in range(4):
                await websocket.recv()
                await asyncio.sleep(0.5)
            await websocket.send(json.dumps({"header": "send_map", 'data': mapData2}))
            msg = json.loads(await websocket.recv())
            assert msg['header'] == 'ready'
            await websocket.close()

    @pytest.mark.asyncio
    async def test_game_started(self, mapData1, mapData2, event_loop):
        helper = HelpTest(event_loop)
        answer = await helper.in_game_helper(mapData1, mapData2)
        print(answer)
        assert answer['header'] == "in_game!!!"

报错栈信息 #

Traceback (most recent call last):
  File "tests.py", line 40, in listen2
    break
  File "SeaBattle\venv\lib\site-packages\websockets\legacy\client.py", line 650, in __aexit__
    await self.protocol.close()
  File "SeaBattle\venv\lib\site-packages\websockets\legacy\protocol.py", line 768, in close
    await asyncio.wait_for(
  File "C:\Users\zayyc\AppData\Local\Programs\Python\Python39\lib\asyncio\tasks.py", line 435, in wait_for
    loop = events.get_running_loop()
RuntimeError: no running event loop

解决方案 #

核心问题分析 #

1. in_game_helper中先执行await asyncio.wait(self.tasks)是逻辑错误：listen1/2是无限循环任务，这行代码会阻塞到任务完成，后续发送消息的代码根本不会执行，实际测试中是因为事件循环被关闭才导致任务中断，进而执行后续代码。
2. 手动传入并管理事件循环与pytest-asyncio的自动循环冲突，导致循环关闭后任务仍在尝试执行。
3. listen方法捕获RuntimeError而不是任务取消的asyncio.CancelledError，无法正确处理任务取消逻辑，导致websocket的async with上下文管理器在循环关闭后尝试执行关闭操作时出错。

具体修改步骤 #

1. 修正HelpTest类的任务管理逻辑

class HelpTest:
    def __init__(self):
        # 不再手动传入loop，使用当前运行的循环
        self.ws1 = None
        self.ws2 = None
        self.msg1 = None
        self.msg2 = None
        self.tasks = []

    async def listen1(self):
        try:
            async with websockets.connect(f"ws://localhost:27000") as websocket:
                self.ws1 = websocket
                while True:
                    self.msg1 = await websocket.recv()
                    await asyncio.sleep(0.1)
        except asyncio.CancelledError:
            # 捕获任务取消异常，正常退出
            pass
        except Exception as e:
            # 处理其他可能的异常
            print(f"listen1 error: {e}")

    async def listen2(self):
        try:
            async with websockets.connect(f"ws://localhost:27000") as websocket:
                self.ws2 = websocket
                while True:
                    self.msg2 = await websocket.recv()
                    await asyncio.sleep(0.1)
        except asyncio.CancelledError:
            pass
        except Exception as e:
            print(f"listen2 error: {e}")

    async def in_game_helper(self, mapData1, mapData2):
        # 使用asyncio.create_task创建任务，自动绑定当前运行的循环
        self.tasks = [asyncio.create_task(self.listen1()), asyncio.create_task(self.listen2())]
        # 先等待连接建立，再发送消息
        await asyncio.sleep(0.5)
        
        await self.ws1.send(generate_payload("knock-knock", {"nick": 'a'}))
        await asyncio.sleep(0.4)
        await self.ws2.send(generate_payload("knock-knock", {"nick": 'b'}))
        await asyncio.sleep(0.4)
        await self.ws1.send(json.dumps({"header": "send_map", 'data': mapData1}))
        await asyncio.sleep(0.4)
        await self.ws1.send(json.dumps({"header": "send_map", 'data': mapData2}))
        await asyncio.sleep(3)
        
        # 主动取消任务并等待任务结束
        for task in self.tasks:
            task.cancel()
        await asyncio.gather(*self.tasks, return_exceptions=True)
        
        return json.loads(self.msg1)

2. 修正TestServer类的事件循环管理

class TestServer:
    # 移除setup_class中手动创建的loop，不需要手动管理
    @pytest.fixture
    def mapData1(self):
        with open("map1.json", 'r') as f:
            data = json.loads(f.read())
        return data

    @pytest.fixture
    def mapData2(self):
        with open("map2.json", 'r') as f:
            data = json.loads(f.read())
        return data

    # 移除自定义的event_loop fixture，使用pytest-asyncio默认的循环管理

    # ... 其他测试方法保持不变 ...

    @pytest.mark.asyncio
    async def test_game_started(self, mapData1, mapData2):
        helper = HelpTest()  # 不再传入loop
        answer = await helper.in_game_helper(mapData1, mapData2)
        print(answer)
        assert answer['header'] == "in_game!!!"

修改说明 #

• 移除手动管理事件循环的逻辑，完全依赖pytest-asyncio提供的运行中循环，避免循环关闭后任务仍执行的问题。
• 调整in_game_helper的执行顺序：先启动监听任务，等待连接建立后再发送消息，最后主动取消任务并等待任务完成，确保websocket连接在循环运行时正常关闭。
• 修改listen方法的异常捕获逻辑，正确处理任务取消的CancelledError，让async with上下文管理器在循环运行时完成关闭操作，避免循环关闭后执行异步操作。

内容的提问来源于stack exchange，提问作者Юрий Зайцев