如何为测试目的准确配置数据库。我做错了什么，我已经阅读了网站上的 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)