python - Fastapi 数据库测试隔离与 starlette 配置

Question

如何为测试目的准确配置数据库。我做错了什么，我已经阅读了网站上的 starlette 配置，但我认为我没有正确理解它。我尝试构建它，但收到以下错误消息:

tests\test_main.py .                                                                                                                        [ 33%]
tests\test_players\test_players.py FF                                                                                                       [100%]

==================================================================== FAILURES ====================================================================
________________________________________________________________ test_post_player ________________________________________________________________

test_app = <httpx.AsyncClient object at 0x0000021B01F8C2E0>

    @pytest.mark.asyncio
    async def test_post_player(test_app):
        test_request_payload = {"first_name": "Jane", "last_name": "Doe", "nationality": "American", "date_of_birth": "1985-04-15"}
        test_response_payload = {"first_name": "Jane", "last_name": "Doe", "nationality": "American"}


>       response = await test_app.post("/api/players/", data=json.dumps(test_request_payload),)

tests\test_players\test_players.py:12:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv\lib\site-packages\httpx\_client.py:1633: in post
    return await self.request(
venv\lib\site-packages\httpx\_client.py:1371: in request
    response = await self.send(
venv\lib\site-packages\httpx\_client.py:1406: in send
    response = await self._send_handling_auth(
venv\lib\site-packages\httpx\_client.py:1444: in _send_handling_auth
    response = await self._send_handling_redirects(
venv\lib\site-packages\httpx\_client.py:1476: in _send_handling_redirects
    response = await self._send_single_request(request, timeout)
venv\lib\site-packages\httpx\_client.py:1502: in _send_single_request
    (status_code, headers, stream, ext,) = await transport.arequest(
venv\lib\site-packages\httpx\_transports\asgi.py:148: in arequest
    await self.app(scope, receive, send)
venv\lib\site-packages\fastapi\applications.py:199: in __call__
    await super().__call__(scope, receive, send)
venv\lib\site-packages\starlette\applications.py:111: in __call__
    await self.middleware_stack(scope, receive, send)
venv\lib\site-packages\starlette\middleware\errors.py:181: in __call__
    raise exc from None
venv\lib\site-packages\starlette\middleware\errors.py:159: in __call__
    await self.app(scope, receive, _send)
venv\lib\site-packages\starlette\middleware\cors.py:78: in __call__
    await self.app(scope, receive, send)
venv\lib\site-packages\starlette\exceptions.py:82: in __call__
    raise exc from None
venv\lib\site-packages\starlette\exceptions.py:71: in __call__
    await self.app(scope, receive, sender)
venv\lib\site-packages\starlette\routing.py:566: in __call__
    await route.handle(scope, receive, send)
venv\lib\site-packages\starlette\routing.py:227: in handle
    await self.app(scope, receive, send)
venv\lib\site-packages\starlette\routing.py:41: in app
    response = await func(request)
venv\lib\site-packages\fastapi\routing.py:201: in app
    raw_response = await run_endpoint_function(
venv\lib\site-packages\fastapi\routing.py:148: in run_endpoint_function
    return await dependant.call(**values)
src\app\api\routers\players\players.py:11: in post_player
    player = await crudplayers.create_player(payload)
src\app\crud\crudplayers\crudplayers.py:13: in create_player
    await database.execute(query=query)
venv\lib\site-packages\databases\core.py:160: in execute
    async with self.connection() as connection:
venv\lib\site-packages\databases\core.py:219: in __aenter__
    await self._connection.acquire()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <databases.backends.postgres.PostgresConnection object at 0x0000021B01F980D0>

    async def acquire(self) -> None:
        assert self._connection is None, "Connection is already acquired"
>       assert self._database._pool is not None, "DatabaseBackend is not running"
E       AssertionError: DatabaseBackend is not running

venv\lib\site-packages\databases\backends\postgres.py:148: AssertionError
________________________________________________________________ test_read_player ________________________________________________________________

test_app = <httpx.AsyncClient object at 0x0000021B02374EB0>

    @pytest.mark.asyncio
    async def test_read_player(test_app):
        test_response_payload = {"id": 1, "first_name": "Jane", "last_name": "Doe", "nationality": "American"}
>       response = await test_app.get("/api/players/1")

tests\test_players\test_players.py:21:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv\lib\site-packages\httpx\_client.py:1548: in get
    return await self.request(
venv\lib\site-packages\httpx\_client.py:1371: in request
    response = await self.send(
venv\lib\site-packages\httpx\_client.py:1406: in send
    response = await self._send_handling_auth(
venv\lib\site-packages\httpx\_client.py:1444: in _send_handling_auth
    response = await self._send_handling_redirects(
venv\lib\site-packages\httpx\_client.py:1476: in _send_handling_redirects
    response = await self._send_single_request(request, timeout)
venv\lib\site-packages\httpx\_client.py:1502: in _send_single_request
    (status_code, headers, stream, ext,) = await transport.arequest(
venv\lib\site-packages\httpx\_transports\asgi.py:148: in arequest
    await self.app(scope, receive, send)
venv\lib\site-packages\fastapi\applications.py:199: in __call__
    await super().__call__(scope, receive, send)
venv\lib\site-packages\starlette\applications.py:111: in __call__
    await self.middleware_stack(scope, receive, send)
venv\lib\site-packages\starlette\middleware\errors.py:181: in __call__
    raise exc from None
venv\lib\site-packages\starlette\middleware\errors.py:159: in __call__
    await self.app(scope, receive, _send)
venv\lib\site-packages\starlette\middleware\cors.py:78: in __call__
    await self.app(scope, receive, send)
venv\lib\site-packages\starlette\exceptions.py:82: in __call__
    raise exc from None
venv\lib\site-packages\starlette\exceptions.py:71: in __call__
    await self.app(scope, receive, sender)
venv\lib\site-packages\starlette\routing.py:566: in __call__
    await route.handle(scope, receive, send)
venv\lib\site-packages\starlette\routing.py:227: in handle
    await self.app(scope, receive, send)
venv\lib\site-packages\starlette\routing.py:41: in app
    response = await func(request)
venv\lib\site-packages\fastapi\routing.py:201: in app
    raw_response = await run_endpoint_function(
venv\lib\site-packages\fastapi\routing.py:148: in run_endpoint_function
    return await dependant.call(**values)
src\app\api\routers\players\players.py:18: in read_player
    player = await crudplayers.get_player(player_id)
src\app\crud\crudplayers\crudplayers.py:19: in get_player
    player = await database.fetch_one(query=query)
venv\lib\site-packages\databases\core.py:145: in fetch_one
    async with self.connection() as connection:
venv\lib\site-packages\databases\core.py:219: in __aenter__
    await self._connection.acquire()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <databases.backends.postgres.PostgresConnection object at 0x0000021B023757F0>

    async def acquire(self) -> None:
        assert self._connection is None, "Connection is already acquired"
>       assert self._database._pool is not None, "DatabaseBackend is not running"
E       AssertionError: DatabaseBackend is not running

venv\lib\site-packages\databases\backends\postgres.py:148: AssertionError
================================================================ warnings summary ================================================================
venv\lib\site-packages\aiofiles\os.py:10
venv\lib\site-packages\aiofiles\os.py:10
venv\lib\site-packages\aiofiles\os.py:10
venv\lib\site-packages\aiofiles\os.py:10
venv\lib\site-packages\aiofiles\os.py:10
  c:\users\mathiaskolie\pycharmprojects\microservices\livegolf\venv\lib\site-packages\aiofiles\os.py:10: DeprecationWarning: "@coroutine" decorator
 is deprecated since Python 3.8, use "async def" instead
    def run(*args, loop=None, executor=None, **kwargs):

-- Docs: https://docs.pytest.org/en/stable/warnings.html
============================================================ short test summary info =============================================================
FAILED tests/test_players/test_players.py::test_post_player - AssertionError: DatabaseBackend is not running
FAILED tests/test_players/test_players.py::test_read_player - AssertionError: DatabaseBackend is not running
==================================================== 2 failed, 1 passed, 5 warnings in 7.23s =====================================================

conftest.py 中的 Starlette 配置

import pytest
from httpx import AsyncClient
from starlette.config import environ
from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database, drop_database
from src.app.domain.models.models import metadata

environ['TESTING'] = 'True'

from src.app.main import app
from src.app.core.config import TEST_DATABASE_URL


@pytest.fixture(scope="function", autouse=True)
async def create_test_database():
    url = str(TEST_DATABASE_URL)
    engine = create_engine(url)
    create_database(url)
    metadata.create_all(engine)
    yield                    **#it returns None**
    drop_database(url) 


@pytest.fixture(scope="function")
async def test_app():
    async with AsyncClient(app=app, base_url="http://localhost:8000", headers={"Content-Type": "application/json"},) as ac:
        yield ac

数据库配置

from starlette.datastructures import Secret, CommaSeparatedStrings
from starlette.config import Config
from pydantic import BaseSettings
from typing import List
from databases import DatabaseURL
from sqlalchemy import create_engine



class Settings(BaseSettings):
    POSTGRES_USER: str
    POSTGRES_PASSWORD: str
    POSTGRES_HOST:str
    POSTGRES_PORT: int
    DATABASE_NAME: str
    ALLOWED_HOSTS: List[str]
    TESTING: str
    TESTING_DATABASE_NAME: str

    class Config:
        env_file = "src/app/core/.env"


config = Config(Settings.Config.env_file)

SECRET_KEY = config('SECRET_KEY', cast=Secret)

TESTING = config('TESTING', cast=bool, default=False)
POSTGRES_USER = config('POSTGRES_USER')
POSTGRES_PASSWORD = config('POSTGRES_PASSWORD', cast=Secret)
POSTGRES_HOST = config('POSTGRES_HOST')
POSTGRES_PORT = config('POSTGRES_PORT', cast=int, default=5432)
DATABASE_NAME = config('DATABASE_NAME')
ALLOWED_HOSTS = config('ALLOWED_HOSTS', cast=CommaSeparatedStrings)
TESTING_DATABASE_NAME = config('TESTING_DATABASE_NAME')


DATABASE_URL = config(
    "DATABASE_URL",
    default=f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{DATABASE_NAME}",
)

TEST_DATABASE_URL = config(
    "TEST_DATABASE_URL",
    default=f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{TESTING_DATABASE_NAME}",
)

engine = create_engine(DATABASE_URL, pool_size=3, max_overflow=0)

Answer 1

我找到了这个解决方案 here :
“在上下文管理器中使用您的 TestClient”

@pytest.fixture(scope="module")
def client() -> Generator:
    with TestClient(api) as c:
        yield c

好作品对我有同样的麻烦