我认为下面的例子是一个非常常见的用例：
改变@pytest.fixture(scope="module")原因的范围ScopeMismatch: You tried to access the 'function' scoped fixture 'event_loop' with a 'module' scoped request object, involved factories。
此外，test_insert和test_find协程不需要 event_loop 参数，因为可以通过传递连接来访问循环。
任何想法如何解决这两个问题？
import pytest
@pytest.fixture(scope="function") # <-- want this to be scope="module"; run once!
@pytest.mark.asyncio
async def connection(event_loop):
""" Expensive function; want to do in the module scope. Only this function needs `event_loop`!
"""
conn await = make_connection(event_loop)
return conn
@pytest.mark.dependency()
@pytest.mark.asyncio
async def test_insert(connection, event_loop): …我使用 dbus 进行 IPC。为了在程序的整个生命周期中只有一辆总线，我在这里使用单例。为了演示，我连接到 NetworkManager，但可以交换。此外，我正在asyncio整个项目中使用。这是该模块的一个简约工作示例，它将突出显示下面描述的问题：
import asyncio # noqa
from dbus_next.aio import MessageBus
from dbus_next import BusType
BUS = None
async def get_bus():
# Returns a BUS singleton
global BUS
if not BUS:
BUS = await MessageBus(bus_type=BusType(2)).connect()
return BUS
async def introspect():
# Get the dbus singleton and call a method on that singleton
bus = await get_bus()
return await bus.introspect(
'org.freedesktop.NetworkManager',
'/org/freedesktop/NetworkManager',
)
我正在使用pytest该pytest-asyncio插件进行测试，除了这种情况之外，它的工作原理就像魅力一样。这是一个简约的工作测试模块：
import pytest
from example import introspect
@pytest.mark.asyncio
async def …我有两个单元测试，如果我一一运行它们，它们就会通过。如果我在类级别运行它们，一个通过，另一个失败response = await ac.post(并显示错误消息：RuntimeError: Event loop is closed
@pytest.mark.asyncio
async def test_successful_register_saves_expiry_to_seven_days(self):
async with AsyncClient(app=app, base_url="http://127.0.0.1") as ac:
response = await ac.post(
"/register/",
headers={},
json={
"device_id": "u1",
"device_type": DeviceType.IPHONE.value,
},
)
query = device.select(whereclause=device.c.id == "u1")
d = await db.fetch_one(query)
assert d.expires_at == datetime.utcnow().replace(
second=0, microsecond=0
) + timedelta(days=7)
@pytest.mark.asyncio
async def test_successful_register_saves_device_type(self):
async with AsyncClient(app=app, base_url="http://127.0.0.1") as ac:
response = await ac.post(
"/register/",
headers={},
json={
"device_id": "u1",
"device_type": DeviceType.ANDROID.value,
},
)
query = device.select(whereclause=device.c.id == "u1") …我收到错误
运行时错误：事件循环已关闭
每次我尝试在测试中进行多个异步调用时。我已经尝试使用其他 Stack Overflow 帖子中的所有其他建议来重写固定event_loop装置，但没有任何效果。我想知道我错过了什么？
运行测试命令：
python -m pytest tests/ --asyncio-mode=auto
要求.txt
pytest==7.1.2
pytest-asyncio==0.18.3
pytest-html==3.1.1
pytest-metadata==2.0.1
测试.py
async def test_user(test_client_fast_api):
assert 200 == 200
# works fine
request_first = test_client_fast_api.post("/first_route")
# recieve RuntimeError: Event loop is closed
request_second = test_client_fast_api.post("/second_route")
测试.py
@pytest.fixture()
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
我一直在尝试学习 asyncio，但找不到任何创建可用于测试服务器代码的 pytest 夹具的示例。一旦服务器启动，我猜它会阻止其他一切，因此测试永远不会运行。pytest-asyncio 是否有办法在单独的线程中运行固定装置或其他东西？还是需要自己写线程代码？或者，还有更好的方法？下面是我一直在弄乱的一些代码。它是使用带有 pytest 固定装置的流文档从官方 TCP echo 服务器直接复制和粘贴并在最后进行测试：
import asyncio
import pytest
async def handle_echo(reader, writer):
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print(f"Received {message!r} from {addr!r}")
print(f"Send: {message!r}")
writer.write(data)
await writer.drain()
print("Close the connection")
writer.close()
async def main():
server = await asyncio.start_server(
handle_echo, '127.0.0.1', 8888)
addr = server.sockets[0].getsockname()
print(f'Serving on {addr}')
async with server:
await server.serve_forever()
@pytest.fixture(scope="session")
async def server():
return await main()
@pytest.mark.asyncio
def test_something(server):
assert False
我有一个简单的回显服务器功能。如果我通过一个测试启动 pytest，一切都会正常，但如果我通过两个测试启动它，那么第二个测试就会挂起，等待服务器启动，我不明白为什么。这是完整的代码。
服务器：
async def handle_echo(reader, writer):
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print(f"SERVER: Received {message!r} from {addr!r}")
writer.write(data)
await writer.drain()
print(f"SERVER: Sent: {message!r}")
writer.close()
print("SERVER: Closed the connection")
测试设置：
HOST = "localhost"
@pytest.fixture()
def event_loop():
return asyncio.get_event_loop()
async def _async_wait_for_server(event_loop, addr, port):
while True:
a_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
await event_loop.sock_connect(a_socket, (addr, port))
return
except ConnectionRefusedError:
await asyncio.sleep(0.001)
finally:
a_socket.close()
@pytest.fixture()
def server(event_loop):
unused_tcp_port = 65432
cancel_handle = asyncio.ensure_future(main(unused_tcp_port), loop=event_loop)
event_loop.run_until_complete(asyncio.wait_for(
_async_wait_for_server(event_loop, HOST, unused_tcp_port), 5.0)) …我有一些像这样的数据库操作AsyncIOMotorClient：
async def get_user_data(token: str, client = MongoClient()):
""" Get user data from database for given token """
return await client.db.user_data.find_one({"token": token})
还有一些像这样的基本测试：
def test_read_user_data(auth_token):
response = client.post("/crud/read",
json={"token":"xxx"},
headers=HEADERS(auth_token)
)
print(response.json())
assert response.status_code == 200
一组中的第一个测试运行良好，但第二个测试在其中一个数据库操作上总是失败：
self = <_UnixSelectorEventLoop running=False closed=True debug=False>
def _check_closed(self):
if self._closed:
> raise RuntimeError('Event loop is closed')
E RuntimeError: Event loop is closed
/usr/lib/python3.8/asyncio/base_events.py:508: RuntimeError
================================================== short test summary info ==================================================
FAILED test_routes.py::test_read_user_data - RuntimeError: Event loop …我在做什么
我正在通过构建一个 REST api 来学习 aiohttp，我正在使用 Pytest（及其 async 和 aiohttp 插件）进行测试。
对于我的第一次测试（我从一开始就使用 TDD）我有以下代码：
@pytest.mark.asyncio
async def test_handle_user_create(
aiohttp_client, init_test_app, create_test_user_table
):
payload = {
"email": "tintin@gmail.com",
"username": "Tintin",
"password": "y0u != n00b1e",
}
client = await aiohttp_client(init_test_app)
resp = await client.post("/users/", json=payload)
...
aiohttp_client 是客户端装置来自 pytest-aiohttpinit_test_app 是一个固定装置，它基本上反映了我将要构建的应用程序create_test_user_table 是我在测试数据库中为用户创建表的夹具它有什么问题
我的第一个测试是在上面代码块的最后一行抛出以下运行时错误：
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ …我一直在用来unittest.IsolatedAsyncioTestCase测试我的异步方法。我一直在使用setUpClass,asyncSetUp来创建固定装置并asyncTearDown进行清理。到目前为止一切进展顺利:-)
但现在我有一个新的要求，即每个测试类异步创建一些固定装置，并在整个测试方法中使用它。
我知道setUpClass每个测试类运行一次，setUp每个测试方法运行一次。
asyncSetUp是异步等价于setUp. 但我似乎没有找到异步等价物setUpClass。
那么，每次测试异步创建和清理夹具的最佳方法是什么？
我在https://docs.python.org/3/library/unittest.html#unittest.TestCase.setUpClass尝试了官方的unittest文档，但它只记录了有关asyncSetUp.
我使用的是 Python 3.10 并使用 pytest。
我正在测试我的消费者，它使用scope['url_route']但使用HttpCommunicatoror ApplicationCommunicator，该参数未设置。我该如何设置这个参数？测试文档非常有限，并且没有关于此https://channels.readthedocs.io/en/latest/topics/testing.html的任何文档。
测试.py
from channels.testing import HttpCommunicator
import pytest
from my_app import consumers
@pytest.mark.asyncio
async def test_consumers():
communicator = HttpCommunicator(consumers.MyConsumer, 'GET', '/project/1')
response = await communicator.get_response()
assert response['status'] == 400
my_app.py
import json
from channels.db import database_sync_to_async
from channels.generic.http import AsyncHttpConsumer
from projects import model
class MyConsumer(AsyncHttpConsumer):
async def handle(self, body):
_id = int(self.scope['url_route']['kwargs'].get('project_id', 1))
record = await database_sync_to_async(models.Project.objects.get)(id=_id)
data = json.dumps({"id": _id})
await self.send_response(200, data.encode('utf8'))
我不明白装饰器@pytest.mark.asyncio可以用于哪些目的。
我试图运行下面的代码片段与pytest和pytest-asyncio插件安装失败了，所以我订立无装饰是pytest收集测试协同程序。为什么会这样存在？
async def test_div():
return 1 / 0
在为我的聊天消费者编写测试的过程中，我遇到了无法使用 WebSocketCommunicator 在测试中进行身份验证的问题。我有自定义 JwtTokenAuthMiddleware，它通过在请求查询中使用令牌来实现套接字中的身份验证，因为据我所知，使用授权标头进行适当的身份验证尚不可能。你们能给我建议或者向我提供我在网上找不到的示例代码吗？顺便说一句，我的聊天工作没有问题。另外测试应该完全没问题，我从官方文档 Django Channels 2.x 测试中获取了指南。
--JwtTokenAuthMiddlewate--
class JwtTokenAuthMiddleware:
def __init__(self, inner):
self.inner = inner
def __call__(self, scope):
close_old_connections()
try:
raw_token = scope['query_string'].decode().split('=')[1]
auth = JWTAuthentication()
validated_token = auth.get_validated_token(raw_token)
user = auth.get_user(validated_token)
scope['user'] = user
except (IndexError, InvalidToken, AuthenticationFailed):
scope['user'] = AnonymousUser()
return self.inner(scope)
JwtTokenAuthMiddlewareStack = lambda inner: JwtTokenAuthMiddleware(AuthMiddlewareStack(inner))
--测试示例--
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_trainer_auth_success():
room = await database_sync_to_async(RoomFactory.create)()
trainer = room.trainer
trainer_token = await sync_to_async(get_token_for_user)(trainer.user)
room_url = f'ws/room/{room.id}/'
trainer_communicator = WebsocketCommunicator(application, f'{room_url}?t={trainer_token}')
connected, _ = await trainer_communicator.connect() …django python-3.x pytest-django django-channels pytest-asyncio
pytest-asyncio ×12
pytest ×8
python ×5
python-3.x ×5
fastapi ×2
aiohttp ×1
django ×1
python-3.9 ×1