FastAPI와 SQLAlchemy 환경에서 Database Test

FIRST 원칙

단위 테스트를 수행하는 데 있어 여러 가지 가이드가 있지만 일반적으로 적용하고 있는 FIRST 원칙은 다음과 같습니다.

Fast
Independent
Repeatable
Self-Validating
Timely

하지만 22년 6월 기준 FastAPI와 SQLAlchemy에서 제공하는 공식문서상의 튜토리얼을 따라가면 2번째 항목인 Independent를 만족하지 못하는 현상이 발생합니다. 이는 매우 안타까운 상황이며, 많은 기술 블로그를 보아도 위와 같은 상황으로 고민하는 개발자들을 만나볼 수 있죠.
이 포스팅은 이런 문제에 대해 다른 Unit-Test lib에서와 같이 독립성을 보장하는 데이터베이스 환경을 구성하는 방향으로 잡아보았습니다.

Database Engine

FastAPI에서 제공하는 공식 문서대로 테스트 환경에서 Database를 이용해야 할 때 sqlite:///:memory:와 같이 SQLite를 이용하라고 권고하고 있습니다.
이와 같은 방식은 대부분 문제가 발생하지 않지만, 실제 서비스가 sqlite로 구성되어있지 않다면 몇몇 field 등에서는 실제로 구성된 스키마와 다른 문제 때문에 완벽한 테스트를 보장할 수 없다는 문제가 있습니다. Test가 성공적으로 수행되고 난 후의 코드는 실제상황에서 안정적인 동작을 보장해야 하는데 MySQL을 사용하는 우리 서비스에서 SQLite를 이용한 Database 테스트를 수행할 수는 없었습니다.
또한 실제 환경과 동일한 환경을 구성하기 위한다는 명분으로 아무리 개발 서버라고 하더라도 Test를 그곳에서 실행할 수는 없다는 판단과 Test의 Performance를 위해 로컬 환경에 MySQL 서버를 생성하는 쪽으로 방향을 잡고 Docker를 이용해 MySQL 서버를 띄워보았습니다.

docker run --platform linux/amd64 -p 3306:3306 --name test_db -e MYSQL_ROOT_PASSWORD=password -e MYSQL_DATABASE=test_db -e MYSQL_PASSWORD=password --restart=unless-stopped -d mysql

Test 환경변수 설정

Test가 동작하는 환경이라고 하더라도 기본적으로 실제 환경과 거의 동일한 환경에서 작동되어야 합니다.
하지만 Test 환경에서만 적용되어야 하는 환경변수에 대한 Mocking이 필요한데 pytest에서 제공해주는 pytest.ini를 통해 다음과 같이 Database의 환경변수 위주로 재구성하였습니다. 위 Section에서도 설명했듯이 실제 Database에서 테스트를 진행할 수는 없으니까 말이죠.

[pytest]
env =
    APP_ENV=test
    DATABASE_HOST=localhost
    DATABASE_USER=root
    DATABASE_PASSWORD=password
    DATABASE_NAME=test_db

Database Engine 생성

아이메디신 에서는 Database 접근에 대해서 SQLAlchemy라는 객체 형식으로 관리하고 있습니다. SQLAlchemy는 Python 환경에서 ORM을 제공해주는 라이브러리로, 상당히 많은 Python 기반의 프로젝트에서 사용하고 있을 정도로 범용성이 높다고 볼 수 있습니다. FastAPI의 공식 문서에서도 SQLAlchemy를 다루고 있기도 하죠.

class Settings(BaseSettings):

    ...

    @property
    def database_url(self):
        return f'mysql+pymysql://{self.DATABASE_USER}:{self.DATABASE_PASSWORD}@' \
               f'{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}?charset=utf8mb4'

...

class SQLAlchemy:
    @classmethod
    def create_engine(cls):
        setting_dict = dict(
            DATABASE_URL=settings.database_url,
            DATABASE_POOL_RECYCLE=settings.DATABASE_POOL_RECYCLE,
            DATABASE_ECHO=settings.DATABASE_ECHO,
        )
        return SQLAlchemy(**setting_dict)

    ...

...

database = SQLAlchemy.create_engine()

Settings 객체를 보면 database_url을 조합, 리턴 해주는 property에서 DATABASE_HOST, DATABASE_USER… 등의 환경변수가 있는데, 만약 구동되는 환경이 Test 환경이라면 앞서 pytest.ini에서 설정해준 환경변수가 적용되어 local.database를 바라보게 될 것입니다.

conftest.py

