FastAPI에서 테스트 간에 데이터베이스를 설정하고 해체하는 방법은 무엇입니까?

FastAPI에서 테스트 간에 데이터베이스를 설정하고 해체하는 방법은 무엇입니까?

나는 FastAPI에 따라 유닛 테스트를 설정했지만, 테스트 중에 데이터베이스가 유지되는 경우에만 적용됩니다.

테스트당 데이터베이스를 빌드하고 해체하려면 어떻게 해야 합니까? (예를 들어, 첫 번째 테스트 후 데이터베이스가 더 이상 비어 있지 않기 때문에 아래 두 번째 테스트는 실패합니다.).

저는 현재 각 테스트의 시작과 끝에 전화를 걸어 (아래 코드로 의견을 제시함) 이 작업을 수행하고 있지만, 이것은 분명히 이상적이지 않습니다(테스트가 실패할 경우 데이터베이스가 절대 해체되지 않아 다음 테스트 결과에 영향을 미칩니다).

어떻게 하면 제대로 할 수 있을까요? 의존성을 중심으로 일종의 파이테스트 고정장치를 만들어야 할까요?

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from main import app, get_db
from database import Base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base.metadata.create_all(bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_get_todos():
    # Base.metadata.create_all(bind=engine)

    # create
    response = client.post('/todos/', json={'text': 'some new todo'})
    data1 = response.json()
    response = client.post('/todos/', json={'text': 'some even newer todo'})
    data2 = response.json()

    assert data1['user_id'] == data2['user_id']

    response = client.get('/todos/')

    assert response.status_code == 200
    assert response.json() == [
        {'id': data1['id'], 'user_id': data1['user_id'], 'text': data1['text']},
        {'id': data2['id'], 'user_id': data2['user_id'], 'text': data2['text']}
    ]

    # Base.metadata.drop_all(bind=engine)

def test_get_empty_todos_list():
    # Base.metadata.create_all(bind=engine)

    response = client.get('/todos/')

    assert response.status_code == 200
    assert response.json() == []

    # Base.metadata.drop_all(bind=engine)

---

테스트에 실패한 경우에도 테스트 후 정리(및 테스트 전 설정)를 위해 pytest는 을 제공합니다.

이 경우 각 검정 전에 모든 표를 만들고 나중에 다시 삭제할 수 있습니다. 이는 다음과 같은 고정 장치를 사용하여 달성할 수 있습니다:

@pytest.fixture()
def test_db():
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)

그런 다음 테스트에 다음과 같이 사용합니다:

def test_get_empty_todos_list(test_db):
    response = client.get('/todos/')

    assert response.status_code == 200
    assert response.json() == []

인수 목록에 pytest가 있는 각 테스트에 대해 먼저 실행된 다음 테스트 코드를 반환하고 테스트가 실패한 경우에도 해당 테스트가 실행되는지 확인합니다.

전체 코드:

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from main import app, get_db
from database import Base


SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()


@pytest.fixture()
def test_db():
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)


def test_get_todos(test_db):
    response = client.post("/todos/", json={"text": "some new todo"})
    data1 = response.json()
    response = client.post("/todos/", json={"text": "some even newer todo"})
    data2 = response.json()

    assert data1["user_id"] == data2["user_id"]

    response = client.get("/todos/")

    assert response.status_code == 200
    assert response.json() == [
        {"id": data1["id"], "user_id": data1["user_id"], "text": data1["text"]},
        {"id": data2["id"], "user_id": data2["user_id"], "text": data2["text"]},
    ]


def test_get_empty_todos_list(test_db):
    response = client.get("/todos/")

    assert response.status_code == 200
    assert response.json() == []

---

응용프로그램이 증가함에 따라 각 테스트에 대한 전체 데이터베이스 설정 및 해체 속도가 느려질 수 있습니다.

이에 대한 해결책은 DB를 한 번만 설정한 다음 실제로는 아무것도 커밋하지 않는 것입니다. 중첩된 트랜잭션 및 롤백을 사용하여 이 작업을 수행할 수 있습니다:

import pytest
import sqlalchemy as sa
from fastapi.testclient import TestClient
from sqlalchemy.orm import sessionmaker

from database import Base
from main import app, get_db

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = sa.create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Set up the database once
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)


