[Python] 비동기 테스트를 하려면? (a.k.a pytest-asyncio)

비동기 테스트!

들어가며

python은 async/await 문법과 single-thread 기반 event loop를 통해 비동기적인 작업을 지원합니다. 하지만, 비동기적으로 만들어진 코드를 테스트하기 위해서는 어떻게 해야 할까요? 이번 포스팅에서는 비동기 테스트를 위한 pytest 익스텐션인 pytest-asyncio에 대해 알아보겠습니다.

아래과 같은 개념을 알고 있어야 이 글을 이해하기 수월합니다.

• Pytest를 사용한 기본적인 테스트 코드 작성 방법
• pytest scope에 대한 이해
• Python의 비동기(Async) 개념(async/await, 코루틴, event loop)

그리고 이 포스팅을 통해 알 수 있는 내용은 다음과 같습니다.

• pytest-asyncio의 테스트 스코프(scope)
• async한 코드의 async한 테스트 코드를 작성하는 방법

---

비동기(async) 테스트가 필요한 경우?

비동기 테스트는 어떤 상황에서 필요할까요? 당연하게도 FastAPI와 같은 비동기적인 작업을 지원하는 프레임워크, 패키지를 사용한 코드를 테스트할 때 비동기 테스트를 수행하게 됩니다.

 
아래의 예시는 FastAPI로 작성된 간단한 엔드포인트를 테스트하는 샘플 코드입니다.

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Tomato"}

import pytest
from httpx import AsyncClient

from .main import app


@pytest.mark.asyncio # @pytest.mark.anyio를 변경한 예제
async def test_root():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

참고 : https://fastapi.tiangolo.com/advanced/async-tests/

 
위 코드는 fastapi로 async하게 정의된 엔드포인트를 테스트하는 예시입니다. 테스트 코드에서는 AsyncClinet를 사용해 비동기 엔드포인트를 호출하고 async with 블럭 내에서 요청/응답을 처리합니다.
 
이외에도 비동기 Sqlalchemy IO 요청에 대한 테스트, 직접 작성한 async/await 함수에 대한 테스트를 수행할 경우 비동기적인 테스트를 작성하게 됩니다.

---

pytest with pytest-asyncio

이번에는 pytest-asyncio에 대해 알아보겠습니다. pytest는 python에서 사용되는 테스트 프레임워크 입니다. 공식 문서에서는 pytest를 다음과 같이 정의하고 있습니다.

The pytest framework makes it easy to write small, readable tests, and can scale to support complex functional testing for applications and libraries.

참고 : https://docs.pytest.org/en/8.2.x/

 
이번 포스팅의 주인공인 pytest-asyncio 는 pytest의 플러그인으로 비동기 테스트를 간편하게 테스트할 수 있게 해 줍니다. pytest도 기본적으로 async 한 함수에 대한 테스트를 수행할 수 있지만, asyncio.run 함수로 비동기 함수를 처리해야하는 불편함이 있습니다.

import asyncio
import pytest

async def async_add(a, b):
    await asyncio.sleep(1)  # 비동기 작업을 시뮬레이션
    return a + b

def test_async_add():
    result = asyncio.run(async_add(3, 4)) # <- async 함수를 await으로 처리하지 못함...
    assert result == 7

 
위 코드에서처럼 async하게 작성된 async_add 함수를 asyncio.run 함수 내에서 실행해 테스트를 진행합니다. 이러한 형태는 복잡한 비동기 코드를 테스트하기에는 부족해 보입니다.
 
다음은 pytest-asyncio를 사용한 경우 테스트 코드입니다.

import asyncio
import pytest

async def async_add(a, b):
    await asyncio.sleep(1)  # 비동기 작업을 시뮬레이션
    return a + b

@pytest.mark.asyncio
async def test_async_add():
    result = await async_add(3, 4) # 코루틴을 await하게 호출!
    assert result == 7

 
pytest-asyncio를 사용한 경우 테스트 함수를 async def로 정의할 수 있고 비동기 함수를 테스트 코드 내에서 await 하게 호출할 수 있습니다. @pytest. mark.asyncio는 해당 테스트가 비동기로 실행되어야 함을 의미합니다. 이 방법은 비동기 테스트를 보다 직관적으로 테스트할 수 있게 해 줍니다.

---

pytest-asyncio 테스트 모드

pytest-asyncio는 strict, auto인 두 가지 테스트 모드를 지원합니다. 

Strict mode (default)

실행 옵션을 지정하지 않는다면 default 모드인 strict mode로 테스트가 실행됩니다. asyncio 마커(@pytest.mark.asyncio, @pytest_asyncio. fixture)로 데코데이트 된 요소만 pytest-asyncio에서 처리되도록 합니다. 다른 비동기 테스트 라이브러리 혹은 플러그인(ie. anyio)등을 사용할 경우 다른 테스트 라이브러리의 실행을 방해하지 않기(공존을) 위한 테스트 모드입니다.

Auto mode

auto mode는 모든 비동기 테스트를 asyncio 마커를 자동으로 추가합니다. 이 모드는 asynio를 유일한 비동기 라이브러리로 사용할 경우를 위한 모드입니다. 만일 여러 비동기 라이브러리를 함께 사용한다면 strict모드가 권장됩니다.

모드 적용

테스트 모드는 테스트 실행 시 asyncio_mode 옵션, pytest.ini, 그리고 pyporject.toml에 명시할 수 있습니다
 
1. CLI에서 옵션으로 명시할 경우

pytest --asyncio-mode=auto

 
2. pytest.ini 에서 사용할 경우

[pytest]
asyncio_mode = strict

 
3. pyporject.toml의 경우

[tool.pytest.ini_options]
asyncio_mode = "auto"

---

