abersheeran / asgi-ratelimit

A ASGI Middleware to rate limit
Apache License 2.0
292 stars 11 forks source link

`Event loop is closed` when testing with `TestClient` #72

Closed lainiwa closed 10 months ago

lainiwa commented 10 months ago

I try to follow FastAPI docs on testing, and I'm getting RuntimeError: Event loop is closed error when I have ratelimiter enabled.

In main.py:

async def _dummy_auth_function(scope: Scope) -> tuple[str, str]:
    return "dummy_uid", "default"

app.add_middleware(
    RateLimitMiddleware,
    authenticate=_dummy_auth_function,
    backend=RedisBackend(StrictRedis(
        ...
    )),
    config={
        r"^/v3/": [Rule(minute=1000, second=20)],  # <-- Works when this is deleted
    },
)

My test_ping.py file:

import pytest
from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)

@pytest.mark.integration()
def test_ping():
    response = client.get("/v3/test/ping")
    response = client.get("/v3/test/ping")  # <-- this doesn't work
    assert response.status_code == 200

This code fails with Event loop is closed on second request, unless the endpoint is not ratelimited.

abersheeran commented 10 months ago

Please use https://www.starlette.io/testclient/#asynchronous-tests

lainiwa commented 10 months ago

I did try it, but I'm still having the same error:

@pytest.mark.asyncio()
async def test_ping():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/v3/test/ping")
        response = await client.get("/v3/test/ping")  # <-- works

@pytest.mark.asyncio()
async def test_pong():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/v3/test/ping")  # <-- fails
        response = await client.get("/v3/test/ping")

Upd: it seems i managed to fix it with

@pytest.fixture(scope='session')
def event_loop():
    policy = asyncio.get_event_loop_policy()
    loop = policy.new_event_loop()
    yield loop
    loop.close()