pytest에서는 테스트 환경에 필요한 fixture, plugin, module 등을 관리하기 위해 conftest.py라는 파일을 사용할 것을 권하고 있습니다. 이는 단위테스트가 동작할 때 필요한 given을 미리 정의 해놓은 파일로, 일종의 setUp 함수와 비슷하지만 동일하지는 않죠. 또한 모든 단위테스트에서 단 하나의 session을 호출해야 테스트가 종료되었을 때 후속 작업을 일괄되게 처리할 수 있다는 점과 한 번의 단위테스트 내에서 몇 단계에 걸쳐 fixture를 호출해도 멱등성을 보장한다는 특징은 conftest를 사용해야 하는 이유이기도 하죠.

이제 conftest에 Test 용 Database 세션을 관리하도록 해보겠습니다.

from app.app_core.databases.conn import database


@fixture(scope='session', autouse=True)
def db():
    from sqlalchemy_utils import create_database, database_exists, drop_database

    engine = database.engine
    _db = {
        'engine': engine,
        'session': sessionmaker(autocommit=False, autoflush=False, bind=engine),
    }

    if database_exists(engine.url):
        drop_database(engine.url)

    create_database(engine.url)

    try:
        from alembic.config import Config as AlembicConfig
        from alembic.command import upgrade as alembic_upgrade, downgrade as alembic_downgrade
        from app.app_core.settings import settings

        alembic_config = AlembicConfig()
        alembic_config.set_main_option('sqlalchemy.url', settings.database_url)
        alembic_config.set_main_option('script_location', 'revision')
        alembic_upgrade(alembic_config, 'head')

        yield _db
    finally:
        engine.dispose()

fixture에 제공되는 매개변수 중 scope='session'을 통해 해당 fixture가 테스트가 유지되는 동안 같은 인스턴스를 반환하도록 설정하였습니다. 이는 테스트 건당 데이터베이스를 생성하는 건 테스트의 성능 저하와도 밀접한 연관이 있기 때문입니다. 또한 autouse=True 옵션을 주어 명시적으로 db라는 fixture를 호출하지 않아도 데이터베이스가 구성될 수 있도록 처리하였습니다.
이제 db 인스턴스가 필요한 어느 곳에서든 def test_some(db):와 같은 방식으로 database에 접근할 수 있습니다.

하지만 대부분의 코드에서는 database 객체에 직접 접근 하는 경우보단 하나의 session에 접근하는 게 일반적입니다. 따라서 session 객체 역시 fixture로 관리해야 할 필요성을 느낍니다.

@fixture
def engine(db):
    engine = db['engine']
    connection = engine.connect()
    transaction = connection.begin()
    yield engine
    transaction.rollback()


@fixture
def session(db) -> Session:
    session = db['session']()
    session.begin_nested()
    yield session
    session.rollback()
    session.close()

현재까지 아이메디신의 FastAPI 프로젝트는 engine을 직접 제어하지는 않지만, 추후 필요할지도 모른다는 생각에 session을 만들면서 함께 만들어 보았습니다. 단위 테스트 내에서 반드시 conftest에 정의된 session fixture를 사용할 필요는 없지만, 앞서 설명한 것과 같이 테스트가 종료된 후 다른 테스트에 영향을 주지 않기 위해서 처리하는 session.rollback() 부분이 정상적으로 동작하기 위해선 session fixture를 사용할 것을 권합니다.

# 호출할 때마다 다른 인스턴스 생성
# session.commit(), session.close() 처리가 완료되기 전에 다른 session에서 접근 시
# 값이 반영되어 있지 않은 문제가 있음
# 또한 fixture에서 사용한 rollback() 처리가 동작하지 않음
def test_some():
session = next(database.session())
...
# 호출할 때마다 같은 인스턴스 생성
# flush() 처리만 되어있어도 데이터베이스에 값이 반영된 것처럼 보임
# 가장 중요한 rollback() 처리 한 번으로 다른 테스트 코드에 영향을 주지 않음
def test_some(session):

dependency_overrides

FastAPI에서는 router 호출 시 Depends를 이용해 여러 곳에서 같은 함수를 호출하더라도 단 하나의 인스턴스를 이용할 수 있는 방식을 제공합니다. 일종의 Singleton 패턴과도 같은 방식이죠. 그리고 테스트 환경에서 이 의존성을 Mocking 할 수 있는 방법을 dependency_overrides라는 방식으로 제공해주고 있습니다.
dependency_overrides를 반드시 사용해야 하는 이유는 실제 동작 하는 코드인 router에서는 conftest의 fixture를 가져올 수 없기 때문입니다.
일반적으로 router 안에서 사용되는 database.session은 다음과 같은 방식으로 구현됩니다.

@router.post(
    '/id',
    summary='아이디 찾기',
    ...
)
async def post_id(
        session: Session = Depends(database.session),
):
    ...

만약 dependency_overrides를 정의해 놓지 않은 상황이라면 router를 호출할 때마다 다른 인스턴스의 session이 생성될 것이며, 다음과 같은 테스트 코드에서 그 동작을 신뢰성을 보장할 수 없습니다.

