Fastapi example test not working

[vjousse]

Describe the bug

The Fastapi example test is not working.

To Reproduce

Go to the directory examples/fastapi and run the tests with:

pytest _tests.py

(venv) ➜  fastapi git:(develop) pytest _tests.py
===================================== test session starts =====================================
platform linux -- Python 3.10.1, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /home/vjousse/usr/src/python/tortoise-orm/examples/fastapi
plugins: xdist-2.5.0, forked-1.4.0, anyio-3.4.0, cov-3.0.0
collected 1 item

_tests.py E                                                                             [100%]

=========================================== ERRORS ============================================
_____________________________ ERROR at setup of test_create_user ______________________________

client = <starlette.testclient.TestClient object at 0x7f5d4187b700>

    @pytest.fixture(scope="module")
    def event_loop(client: TestClient) -> Generator:
>       yield client.task.get_loop()  # type: ignore
E       AttributeError: 'Future' object has no attribute 'get_loop'

_tests.py:24: AttributeError
====================================== warnings summary =======================================
_tests.py::test_create_user
_tests.py::test_create_user
  /home/vjousse/usr/src/python/tortoise-orm/tortoise/contrib/test/__init__.py:110: DeprecationWarning: There is no current event loop
    loop = loop or asyncio.get_event_loop()

-- Docs: https://docs.pytest.org/en/stable/warnings.html
=================================== short test summary info ===================================
ERROR _tests.py::test_create_user - AttributeError: 'Future' object has no attribute 'get_loop'
================================ 2 warnings, 1 error in 0.91s =================================

Steps to reproduce the behavior, preferaby a small code snippet.

Expected behavior
I would expect the test to be green.

[joshua-hashimoto]

I have the same problem.

Python: 3.9.7
FastAPI: 0.70.1
Tortoise-ORM: 0.18.0

[NielXu]

Same issue.

Python 3.8.8
fastapi 0.72.0
tortoise-orm 0.18.1

[voneiden]

You can get at least the test passing by swapping the function with

@pytest.fixture(scope="module")
def event_loop() -> Generator:
    yield asyncio.get_event_loop()

Disclaimer: I have no idea if this breaks something. But I'm guessing it's OK to not use starlette's test client event loop for the DB stuff.

[greedWizard]

That one worked for me, the tests pass, but the consequences are still unknown. Can anyone who disliked the comment explain me why this decision is bad/wrong if it works?

[TPXP]

voneiden's approach still gave me a warning with Python 3.10:

DeprecationWarning: There is no current event loop

I had a look at pytest-asyncio to get a better understanding of async handling in pytest. The main idea is to instantiate an event loop at some point and pass it to tests when needed, then wrap async logic in event_loop.run_until_complete. By default, pytest-asyncio will create an event loop for each test ("function scope"), which does not work for us since we want to initialize tortoise once for all tests (that is, with a "module scope").

The event loop is instantiated in pytest-asyncio like this: https://github.com/pytest-dev/pytest-asyncio/blob/f979af9/pytest_asyncio/plugin.py#L486

As said in their docs, we would have to write the event_loop fixture anyway: https://github.com/pytest-dev/pytest-asyncio#async-fixtures

All scopes are supported, but if you use a non-function scope you will need to redefine the event_loop fixture to have the same or broader scope. Async fixtures need the event loop, and so must have the same or narrower scope than the event_loop fixture.

All we need to do is add a scope="module" to the fixture definition and pass it to Tortoise directly so that it does not try to create or get another one:

# conftest.py
@pytest.fixture(scope="module")
def event_loop() -> Iterator[asyncio.AbstractEventLoop]:
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

# Test client
@pytest.fixture(scope="module")
def client(event_loop: asyncio.BaseEventLoop) -> Iterator[TestClient]:
    initializer(TORTOISE_ORM["apps"]["models"]["models"], loop=event_loop)
    with TestClient(app) as c:
        yield c
    finalizer()

As I was writing my tests, I found out that Tortoise expected the app to have the same configuration as tests. If you connect to a separate database by default, or don't set generate_schemas=True in register_tortoise, you may face database connection issues or missing database tables:

socket.gaierror: [Errno 8] nodename nor servname provided, or not known
tortoise.exceptions.OperationalError: no such table: user

Code tweaks

# app.py
import init_db from db

app = FastAPI()

# Your app code...

init_db(app)

# db.py
TORTOISE_ORM = {
    "connections": {"default": f"postgres://{DATABASE_USERNAME}:{DATABASE_PASSWORD}@{DATABASE_HOST}:{DATABASE_PORT}/{DATABASE_NAME}"},
    "apps": {
        "models": {
            "models": [
                # Aerich migrations
                "aerich.models",
                # Our custom models
                "models"
            ],
            "default_connection": "default",
        },
    },
}

generate_schemas = False

def switch_to_test_mode():
    global TORTOISE_ORM, generate_schemas
    TORTOISE_ORM['connections']['default'] = 'sqlite://:memory:'
    generate_schemas = True


def init(app: FastAPI):
    register_tortoise(
        app,
        TORTOISE_ORM,
        add_exception_handlers=True,
        generate_schemas=generate_schemas
    )

# tests/conftest.py

from models import TORTOISE_ORM, switch_to_test_mode
switch_to_test_mode() # Tortoise ORM hack - has to be done before importing app

import app from app
# ...

[skyanth]

Got it working with @TPXP's excellent guide and explanation, thanks! (Also thanks for the test db config, I needed that too!)

[ada0l]

@vjousse You can see my template. There is work tests. @ada0l/fastapi_tortoise_aerich_template

[ljhenne]

Went down quite the rabbit-hole with this one.

Okay. So it appears the change that breaks this example was introduced in Starlette v0.15.0.

Specifically, here's the PR that introduced the change: encode/starlette#1157.

The release notes offer the following brief explanation: "Starlette now supports Trio as an async runtime via AnyIO".

The result was an abstraction of the previous async implementation via AnyIO so that a number of async backends could be supported.

That meant that the Type of TestClient.task changed and was now a Future which does not have a get_loop method. Hence, the test fails. See here for the source code.

The good news is, the same pattern is still roughly available. Its just that the names have changed. To implement AnyIO, a BlockingPortal is initiated inside of TestClient. I am not too extremely familiar with this, but it basically seems like a new API for some kind of async processing "thing". Now, instead of calling event_loop.run_until_complete, you can call blocking_portal.call to the same effect.

The code ends up looking like this:

...
@pytest.fixture(scope="module")
def blocking_portal(client: TestClient) -> Iterator[BlockingPortal]:
    yield client.portal


def test_create_user(client: TestClient, blocking_portal: BlockingPortal):  # nosec
    response = client.post("/users", json={"username": "admin"})
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["username"] == "admin"
    assert "id" in data
    user_id = data["id"]

    async def get_user_by_db():
        user = await Users.get(id=user_id)
        return user

    user_obj = blocking_portal.call(get_user_by_db)
    assert user_obj.id == user_id

Of course at this point, I don't really see the point of a separate blocking_portal fixture (does anyone else?), so I think that can just be dropped in favor of something like this:

def test_create_user(client: TestClient):  # nosec
    response = client.post("/users", json={"username": "admin"})
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["username"] == "admin"
    assert "id" in data
    user_id = data["id"]

    async def get_user_by_db():
        user = await Users.get(id=user_id)
        return user

    user_obj = client.portal.call(get_user_by_db)
    assert user_obj.id == user_id

It seems like this is the most consistent and least hacky option available to obtain the event loop (or event-loop-like object) from the TestClient.

I'll let this stew for a few days, but I plan on submitting a PR with this change if no one can educate me further on this.