# These two event listeners are only needed for sqlite for proper
# SAVEPOINT / nested transaction support. Other databases like postgres
# don't need them. 
# From: https://docs.sqlalchemy.org/en/14/dialects/sqlite.html#serializable-isolation-savepoints-transactional-ddl
@sa.event.listens_for(engine, "connect")
def do_connect(dbapi_connection, connection_record):
    # disable pysqlite's emitting of the BEGIN statement entirely.
    # also stops it from emitting COMMIT before any DDL.
    dbapi_connection.isolation_level = None


@sa.event.listens_for(engine, "begin")
def do_begin(conn):
    # emit our own BEGIN
    conn.exec_driver_sql("BEGIN")


# This fixture is the main difference to before. It creates a nested
# transaction, recreates it when the application code calls session.commit
# and rolls it back at the end.
# Based on: https://docs.sqlalchemy.org/en/14/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites
@pytest.fixture()
def session():
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)

    # Begin a nested transaction (using SAVEPOINT).
    nested = connection.begin_nested()

    # If the application code calls session.commit, it will end the nested
    # transaction. Need to start a new one when that happens.
    @sa.event.listens_for(session, "after_transaction_end")
    def end_savepoint(session, transaction):
        nonlocal nested
        if not nested.is_active:
            nested = connection.begin_nested()

    yield session

    # Rollback the overall transaction, restoring the state before the test ran.
    session.close()
    transaction.rollback()
    connection.close()


# A fixture for the fastapi test client which depends on the
# previous session fixture. Instead of creating a new session in the
# dependency override as before, it uses the one provided by the
# session fixture.
@pytest.fixture()
def client(session):
    def override_get_db():
        yield session

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    del app.dependency_overrides[get_db]


def test_get_empty_todos_list(client):
    response = client.get("/todos/")

    assert response.status_code == 200
    assert response.json() == []

여기에 두 개의 고정 장치( 및 )가 있으면 다음과 같은 추가적인 이점이 있습니다:

테스트가 API로만 대화하는 경우, 명시적으로 DB 픽스처를 추가하는 것을 기억할 필요가 없습니다(그러나 여전히 암묵적으로 호출됩니다). 그리고 DB와 직접 대화하는 테스트를 작성하려면 다음과 같이 할 수 있습니다:

def test_something(session):
    session.query(...)

예를 들어 API 호출 전에 db 상태를 준비하려면 다음과 같이 하십시오:

def test_something_else(client, session):
    session.add(...)
    session.commit()
    client.get(...)

애플리케이션 코드와 테스트 코드 모두 DB의 동일한 상태를 확인합니다.

---

각 테스트 실행 후 테이블을 잘라낼 수도 있습니다. 이렇게 하면 스키마를 실제로 제거하지 않고 모든 데이터가 지워지므로 Base.metadata.drop_all(bind=engine)을 수행하는 것만큼 느리지 않습니다:

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from contextlib import contextmanager

engine = create_engine('postgresql://...')
Session = sessionmaker(bind=engine)
Base = declarative_base()


@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()


def clear_tables():
    with session_scope() as conn:
        for table in Base.metadata.sorted_tables:
            conn.execute(
                f"TRUNCATE {table.name} RESTART IDENTITY CASCADE;"
            )
        conn.commit()


@pytest.fixture
def test_db_session():
    yield engine
    engine.dispose()
    clear_tables()


def test_some_feature(test_db_session):
    test_db_session.query(...)
    (...)

---

다음은 데이터베이스 설정 및 해체를 포함한 전체 FastAPI 테스트 환경을 위한 솔루션입니다. 이미 받아들여진 답변이 있음에도 불구하고, 저는 제 생각을 기여하고 싶습니다.

테스트 환경을 구성할 때 이러한 고정 장치를 conftest.py 파일에 포함할 수 있습니다. 그 안에 정의된 고정 장치는 테스트 패키지에 포함된 모든 테스트에 자동으로 액세스할 수 있습니다.

a) 우선, 수입을 하세요.

당신의 수입 경로가 나의 수입 경로와 다를 수 있으니 그것도 다시 확인하세요.

import pytest
from fastapi.testclient import TestClient

# Import the SQLAlchemy parts
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from app.main import app
from app.database import get_db,Base

