TDD 를 이용하여서 3개의 새로운 endpoint를 RESTful 개발로 진행하겠습니다.
/summaries | GET | READ | get all summaries |
/summaries/:id | GET | READ | get a single summary |
/summaries | POST | CREATE | add a summary |
각각의 엔드포인트에서
1. 테스트를 작성합니다.
2. 테스트 코드가 실패하는 것을 확인합니다.(레드)
3. 테스트 코드가 통과 되도록 코드를 작성합니다(그린)
4. 리팩토링합니다.
Post Route
app 디렉토리 아래 api 디렉토리 아래 summaries.py 모듈을 작성하겠습니다.
# project/app/api/summaries.py
from fastapi import APIRouter, HTTPException
from app.api import crud
from app.models.pydantic import SummaryPayloadSchema, SummaryResponseSchema
router = APIRouter()
@router.post("/", response_model=SummaryResponseSchema, status_code=201)
async def create_summary(payload: SummaryPayloadSchema) -> SummaryResponseSchema:
summary_id = await crud.post(payload)
response_object = {
"id": summary_id,
"url": payload.url
}
return response_object
payload가 SummaryPayloadSchema 인 핸들러를 호출합니다. 그리고 REQUEST BODY의 데이터에 대한 유효성 검증을 수행하게 됩니다.
asymc 키워드를 통해서 handler를 작성하였으므로 DB와 커뮤니케이션을 하는 것에 blocking I/O가 발생하지 않게 됩니다.
다음으로 crud.py 파일을 api 디렉토리 아래 명세하겠습니다.
# project/app/api/crud.py
from app.models.pydantic import SummaryPayloadSchema
from app.models.tortoise import TextSummary
async def post(payload: SummaryPayloadSchema) -> int:
summary = TextSummary(
url=payload.url,
summary="dummy summary",
)
await summary.save()
return summary.id
`post`로 명명된 utility 함수를 만들었습니다. 스프링부트의 서시스에 해당한다고 보면 좋습니다.
다음으로 response_model 로 사용할 새로운 pydantic 모델을 정의해야 합니다 .
@router.post("/", response_model=SummaryResponseSchema, status_code=201)
다음과 같이 pydantic.py를 업데이트하세요 .
# project/app/models/pydantic.py
from pydantic import BaseModel
class SummaryPayloadSchema(BaseModel):
url: str
class SummaryResponseSchema(SummaryPayloadSchema):
id: int
위 코드는 스프링부트의 dto, dao와 매칭하면 쉽게 이해 할 수 있습니다.
새롭게 라우터를 만들었으므로 맵핑해주도록 하겠습니다.
# project/app/main.py
import logging
from fastapi import FastAPI
from app.api import ping, summaries # updated
from app.db import init_db
log = logging.getLogger("uvicorn")
def create_application() -> FastAPI:
application = FastAPI()
application.include_router(ping.router)
application.include_router(summaries.router, prefix="/summaries", tags=["summaries"]) # new
return application
app = create_application()
@app.on_event("startup")
async def startup_event():
log.info("Starting up...")
init_db(app)
@app.on_event("shutdown")
async def shutdown_event():
log.info("Shutting down...")
prefix로 "/summaries"를 붙여서 처리 할 수 있습니다. 해당 모듈내의 호출되는 엔드포인트에 모두 적용됩니다.
그리고 OpenAPI schema에 엔드포인트를 tag 키워드 아규먼트로 등록하면 그룹화 시켜 관리할 수 있습니다.
http --json POST http://localhost:8004/summaries/ url=http://testdriven.io
HTTPie 유틸리티를 이용하여 테스트 해 볼 수 있습니다.
$ http --json POST http://localhost:8004/summaries/ url=http://testdriven.io
HTTP/1.1 201 Created
content-length: 37
content-type: application/json
date: Sat, 29 Oct 2022 15:18:20 GMT
server: uvicorn
{
"id": 1,
"url": "http://testdriven.io"
}
Test
test_summaries.py 모듈을 작성합니다. 아래 테스트 코드를 작성합니다.
# project/tests/test_summaries.py
import json
import pytest
def test_create_summary(test_app_with_db):
response = test_app_with_db.post("/summaries/", data=json.dumps({"url": "https://foo.bar"}))
assert response.status_code == 201
assert response.json()["url"] == "https://foo.bar"
conftest 모듈에 새로운 fixture를 등록합니다.
import os
import pytest
from starlette.testclient import TestClient
from app.main import create_application
from app.config import get_settings, Settings
def get_settings_override():
return Settings(testing=1, database_url=os.environ.get("DATABASE_TEST_URL"))
@pytest.fixture(scope="module")
def test_app():
# set up
app = create_application()
app.dependency_overrides[get_settings] = get_settings_override
with TestClient(app) as test_client:
# testing
yield test_client
# tear down
# new
@pytest.fixture(scope="module")
def test_app_with_db():
# set up
app = create_application()
app.dependency_overrides[get_settings] = get_settings_override
register_tortoise(
app,
db_url=os.environ.get("DATABASE_TEST_URL"),
modules={"models": ["app.models.tortoise"]},
generate_schemas=True,
add_exception_handlers=True,
)
with TestClient(app) as test_client:
# testing
yield test_client
# tear down
임포트도 잊지 말고 해야합니다.
from tortoise.contrib.fastapi import register_tortoise
$ docker-compose exec web python -m pytest
=============================== test session starts ===============================
platform linux -- Python 3.12.1, pytest-7.4.4, pluggy-1.3.0
rootdir: /usr/src/app
plugins: anyio-4.2.0
collected 2 items
tests/test_ping.py . [ 50%]
tests/test_summaries.py . [100%]
================================ 2 passed in 0.19s ================================
이번에는 반대로 예외 상황이 발생 할 경우의 테스트 코드도 작성합니다.
# project/tests/test_summaries.py
import json
import pytest
def test_create_summary(test_app_with_db):
response = test_app_with_db.post("/summaries/", data=json.dumps({"url": "https://foo.bar"}))
assert response.status_code == 201
assert response.json()["url"] == "https://foo.bar"
def test_create_summaries_invalid_json(test_app):
response = test_app.post("/summaries/", data=json.dumps({}))
assert response.status_code == 422
assert response.json() == {
"detail": [
{
"input": {},
"loc": ["body", "url"],
"msg": "Field required",
"type": "missing",
"url": "https://errors.pydantic.dev/2.5/v/missing",
}
]
}
프로젝트 구조는 이제 아래와 같습니다.
├── .gitignore
├── docker-compose.yml
└── project
├── .dockerignore
├── Dockerfile
├── app
│ ├── __init__.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── crud.py
│ │ ├── ping.py
│ │ └── summaries.py
│ ├── config.py
│ ├── db.py
│ ├── main.py
│ └── models
│ ├── __init__.py
│ ├── pydantic.py
│ └── tortoise.py
├── db
│ ├── Dockerfile
│ └── create.sql
├── entrypoint.sh
├── migrations
│ └── models
│ └── 0_20211227001140_init.sql
├── pyproject.toml
├── requirements.txt
└── tests
├── __init__.py
├── conftest.py
├── test_ping.py
└── test_summaries.py
GET single summary Route
test 코드 작성을 해보겠습니다.
def test_read_summary(test_app_with_db):
response = test_app_with_db.post("/summaries/", data=json.dumps({"url": "https://foo.bar"}))
summary_id = response.json()["id"]
response = test_app_with_db.get(f"/summaries/{summary_id}/")
assert response.status_code == 200
response_dict = response.json()
assert response_dict["id"] == summary_id
assert response_dict["url"] == "https://foo.bar"
assert response_dict["summary"]
assert response_dict["created_at"]
테스트가 fail 되는지 확인합니다.
> assert response.status_code == 200
E assert 404 == 200
E + where 404 = <Response [404]>.status_code
아래 handler를 명세해줍니다.
@router.get("/{id}/", response_model=SummarySchema)
async def read_summary(id: int) -> SummarySchema:
summary = await crud.get(id)
return summary
path parameter를 통해서 특정 PK 값을 가진 summary에 대한 정보를 DB에 쿼리 할 수 있습니다.
이에 대한 서비스 로직을 crud 모듈의 get 함수를 호출하여 처리 할 수 있습니다.
async def get(id: int) -> Union[dict, None]:
summary = await TextSummary.filter(id=id).first().values()
if summary:
return summary
return None
여기서는 Values 메서드를 사용하여 ValuesQuery 객체를 생성했습니다 . 그런 다음 TextSummary존재하는 경우 이를 사전으로 반환했습니다.
Union[dict, None]는 Optional[dict] 와 동일하므로 둘중 선호 하는 것을 자유롭게 사용하면 됩니다.
다음은 tortoise 모듈에서 pydantic을 생성 해 보도록 하겠습니다.
# project/app/models/tortoise.py
from tortoise import fields, models
from tortoise.contrib.pydantic import pydantic_model_creator # new
class TextSummary(models.Model):
url = fields.TextField()
summary = fields.TextField()
created_at = fields.DatetimeField(auto_now_add=True)
def __str__(self):
return self.url
SummarySchema = pydantic_model_creator(TextSummary) # new
`project/app/api/summaries.py` 모듈의 SummarySchema클래스를 임포트합니다.
from app.models.tortoise import SummarySchema
$ docker-compose exec web python -m pytest
=============================== test session starts ===============================
platform linux -- Python 3.12.1, pytest-7.4.4, pluggy-1.3.0
rootdir: /usr/src/app
plugins: anyio-4.2.0
collected 4 items
tests/test_ping.py . [ 25%]
tests/test_summaries.py ... [100%]
================================ 4 passed in 0.20s ================================
summary ID가 존재 하지 않을 경우의 테스트 코드도 작성합니다.
def test_read_summary_incorrect_id(test_app_with_db):
response = test_app_with_db.get("/summaries/999/")
assert response.status_code == 404
assert response.json()["detail"] == "Summary not found"
그리고 테스트 하면 fail이 납니다.
> assert response.status_code == 404
E assert 200 == 404
E + where 200 = <Response [200]>.status_code
코드 수정을 해줍니다.
@router.get("/{id}/", response_model=SummarySchema)
async def read_summary(id: int) -> SummarySchema:
summary = await crud.get(id)
if not summary:
raise HTTPException(status_code=404, detail="Summary not found")
return summary
이제는 아래와 같이 테스트 코드가 통과 됩니다.
=============================== test session starts ===============================
platform linux -- Python 3.12.1, pytest-7.4.4, pluggy-1.3.0
rootdir: /usr/src/app
plugins: anyio-4.2.0
collected 5 items
tests/test_ping.py . [ 20%]
tests/test_summaries.py .... [100%]
================================ 5 passed in 0.21s ================================
계속 진행하기 전에 브라우저 or Curl or HTTPie or API 문서를 통해 새 엔드포인트를 수동으로 테스트하세요.
GET all summaries Route
다건 엔드포인트 테스트 코드를 만듭니다.
def test_read_all_summaries(test_app_with_db):
response = test_app_with_db.post("/summaries/", data=json.dumps({"url": "https://foo.bar"}))
summary_id = response.json()["id"]
response = test_app_with_db.get("/summaries/")
assert response.status_code == 200
response_list = response.json()
assert len(list(filter(lambda d: d["id"] == summary_id, response_list))) == 1
테스트 fail나는지 확인하고 아래 handler코드를 추가해줍니다.
@router.get("/", response_model=List[SummarySchema])
async def read_all_summaries() -> List[SummarySchema]:
return await crud.get_all()
typing 모듈에서 List를 임포트 해주세요.
from typing import List
crud.py 모듈에서 get_all 함수를 정의해줍니다.
async def get_all() -> List:
summaries = await TextSummary.all().values()
return summaries
테스트 코드를 돌려보고 통과되는지 확인합니다.
=============================== test session starts ===============================
platform linux -- Python 3.12.1, pytest-7.4.4, pluggy-1.3.0
rootdir: /usr/src/app
plugins: anyio-4.2.0
collected 6 items
tests/test_ping.py . [ 16%]
tests/test_summaries.py ..... [100%]
================================ 6 passed in 0.22s ================================
개별 테스트 선택하여 실행
특정 테스트를 선택 하고 실행 할 수 있습니다.
예를 들어 `ping`이 포함된 모든 테스트를 실행하려면 다음을 수행합니다.
$ docker-compose exec web python -m pytest -k ping
=============================== test session starts ===============================
platform linux -- Python 3.12.1, pytest-7.4.4, pluggy-1.3.0
rootdir: /usr/src/app
plugins: anyio-4.2.0
collected 6 items / 5 deselected / 1 selected
tests/test_ping.py . [100%]
========================= 1 passed, 5 deselected in 0.14s =========================
따라서 test_summaries.py 의 5개 테스트를 skip한 반면 test_ping.py 의 단일 테스트가 실행(및 통과)된 것을 볼 수 있습니다 .
이 명령을 실행하면 어떤 테스트가 실행될까요?
Pytest Commands
pytest 명령어에 대해 잠깐 확인해 봅시다.
# normal run
$ docker-compose exec web python -m pytest
# disable warnings
$ docker-compose exec web python -m pytest -p no:warnings
# run only the last failed tests
$ docker-compose exec web python -m pytest --lf
# run only the tests with names that match the string expression
$ docker-compose exec web python -m pytest -k "summary and not test_read_summary"
# stop the test session after the first failure
$ docker-compose exec web python -m pytest -x
# enter PDB after first failure then end the test session
$ docker-compose exec web python -m pytest -x --pdb
# stop the test run after two failures
$ docker-compose exec web python -m pytest --maxfail=2
# show local variables in tracebacks
$ docker-compose exec web python -m pytest -l
# list the 2 slowest tests
$ docker-compose exec web python -m pytest --durations=2
'프로그래밍 언어 > 파이썬' 카테고리의 다른 글
FastAPI를 이용한 TDD 개발 - 7 (0) | 2024.03.19 |
---|---|
FastAPI를 이용한 TDD 개발 - 8 (0) | 2024.03.14 |
FastAPI를 이용한 TDD 개발 - 5 (0) | 2024.03.14 |
FastAPI를 이용한 TDD 개발 - 4 (0) | 2024.03.14 |
FastAPI를 이용한 TDD 개발 - 3 (0) | 2024.03.12 |