OO님 FastAPI 테스트는 어떻게 할까요?
제가 알아서 잘 … 하면 안되겠죠? 어떻게 접근하고 테스트를 짤 수 있을지 정리합니다.
항목 정리
시작하기 전에 필요한 정리할 목록을 나열합니다.
1.
테스트 프레임워크 선택 : unittest vs pytest
2.
비동기 코드 테스트 지원
3.
인프라 세팅: 최대한 단순화
4.
테스트 동작 세팅: 테스트 레이어 분리 실행 가능
필요사항 맞추기
테스트 프레임워크 선택
1.
Unittest
•
내장 라이브러리, class 기반 테스트 코드 작성, instance method 를 사용하여 테스트 결과를 검증합니다.
2.
Pytest
•
외부 라이브러리, function 기반 테스트 코드 작성, assert를 사용해 테스트 결과를 검증합니다.
더 자세한 설명은 아래 링크를 참조하세요.
간편한 사용성 및 코드 작성, 다중 인자 테스트 지원, 헷갈리지 않는 Mock 사용, 직접 사용 경험이 있는 pytest를 사용합니다.
비동기 코드 테스트 지원
1.
pytest 만으론 테스트 실행이 안됩니다.
# pytest 만으로 async 코드 실행 결과
...
PytestUnhandledCoroutineWarning: async def functions are not natively supported and have been skipped.
You need to install a suitable plugin for your async framework, for example:
- anyio
- pytest-asyncio
- pytest-tornasync
- pytest-trio
- pytest-twisted
warnings.warn(PytestUnhandledCoroutineWarning(msg.format(nodeid)))
Python
복사
2.
pytest-asyncio 라이브러리를 설치하면, async 테스트 코드를 돌릴 수 있습니다.
실행 시키면 asyncio: mode가 표시되는 걸 볼 수 있습니다.
>>>
...
asyncio: mode=Mode.AUTO
...
Python
복사
mode 설명
In auto mode, the pytest.mark.asyncio marker can be omitted, the marker is added automatically to async test functions.
•
auto: async 함수 알아서 동작합니다.
•
strict: async mark 명시해줘야 동작합니다.
# mode=auto
async def some_test():
...
# mode=strict
@pytest.mark.asyncio
async def some_test():
...
Python
복사
3.
테스트 코드들이 이벤트루프에서 동시에 비동기로 실행되면, 순서는 어떻게 맞춰야 하나 생각이 들 수 있습니다.
결론, 걱정할 필요 없습니다.
pytest-asyncio는 함수를 비동기적으로 동시에 트리거 하지 않습니다. (동시에 트리거 할 수 있는 pytest plugin 별도로 존재하긴 합니다. ) 비동기 함수라도 테스트 코드를 순차 실행 합니다.
테스트 코드 동작을 검증합니다.
import pytest
import random
import asyncio
async def get_random_sleep():
dur = random.randint(1, 10)
print(f"sleeping for {dur} seconds")
await asyncio.sleep(dur)
print("done sleeping")
@pytest.mark.asyncio
async def test_get_user_list():
await get_random_sleep()
@pytest.mark.asyncio
async def test_create_user():
await get_random_sleep()
@pytest.mark.asyncio
async def test_is_admin():
await get_random_sleep()
>>>
tests/app/user/services/test_user.py
.sleeping for 6 seconds
done sleeping
.sleeping for 3 seconds
done sleeping
.sleeping for 3 seconds
done sleeping
Python
복사
async 테스트 함수들이 순차 실행 되는 것을 볼 수 있습니다.
함수 내부에서 asyncio.gather()나 asyncio.TaskGroup() 으로 동시에 실행하면 비동기 동작하는 걸 확인할 수 있습니다.
# test gather
...
@pytest.mark.asyncio
async def test_get_user_list():
coro_list = []
for _ in range(3):
coro_list.append(get_random_sleep())
await asyncio.gather(*coro_list)
>>>
sleeping for 1 seconds
sleeping for 6 seconds
sleeping for 3 seconds
done sleeping
done sleeping
done sleeping
Python
복사
모듈 단위에서도 코드는 순차 실행이 됩니다.
# tests/test_sample/test_sample1.py
...
async def get_random_sleep():
dur = random.randint(1, 10)
print("test1")
print(f"sleeping for {dur} seconds")
await asyncio.sleep(dur)
print("done sleeping")
@pytest.mark.asyncio
async def test_get_user_list():
coro_list = []
for _ in range(2):
coro_list.append(get_random_sleep())
result = await asyncio.gather(*coro_list)
# tests/test_sample/test_sample2.py
...
async def get_random_sleep():
...
print("test2")
...
async def test_get_user_list():
...
# tests/test_sample/test_sample3.py
...
async def get_random_sleep():
...
print("test3")
...
async def test_get_user_list():
...
>>>
tests/test_sample/test_sample1.py
test1
sleeping for 5 seconds
test1
sleeping for 4 seconds
done sleeping
done sleeping
.
tests/test_sample/test_sample2.py
test2
sleeping for 3 seconds
test2
sleeping for 2 seconds
done sleeping
done sleeping
.
tests/test_sample/test_sample3.py
test3
sleeping for 5 seconds
test3
sleeping for 5 seconds
done sleeping
done sleeping
.
Python
복사
테스트 TearUp, TearDown 도 async 동작을 테스트 해봅니다. (pytest의 TearUp, TearDown 은 테스트 세션, 모듈, 클래스, 클래스 메소드, 함수 스코프로 적용 할 수 있습니다.)
# tests/test_sample/test_sample3.py
import pytest
import random
import asyncio
async def get_random_sleep(label=""):
dur = random.randint(1, 3)
print(f"sleeping for {dur} seconds: {label}")
await asyncio.sleep(dur)
print("done sleeping")
@pytest.fixture(scope="function")
async def tear_up_and_tear_down():
print("tear_up")
await get_random_sleep(label="tear_up")
yield
print("tear_down")
await get_random_sleep(label="tear_down")
@pytest.mark.asyncio
async def test_test(tear_up_and_tear_down):
coro_list = []
for _ in range(3):
coro_list.append(get_random_sleep(label="gather"))
result = await asyncio.gather(*coro_list)
>>>
tests/test_sample/test_sample3.py
tear_up
sleeping for 2 seconds: tear_up
done sleeping
sleeping for 2 seconds: gather
sleeping for 1 seconds: gather
sleeping for 3 seconds: gather
done sleeping
done sleeping
done sleeping
.
tear_down
sleeping for 2 seconds: tear_down
done sleeping
Python
복사
예측 가능한 순서로 함수들이 동작함을 확인할 수 있습니다.
4.
테스트 동작시 하나의 공통 이벤트 루프가 필요할 수 있습니다. API는 단위별로 하나의 이벤트 루프만 바라봅니다. 앱 라이프 사이클에 맞물려 동작하는 다른 비동기 라이브러리들의 경우 시작 이벤트루프가 종료되고 새로운 이벤트 루프가 생길 경우 이벤트루프 관련 에러가 발생하기도 합니다. 이를 해결하기 위해 event_loop를 세션 단위로 오버라이드 합니다.
...
@pytest.fixture(scope="session")
def event_loop(request):
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
Python
복사
정리하면,
1.
비동기 코드지만, 동기식으로 테스트를 진행할 수 있습니다. (순서가 꼬일 염려가 없습니다.)
2.
원하는 스코프에서 비동기 코드의 테스트 사전 준비, 테스트 동작, 테스트 사후 처리를 예측 가능한 순서로 정의할 수 있습니다.
인프라 세팅
안내
•
[23-11-15] 인프라 세팅은 예시만 있습니다. 제가 사용해보면서 최적화된 구성을 업로드 할 예정입니다. 더 좋은 방법이 있다면 코멘트 부탁드립니다.
◦
현재 고려사항들
▪
local 테스트 편의
▪
CI 연동: Github Action
1.
외부 종속성을 최소화 하기 위해 docker compose로 외부 서비스들을 동작 시킵니다.
2.
docker compose 실행 명령어들은 makefile에 스크립트를 정의해서 사용성을 편하게 합니다.
아래는 위 링크에서 가져온 동작 예시 코드 입니다. 이런 방법으로 세팅 할 수 있다는 것을 알려드리기 위한 예시입니다. 실제 동작하는 코드는 프로젝트 상황에 맞춰 다시 작성해야 합니다.
# docker-compose.test.yml
version: "3"
services:
redis_pubsub:
build:
context: .
dockerfile: Dockerfile
image: allocation-image
depends_on:
- postgres
- redis
- mailhog
environment:
- DB_HOST=postgres
- DB_PASSWORD=abc123
- REDIS_HOST=redis
- EMAIL_HOST=mailhog
- PYTHONDONTWRITEBYTECODE=1
volumes:
- ./src:/src
- ./tests:/tests
entrypoint:
- python
- /src/allocation/entrypoints/redis_eventconsumer.py
api:
image: allocation-image
depends_on:
- redis_pubsub
- mailhog
environment:
- DB_HOST=postgres
- DB_PASSWORD=abc123
- API_HOST=api
- REDIS_HOST=redis
- EMAIL_HOST=mailhog
- PYTHONDONTWRITEBYTECODE=1
- FLASK_APP=allocation/entrypoints/flask_app.py
- FLASK_DEBUG=1
- PYTHONUNBUFFERED=1
volumes:
- ./src:/src
- ./tests:/tests
entrypoint:
- flask
- run
- --host=0.0.0.0
- --port=80
ports:
- "5005:80"
postgres:
image: postgres:15
environment:
- POSTGRES_USER=allocation
- POSTGRES_PASSWORD=abc123
ports:
- "54321:5432"
redis:
image: redis:alpine
ports:
- "63791:6379"
mailhog:
image: mailhog/mailhog
ports:
- "11025:1025"
- "18025:8025"
YAML
복사
# makefile
build:
docker-compose build
up:
docker-compose up -d
test: up
docker-compose run --rm --no-deps --entrypoint=pytest api /tests/unit /tests/integration /tests/e2e
unit-tests:
docker-compose run --rm --no-deps --entrypoint=pytest api /tests/unit
integration-tests: up
docker-compose run --rm --no-deps --entrypoint=pytest api /tests/integration
e2e-tests: up
docker-compose run --rm --no-deps --entrypoint=pytest api /tests/e2e
logs:
docker-compose logs --tail=25 api redis_pubsub
down:
docker-compose down --remove-orphans
all: down build up test
Makefile
복사
# 실행은
$make {command}
Shell
복사
위의 방법은 도커 내부에서 pytest 를 실행하는 방법입니다.
pytest와 외부 서비스를 python script로 실행시키고 싶으시다면 아래의 링크를 참조 하실수도 있습니다.
테스트 동작 세팅
custom marker 등록하기
•
marker를 등록하는 방법 입니다.
# pytest.ini
[pytest]
markers =
layer_a: service_layer
layer_b: repository layer
YAML
복사
또는
# pyproject.toml
[tool.pytest.ini_options]
markers = [
"layer_a: service_layer",
"layer_b: repository layer",
]
TOML
복사
•
marker가 등록 되었는지 확인합니다.
$pytest --markers
...
@pytest.mark.layer_a: service_layer
@pytest.mark.layer_b: repository layer
...
Shell
복사
marker 달기
데커레이터로 함수에 마킹을 할 수 있습니다. 클래스 단위, 또는 모듈 단위로 마킹을 할 수도 있습니다.
import pytest
# 데커레이터
...
@pytest.mark.asyncio
@pytest.mark.layer_a
async def test_get_user_list():
coro_list = []
for _ in range(3):
coro_list.append(get_random_sleep())
result = await asyncio.gather(*coro_list)
# 클래스 데커레이터
@pytest.mark.webtest
class TestClass:
def test_startup(self):
pass
def test_startup_and_more(self):
pass
# 모듈 단위 마킹
# 'pytestmark'는 지정된 변수 이름입니다. 변경시 적용되지 않습니다.
# single marking
pytestmark = pytest.mark.webtest
# multiple marking
pytestmark = [pytest.mark.webtest, pytest.mark.slowtest]
Python
복사
marker 실행
# 특정 marker 만 실행하기 (-v: Increase verbosity)
pytest -v -m layer_a
# 특정 marker 들 실행하기
pytest -v -m layer_a -m layer_b
# 특정 marker 만 스킵하기
pytest -v -m "not layer_a"
# 특정 marker 들을 스킵하기
pytest -v -m "not layer_a" -m "not layer_b"
Python
복사
마커를 통한 관리가 번거로우시면 함수 이름으로 테스트 코드를 선택하는 방법도 있습니다. 자세한 설명은 아래 링크를 참조해주세요.
# 특정 이름 함수만 실행하기
$pytest -v -k repository
# 특정 이름 함수만 스킵하기
$pytest -k "not repository" -v
# 이외에도 or and ()등이 지원 됩니다.
Shell
복사
OO님 FastAPI 테스트는 이렇게 합니다.
1.
pytest를 사용합니다.
2.
pytest-asyncio를 사용해 비동기 코드를 검증합니다.
3.
docker-compose.yml를 통해 외부 서비스 의존성을 정의하고 실행합니다.
4.
pytest mark를 사용해 레이어를 분리 실행합니다.
“May the test be with you!”
Ref.
블로그 메인
