python-3.x - 在请求中运行后台异步任务而不在 pytest-sanic 中关闭事件循环

Question

我想在 pytest-sanic 中并行运行 2 个测试，但是当我尝试关闭事件循环时，会引发以下错误。

RuntimeError('Event loop is closed')

这是代码的简化版本的样子

import pytest
import json
from main import make_app

@pytest.yield_fixture
def app():
    app = make_app()
    yield app

@pytest.fixture
def test_cli(loop, app, test_client):

    return loop.run_until_complete(test_client(app))



async def test_upload_api(test_cli):
    data = {
    "urls" : "asd"
    }
    response = await test_cli.post('/v1/images/upload',data=json.dumps(data))

    assert response.status == 200

    response = await test_cli.get('/v1/images')

    assert response.status == 200

如何在同一个测试用例中运行多个请求，并在后台使用异步操作运行一次。

更新：我在 API 调用中运行异步任务，当我禁用它时，我没有收到错误。但是，我需要在后台执行异步任务时测试 API。

===================================================================================== 2 passed in 0.14 seconds =====================================================================================
exception calling callback for <Future at 0x7fa4cc971a90 state=finished returned list>
Traceback (most recent call last):
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 324, in _invoke_callbacks
    callback(self)
  File "/usr/lib/python3.6/asyncio/futures.py", line 419, in _call_set_state
    dest_loop.call_soon_threadsafe(_set_state, destination, source)
  File "/usr/lib/python3.6/asyncio/base_events.py", line 620, in call_soon_threadsafe
    self._check_closed()
  File "/usr/lib/python3.6/asyncio/base_events.py", line 357, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
Task was destroyed but it is pending!
task: <Task pending coro=<UrlCrawler.fetch_urls() running at /home/jibin/PycharmProjects/imgurupload/controllers/UrlCrawler.py:68> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fa4ccab81c8>()]>>
Exception ignored in: <coroutine object UrlCrawler.fetch_urls at 0x7fa4d03ff990>
Traceback (most recent call last):
  File 

"/home/jibin/PycharmProjects/imgurupload/controllers/UrlCrawler.py", line 71, in fetch_urls
    kwargs["finished"][id] = datetime.datetime.utcnow().isoformat()
  File "/usr/local/lib/python3.6/dist-packages/aiohttp/client.py", line 828, in __aexit__
    await self.close()
  File "/usr/local/lib/python3.6/dist-packages/aiohttp/client.py", line 779, in close
    self._connector.close()
  File "/usr/local/lib/python3.6/dist-packages/aiohttp/connector.py", line 657, in close
    ev.cancel()
  File "/usr/local/lib/python3.6/dist-packages/aiohttp/locks.py", line 38, in cancel
    waiter.cancel()
  File "/usr/lib/python3.6/asyncio/base_events.py", line 574, in call_soon
    self._check_closed()
  File "/usr/lib/python3.6/asyncio/base_events.py", line 357, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
Task was destroyed but it is pending!
task: <Task pending coro=<TCPConnector._resolve_host() running at /usr/local/lib/python3.6/dist-packages/aiohttp/connector.py:730> wait_for=<Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/lib/python3.6/asyncio/futures.py:408, <TaskWakeupMethWrapper object at 0x7fa4cc98e2b8>()]> cb=[shield.<locals>._done_callback() at /usr/lib/python3.6/asyncio/tasks.py:679]>

标签： python-3.xpytestsanic