def test_some(client, session, admin_user):
    customized_package = CustomizePackagePrice(...)
    session.add(customized_package)
    url = PATH_PAYMENTS_V1_0 + f'/customize_price_package?packageId={customized_package.id}'
    user = gen_admin_user
    client.force_authenticate(user)
    res = client.get(url)
    ...


@router.get(
    '/customize_price_package',
    ...
)
async def get_customize_package_price(
        query: CustomizePackagePriceQuery = Depends(),
        session: Session = Depends(database.session),
):
    ...
    package_id = query.package_id
    packages = list_customize_package_price(session, package_id, user)
    # dependency_overrides설정이 되어있지 않다면 다른 인스턴스의 session으로 인해 not found 발생
    return paginate(packages)

따라서 단위 테스트 실행 시 반드시 dependency_overrides를 구현해 주어야 동작의 신뢰도를 높일 수 있으며 FastAPI에서 제공하는 공식 문서상 구현 방법은 다음과 같습니다.

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()
app.dependency_overrides[get_db] = override_get_db

하지만 이 방식대로 구현하게 되면 단위 테스트 내에서 사용하는 session과 router에서 사용되는 session이 다른 인스턴스를 사용하게 되는 문제가 있습니다. 위에서 지적한 동일한 인스턴스에 대한 접근이 위배 되는 상황이 발생하며, 무엇보다 한 번의 테스트가 끝난 후 그와 관련된 데이터의 rollback 처리가 수월하지 않다는 문제가 있죠.
이 문제를 해결하기 위해선 이전에 구성한 fixture를 dependency injection 해주어야 하는데 이런 방식을 구상해 보았습니다.

app.dependency_overrides[engine.session] = session

위 코드가 잘 동작한다면 훌륭하겠지만, 아주 커다란 문제가 있습니다. 바로 session이라는 fixture는 db라는 다른 fixture를 매개변수로 받고 있다는 점과 fixture는 일종의 @property라는 점이죠. 아마도 당신이 위 코드를 실행하면 다음과 같은 오류를 만나게 될 것입니다.

'<sqlalchemy.orm.session.Session object at 0x1145520d0> is not a callable object'

난관에 부딪힌 상황이었지만, 다행히도 lambda 키워드를 사용해 생각보다는 간단하게 문제를 해결할 수 있었습니다. 또한 아이메디신에서 현재까지 개발된 테스트 코드는 Depends를 이용하기 위해선 TestClient를 이용해 router를 호출하는 때 외에는 없으므로 dependency_overrides를 TestClient를 호출하는 fixture에서 설정해주도록 구성해보았습니다.

@fixture
def client(session):
    # lambda식을 이용해 function 형태로 inject
    app.dependency_overrides[engine.session] = lambda: session
    from tests.test_routes import get_client
    return get_client()

위와 같은 구성으로 test 코드와 router에서 하나의 인스턴스 session을 사용하게 되어 router 테스트에서 완벽한 테스트 결과를 기대할 수 있게 되었습니다.

환경 구성의 누락 방지

만약 당신이 구현하고 있는 프로젝트의 데이터베이스에 대한 접근제어가 strict 하지 않은 상황이고, 프로젝트를 다른 환경에서 clone 하여 작업을 하려던 중 pytest.ini를 구성하는 걸 잊은 상태라면 pytest 명령어를 입력하는 순간

if database_exists(engine.url):
    drop_database(engine.url)

설정에 따라 실제 데이터 베이스에 모든 테이블이 삭제되는 현상이 발생할 것입니다. 이는 실제 서비스가 중단될지도 모르는 매우 위험한 상황을 내포하고 있죠.

제가 생각하는 이상적인 접근제어라면 database 인스턴스는 VPC 안에서 private 공간에 위치해야 하고, 해당 프로젝트를 개발하고 있는 Engineer라고 하더라도 직접적으로 데이터베이스에 접근할 수 있어서는 안 됩니다. 하지만 여러 가지 상황으로 위와 같은 환경 구성이 어려운 경우 다음과 같은 검증 식을 conftest.py 최상단에 추가 함으로써, 데이터베이스를 삭제하는 실수를 방지할 수 있을것입니다.

assert settings.APP_ENV == 'test', 'test 환경에서만 test 가능, `pytest.ini` 확인 필!'

마치며

단위테스트는 프로젝트를 개발해 나감에 있어 아주 기본적이지만 매우 중요한 절차이기 때문에 다른 포스팅 보단 조금 더 심도 있게 다루어 보았습니다. 아마도 다른 주제로 또다시 단위 테스트에 대해서 포스팅을 작성할 것 같다는 생각이 들기도 하네요.
또 다른 단위 테스트 주제로 만나 뵙기를 희망합니다 :)