# Create the new database session

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

그런 다음 Pytest 고정 장치를 사용합니다. 이 고정 장치는 각 테스트 기능이 적용되기 전에 실행되는 기능입니다.

b). 세션 픽스처

@pytest.fixture()
def session():

    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

    db = TestingSessionLocal()

    try:
        yield db
    finally:
        db.close()

위의 세션 픽스처를 사용하면 테스트가 실행될 때마다 테스트 데이터베이스에 연결하여 테이블을 만든 다음 테스트가 완료되면 테이블을 삭제할 수 있습니다.

c) 고객 고정 장치

@pytest.fixture()
def client(session):

    # Dependency override

    def override_get_db():
        try:

            yield session
        finally:
            session.close()

    app.dependency_overrides[get_db] = override_get_db

    yield TestClient(app)

위의 픽스처는 우리를 새로운 테스트 데이터베이스에 연결하고 메인 앱에 의해 만들어진 초기 데이터베이스 연결을 재정의한다. 이 클라이언트 고정장치가 작동하려면 세션 고정장치가 필요합니다.

그런 다음 아래와 같이 아무것도 가져올 필요 없이 그림과 같이 고정 장치를 사용할 수 있습니다.

def test_index(client):
    res = client.get("/")
    assert res.status_code == 200

이제 전체 conftest.py 파일은 다음과 같습니다:

import pytest
from fastapi.testclient import TestClient

# Import the SQLAlchemy parts
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from app.main import app
from app.database import get_db, Base

# Create the new database session

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


@pytest.fixture()
def session():

    # Create the database

    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()


@pytest.fixture()
def client(session):

    # Dependency override

    def override_get_db():
        try:
            yield session
        finally:
            session.close()

    app.dependency_overrides[get_db] = override_get_db

    yield TestClient(app)

---

테스트 폴더에 파일을 만듭니다. 이 파일에서 테스트 데이터베이스 설정을 유지하고 테스트 API에서 사용할 고정 장치를 만들 것입니다. 나는 postgres 데이터베이스를 사용하고 있다.

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.db.database import Base, get_db
import pytest

TEST_DATABASE_URL = 'postgresql://postgres:admin@localhost:5432/yourdatabsename'
main_app = app    # this app comes from my main file, in which i declared FAST API as app

def start_application():   # start application
  return main_app

SQLALCHEMY_DATABASE_URL = TEST_DATABASE_URL
engine = create_engine(SQLALCHEMY_DATABASE_URL)  # create engine

SessionTesting = sessionmaker(autocommit=False, autoflush=False, bind=engine)  # now we create test-session 


@pytest.fixture(scope="function")
def app():
   """
   Create a fresh database on each test case.
   """
   Base.metadata.create_all(engine)  # Create the tables.
   _app = start_application()
   yield _app
   Base.metadata.drop_all(engine)    # drop that tables


@pytest.fixture(scope="function")
def db_session(app: FastAPI):
   connection = engine.connect()
   transaction = connection.begin()
   session = SessionTesting(bind=connection)
   yield session  # use the session in tests.
   session.close()
   transaction.rollback()
   connection.close()

@pytest.fixture(scope="function")
def client(app: FastAPI, db_session: SessionTesting):
   """
   Create a new FastAPI TestClient that uses the `db_session` fixture to override the `get_db` dependency that is injected into routes.
   """
   
   def _get_test_db():
      db_session = SessionTesting()
      try:
         yield db_session
      finally:
         db_session.close()  

   app.dependency_overrides[get_db] = _get_test_db
   with TestClient(app) as client:
     yield client      

이제 모든 테스트 파일에 사용됩니다

예를들면

나의 조직 유형 API에서 나는 다음과 같은 이름의 파일을 만든다

from app.models.models import OrganizationType


def test_get_organization_type(client, db_session):
    response_post = client.post(URL will comes Here, json={'type_name': 'test_organization'})
    assert response_post.status_code == 201

    response_get = client.get(URL will come Here)  # get_request
    data = response_get.json()
    assert response_get.status_code == 200

: 로컬 데이터베이스와 테스트 데이터베이스는 분리되지만 테스트 데이터베이스에서는 개체가 요청 시 생성되며 테스트 후 자동으로 제거됩니다.