Event loop와 test scope

pytest-asyncio는 기본적으로 function scope의 event loop를 사용합니다. 때문에 좀 더 범위가 넓은 테스트에서 좁은 scope의 fixture를 사용하는 경우 하나의 실행 흐름 내에서 서로 다른 event loop를 사용하게 되어 에러가 발생할 수 있습니다. 다음 코드를 통해 살펴보겠습니다.

 
다음은 session 스코프 테스트에서 function 스코프 테스트를 사용하는 예제 코드입니다.

# test_scope.py
import pytest  
import asyncio  
import pytest_asyncio  

@pytest_asyncio.fixture(scope='function')
async def sleeping_fixture(): # function 스코프의 테스트 fixture
    print("function sleeping...")  
    await asyncio.sleep(0.1)  
    yield "in the middle of function"  
    await asyncio.sleep(0.1)  
    print("function sleeping done")  

@pytest.mark.asyncio(scope="session") # session 스코프의 테스트
async def test_sleeping(sleeping_fixture): # function 스코프의 fixture사용
    assert 1 == 1

 
위 코드를 실행할 경우 테스트와 fixture이 서로 다른 event loop를 사용하게 되어 다음 에러가 발생합니다.
 

tests/test_scope.py - pytest_asyncio.plugin.MultipleEventLoopsRequestedError: Multiple asyncio event loops with different scopes have been requested

 
이 같은 에러를 방지하기 위해서 fixture에서 사용하는 event loop와 테스트 코드에서 사용하는 event loop를 동기화시켜주는 것이 중요합니다.
 
테스트를 다음과 같이 변경하면 에러가 발생하지 않습니다.

@pytest.mark.asyncio(scope="function") # session 스코프의 테스트
async def test_sleeping(sleeping_fixture): # function 스코프의 fixture사용
    assert 1 == 1

---

Async 한 테스트 코드 작성해 보기

지금까지 pytest-asyncio를 사용하는 기본 지식에 대해 알아보았습니다. 이번에는 비동기 IO를 지원하는 sqlalchemy를 사용해 비동기 테스트를 작성해 보겠습니다.
 
PostgreSQL db를 사용하는 테스트 환경을 구성하였습니다.

version: '3.9'

services:
  postgres:
    image: postgres
    restart: always
    ports:
      - "5432:5432"
    environment:
      POSTGRES_DB: "test"
      POSTGRES_USER: "test"
      POSTGRES_PASSWORD: "test"

 
docker-compose로 db 컨테이너를 실행합니다.

docker-compose up -d

 
테스트를 위한 ORM 모델을 선언합니다. User 모델은 id, name 칼럼을 갖는 간단한 모델입니다.

from sqlalchemy import Column, Integer, String  
from sqlalchemy.ext.declarative import declarative_base  

Base = declarative_base()

class User(Base):
    tablename = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

 
conftest.py를 작성합니다. 해당 설정에서는 db engine, 비동기 session과 관련된 fixture를 정의합니다. fixture는 모두 async def 형태로 선언하고 @pytest_asyncio. fixture로 데코레이션 합니다.

import pytest_asyncio  
from sqlalchemy.ext.asyncio import async_sessionmaker  
from tests.model import Base  
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession  

DATABASE_URL = 'postgresql+asyncpg://test:test@localhost/test'  

@pytest_asyncio.fixture(scope='session')  
async def async_engine():  
    engine = create_async_engine(DATABASE_URL)  
    async with engine.begin() as conn:  
        await conn.run_sync(Base.metadata.drop_all)  
        await conn.run_sync(Base.metadata.create_all)  
    yield engine  
    await engine.dispose()  


@pytest_asyncio.fixture(scope='session')  
async def async_session(async_engine):  
    async_session = async_sessionmaker(  
    async_engine, class_=AsyncSession, expire_on_commit=False)  
    yield async_session()

 
테스트 코드를 작성합니다. user_fixture는 user 객체를 영속화하고 테스트 세션이 종료되면 영속화된 user 객체를 데이터베이스로부터 제거합니다. test_user_select 테스트에서는 영속화된 객체를 조회하고 검증합니다

# test_example.py
import pytest  
import pytest_asyncio  
from tests.model import User  
from sqlalchemy import Select  


@pytest_asyncio.fixture(scope='session')
async def user_fixture(async_session):  
    user = User(name="test") 
    async_session.add(user) # user 저장 
    await async_session.commit()  
    yield user  
    await async_session.delete(user) # user 제거
    await async_session.commit()


@pytest.mark.asyncio(scope="session")  
async def test_user_select(user_fixture, async_session):  
    result = await async_session.execute(  
        Select(User).where(User.name == user_fixture.name)  
    ) # user를 이름으로 조회
    user = result.scalars().first()  
    assert user.name == "test"

 
테스트를 실행하여 검증합니다.

pytest

tests/test_example.py .                                                                           [100%]
================ 1 passed, 1 warning in 0.31s =================

---

마치며

python의 비동기 함수와 테스트 코드를 작성하는 기본 배경이 없이는 이해하기 힘든 내용이 많았을 것 같습니다. 글을 쓰면서 event loop, pytest의 scope, fixture에 대한 설명이 부족하다는 생각이 들었는데요... 시간이 된다면 python의 비동기 처리에 대한 내용과 pytest의 fixture, scope에 관한 글을 작성해 보겠습니다.

---

참고

https://docs.pytest.org/en/stable/contents.html
https://dev-in-seoul.tistory.com/46
https://kkminseok.github.io/posts/pytest_Fixtures_Scope/
https://pytest-asyncio.readthedocs.io/en/latest/
https://levelup.gitconnected.com/mastering-unit-testing-in-python-part-3-unleashing-the-power-of-python-async-566202448e95