본문 바로가기
Web 개발/FAST API (인프런 강의 내용)

3 실습2 테스트 코드 - GET 전체조회 API

by yororing 2024. 8. 1.

앞 단계 참조 링크:

 

00 개요

  • 목적: GET 전체조회 API에 테스트 코드 추가하는 실습 진행
  • 실습하기 전 docker 컨테이너를 활성화하고 uvicorn으로 앱을 가동시킨 후 전체 todos 목록을 조회하려고 했는데 다음과 같은 ConfigError가 발생함
INFO:     127.0.0.1:58459 - "GET /todos?order=DESC HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\uvicorn\protocols\http\httptools_impl.py", line 411, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\uvicorn\middleware\proxy_headers.py", line 69, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\fastapi\applications.py", line 1054, in __call__
    await super().__call__(scope, receive, send)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\applications.py", line 123, in __call__
    await self.middleware_stack(scope, receive, send)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\middleware\errors.py", line 186, in __call__
    raise exc
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\middleware\errors.py", line 164, in __call__
    await self.app(scope, receive, _send)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\middleware\exceptions.py", line 65, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\routing.py", line 756, in __call__
    await self.middleware_stack(scope, receive, send)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\routing.py", line 776, in app
    await route.handle(scope, receive, send)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\routing.py", line 297, in handle
    await self.app(scope, receive, send)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\routing.py", line 77, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\routing.py", line 72, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\fastapi\routing.py", line 278, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\fastapi\routing.py", line 193, in run_endpoint_function
    return await run_in_threadpool(dependant.call, **values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\starlette\concurrency.py", line 42, in run_in_threadpool
    return await anyio.to_thread.run_sync(func, *args)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\anyio\to_thread.py", line 56, in run_sync
    return await get_async_backend().run_sync_in_worker_thread(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\anyio\_backends\_asyncio.py", line 2144, in run_sync_in_worker_thread
    return await future
           ^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\anyio\_backends\_asyncio.py", line 851, in run
    result = context.run(func, *args)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\src\main.py", line 58, in get_todos_handler
    todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\관리자\Desktop\projects\todos\Lib\site-packages\pydantic\main.py", line 574, in from_orm
    raise ConfigError('You must have the config attribute orm_mode=True to use from_orm')
pydantic.errors.ConfigError: You must have the config attribute orm_mode=True to use from_orm
  • 에러 내용: from_orm 을 사용하기 위해선 from_orm = true라는 config 속성을 줘야된다고 함
  • 해결 시도: 응답으로 주는 코드의 config 클래스 안에 from_orm = true라는 속성을 주면 해결됨
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List

class ToDoSchema(BaseModel):
    id: int
    contents: str
    is_done: bool

    class Config:
        from_attributes = True
        orm_mode = True          # 이거 추가 해야지 ConfigError 안남

class ToDoListSchema(BaseModel):
    todos: List[ToDoSchema]
  • 저장 후 127.0.0.1:<포트번호>/docs에 들어가서 해당 API를 Try it out 및 Execute 할 경우 정상 작동 확인

<그러나>

  • 다음 '01 GET 전체조회 API에 테스트 코드 추가하기'에서 test_main.py에 테스트 코드를 작성 후 pytest 명령어로 검증하려 할 때 다음과 같은 warning이 출력됨
================================= warnings summary =========================================
..\..\..\..\AppData\Local\Programs\Python\Python312\Lib\site-packages\pydantic\_internal\_config.py:272
  C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pydantic\_inter
    nal\_config.py:272: PydanticDeprecatedSince20: Support for class-based `config` is deprecated,
    use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 
    Migration Guide at https://errors.pydantic.dev/2.6/migration/
    warnings.warn(DEPRECATION_MESSAGE, DeprecationWarning)

..\..\..\..\AppData\Local\Programs\Python\Python312\Lib\site-packages\pydantic\_internal\_config.py:322
  C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pydantic\_inter
    nal\_config.py:322: UserWarning: Valid config keys have changed in V2:
  * 'orm_mode' has been renamed to 'from_attributes'
    warnings.warn(message, UserWarning)

tests/test_main.py::test_get_todos
tests/test_main.py::test_get_todos
tests/test_main.py::test_get_todos
  C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pydantic\main.p
    y:1165: PydanticDeprecatedSince20: The `from_orm` method is deprecated; 
    set `model_config['from_attributes']=True` and use `model_validate` instead. 
    Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration 
    Guide at https://errors.pydantic.dev/2.6/migration/
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
============================== short test summary info ======================================
FAILED tests/test_main.py::test_get_todos - AssertionError: assert {'todos': [{'...done': 
True}]} == {'todos': [{'...done': True}]}
====================== 1 failed, 1 passed, 5 warnings in 1.52s ==============================
  • 에러 내용:
    • class 기반 config는 더 이상 사용되지 않고 ConfigDict를 대신 사용하라고 함
    • 또한, orm_mode가 from_attributes로 바뀌었다고 함
  • 해결 시도: 응답으로 주는 코드의 class config: 를 class ConfigDict: 로 변경 + 앞서 추가한 orm_mode = True 옵션을 지워줌
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List

class ToDoSchema(BaseModel):
    id: int
    contents: str
    is_done: bool

    class ConfigDict:  # 이름 변경
        from_attributes = True
        # orm_mode = True  # 삭제

class ToDoListSchema(BaseModel):
    todos: List[ToDoSchema]
  • 이후 warning 없어진 것 확인
======================= short test summary info ================================= 
FAILED tests/test_main.py::test_get_todos - pydantic_core._pydantic_core.Validati
onError: 1 validation error for ToDoSchema
===================== 1 failed, 1 passed in 2.05s ===============================
  • 여러가지 trials 후 지금까지의 결론:
    • warning들을 해결하려고 조금씩 바꾸면 전체적으로 작동이 안 됨.
    • 하나 하나씩 원인들을 찾기에는 시간이 없어서 일단은 강의 내용을 따라가며 warning은 무시하기로 결정
  • 위에서 pytest failed 이유는 갑을 잘못 넣어서 그랬었음. 제대로 넣으니 통과됨!

01 GET 전체조회 API에 테스트 코드 추가하기 

1. main.py 확인

  • 테스트 코드 적용할 API 확인하기:
 # /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
 
 ...
 
 # GET Method 사용하여 전체 조회 API - DB 참조
@app.get("/todos", status_code=200)
def get_todos_handler(
    order: str | None = None,
    session: Session = Depends(get_db),
) -> ToDoListSchema:
    todos: List[ToDo] = get_todos(session=session)
    if order and order == "DESC":
#       return todos[::-1]
        return ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
#   return todos
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

...
  • 정보:
    • get 요청 사용
    • 경로: "/todos"
    • 함수명: 'get_todos_handler'
    •  정상적인 반환값: order라는 인자를 query str으로 받아 이 order에 따라 정렬되어 반환되는 ToDoListSchema(todos=[ToDoSchema.from_orm(todo) for todo in todos])

2. test_main.py에 쓰기

1) ASC 정렬 확인하는 테스트 코드 추가하기

  • def test_get_todos()를 test_main.py에 추가:
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용
from fastapi.testclient import TestClient
from main import app

client = TestClient(app=app)

def test_health_check():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}
    
def test_get_todos():	# 추가
    # order = ASC
    response = client.get("/todos")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id": 1, "contents": "FastAPI Section 0", "is_done": True},
            {"id": 2, "contents": "FastAPI Section 2", "is_done": True},   # 나의 DB에는 이렇게 저장되어있음
            {"id": 3, "contents": "FastAPI Section 3", "is_done": True},   # 나의 DB에는 이렇게 저장되어있음
        ]
    }
  • src 디렉토리에서 $ pytest 명령어 실행
$ pytest

  • 테스트 통과된 것을 확인
  • 만약에 DB에 저장된 값과 다른 값을 주고 $ pytest 실행 시 에러 발생
    • 예) test_main.py에서 "todos" 리스트 중 하나의 행 값을 제거하거나 (e.g., {“id”: 1, “contents”: “FastAPI Section 0”, “is_done”: True}) 행 안에서 하나의 값을 변경 (e.g., "is_done": False)하여 pytest 실행 시 1개 통과 못 했다고 뜸

# 위 내용
======================== short test summary info =========================
FAILED tests/test_main.py::test_get_todos - AssertionError: assert {...}
 ===================== 1 failed, 1 passed in 0.35s =======================
  • PyCharm 사용 시 상세 에러 메세지 확인 가능:
    • test_main.py에서 해당 line number 우측에 초록색 화살표 클릭 > Run ‘Python tests for tes…’ 클릭
    • → 어느 줄에서 어떤 에러가 발생했는지 더 상세하게 확인 가능

2) DESC 정렬 확인하는 테스트 코드 추가하기

  • Note: 함수를 따로 만들어서 테스트 검증해도 되나 우리는 하나의 함수 안에서 둘 다 (ASC, DESC) 검증할 것
    • 한 번에 하면 어디서 문제가 발생했는지 찾기 어렵다는 단점이 있음
    • 그러나 현재는 테스트 코드를 간단하게 API별로 실습하는 정도이기 때문에 정리를 위해서 한 종류의 API 당 한 함수로 맞추기 위해서 한 함수에 담는 것
  • test_main.py의 def test_get_todos() 안에 다음을 추가:
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용
from fastapi.testclient import TestClient
from main import app

client = TestClient(app=app)

def test_health_check():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}
    
def test_get_todos():
    # order = ASC
    response = client.get("/todos")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id": 1, "contents": "FastAPI Section 0", "is_done": True},
            {"id": 2, "contents": "FastAPI Section 2", "is_done": True},
        ]
            {"id": 3, "contents": "FastAPI Section 3", "is_done": True},
    }
    
    # order = DESC							# 추가
    response = client.get("/todos?order=DESC")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id":3, "contents": "FastAPI Section 3", "is_done": True},
            {"id":2, "contents": "FastAPI Section 2", "is_done": True},
            {"id":1, "contents": "FastAPI Section 0", "is_done": True},
        ]
    }
  • src 디렉토리에서 $ pytest 명령어 실행
$ pytest

  • 테스트 통과된 것을 확인
  • 만약에 DB에 저장된 값과 다른 값을 주고 $ pytest 실행 시 에러 발생 
    • 예) test_main.py에서 "todos" 리스트 중 행의 순서를 변경(i.e., line number 29 - 31 순서처럼 원래 출력되어야 하는 순서가 아니게끔 설정) 후 pytest 실행 시 1개 통과 못 했다고 뜸

# 위 내용
============================== short test summary info =============================
ERROR tests/test_main.py
!!!!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!!!!
================================== 1 error in 0.34s =============================
  • 이러하여 DESC (역정렬) 되는 query str 부분도 잘 검증되고 있는 것을 확인함!

이때까지의 코드들:

  • /c/Users/관리자/Desktop/projects/todos/src/main.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/orm.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/connection.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/repository.py
  • /c/Users/관리자/Desktop/projects/todos/src/schema/request.py
  • /c/Users/관리자/Desktop/projects/todos/src/schema/response.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py (수정 - 추가)
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from typing import List	

from database.connection import get_db	
from database.repository import get_todos, get_todo_by_todo_id, create_todo, update_todo, delete_todo
from database.orm import ToDo
from schema.response import ToDoSchema, ToDoListSchema
from schema.request import CreateToDoRequest

app = FastAPI()

# 첫 화면 API
@app.get("/")
def health_check_handler():
    return {"ping": "pong"}

# GET Method 사용하여 전체 조회 API - DB 참조
@app.get("/todos", status_code=200)
def get_todos_handler(
    order: str | None = None,
    session: Session = Depends(get_db),
) -> ToDoListSchema:
    todos: List[ToDo] = get_todos(session=session)
    if order and order == "DESC":
#       return todos[::-1]
        return ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
#   return todos
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(
    todo_id: int,
    session: Session=Depends(get_db),
    ) -> ToDoSchema:
    todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
    if todo:
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

# POST Medthod 사용하여 todo 생성 API
@app.post("/todos", status_code=201)
def create_todo_handler(
    request: CreateToDoRequest,
    session: Session = Depends(get_db),
) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)            # id=None
    todo: ToDo = create_todo(session=session, todo=todo) # id=int
    return ToDoSchema.from_orm(todo)

# PATCH Method 사용하여 is_done 값 수정 API
@app.patch("/todos/{todo_id}", status_code=200)
def update_todo_handler(
    todo_id: int,
    is_done: bool = Body(..., embed=True),
    session: Session = Depends(get_db),
    ):
    todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
    if todo:
        # update - is_done값이 True이면 todo.done() 실행, False이면 todo.undone() 실행
        todo.done() if is_done else todo.undone()
        todo: ToDo = update_todo(session=session, todo=todo)
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

# DELETE Method 사용하여 todo 아이템 삭제 API
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo_handler(
    todo_id: int,
    session: Session = Depends(get_db),
    ):
    todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="ToDo Not Found")
    delete_todo(session=session, todo_id=todo_id)
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
from schema.request import CreateToDoRequest

Base = declarative_base()

# ToDo 클래스 모델링 한 것
class ToDo(Base):
    __tablename__ = 'todo'
    
    id = Column(Integer, primary_key=True, index=True)
    contents = Column(String(256), nullable=False)
    is_done = Column(Boolean, nullable=False)
    
    def __repr__(self):
        return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
    
    @classmethod
    def create(cls, request: CreateToDoRequest) -> "ToDo":
        return cls(
            # id값은 DB에 의해 자동으로 결정 되서 별도로 지정안해줌
            contents=request.contents,
            is_done=request.is_done,
        )
        
    def done(self) -> "ToDo":
        # ToDo의 is_done 값을 True로 변경 후 ToDo 반환
        self.is_done = True
        return self
    
    def undone(self) -> "ToDo":
        # ToDo의 is_done 값을 False로 변경 후 ToDo 반환
        self.is_done = False
        return self
# /c/Users/관리자/Desktop/projects/todos/src/database/connection.py 내용
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "mysql+pymysql://root:todos@127.0.0.1:3306/todos"

engine = create_engine(DATABASE_URL, echo=True)
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    session = SessionFactory()
    try:
        yield session
    finally:
        session.close()
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from typing import List

from database.orm import ToDo

def get_todos(session: Session) -> List[ToDo]:
    return list(session.scalars(select(ToDo)))

def get_todo_by_todo_id(session: Session, todo_id: int) -> ToDo | None:
    return session.scalar(select(ToDo).where(ToDo.id == todo_id))

def create_todo(session: Session, todo: ToDo) -> ToDo:
    session.add(instance=todo)
    session.commit()
    session.refresh(instance=todo)
    return todo

def update_todo(session: Session, todo: ToDo) -> ToDo:
    session.add(instance=todo)
    session.commit()
    session.refresh(instance=todo)
    return todo

def delete_todo(session: Session, todo_id: ToDo) -> None:
    session.execute(delete(ToDo).where(ToDo.id == todo_id))
    session.commit()
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용

from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
    contents: str
    is_done: bool
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List

class ToDoSchema(BaseModel):
    id: int
    contents: str
    is_done: bool

    class Config:
        from_attributes = True
        orm_mode = True # 이거 추가 해야지 ConfigError 안남

class ToDoListSchema(BaseModel):
    todos: List[ToDoSchema]
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용
from fastapi.testclient import TestClient
from main import app

client = TestClient(app=app)

def test_health_check():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}
    
def test_get_todos():				# 추가
    # order = ASC
    response = client.get("/todos")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id": 1, "contents": "FastAPI Section 0", "is_done": True},
            {"id": 2, "contents": "FastAPI Section 2", "is_done": True},
        ]
            {"id": 3, "contents": "FastAPI Section 3", "is_done": True},
    }
    
    # order = DESC
    response = client.get("/todos?order=DESC")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id":3, "contents": "FastAPI Section 3", "is_done": True},
            {"id":2, "contents": "FastAPI Section 2", "is_done": True},
            {"id":1, "contents": "FastAPI Section 0", "is_done": True},
        ]
    }

(test_main.py 수정 - 추가)

'Web 개발 > FAST API (인프런 강의 내용)' 카테고리의 다른 글

3 실습4 테스트 코드 - PyTest Fixture  (0) 2024.08.03
3 실습3 테스트 코드 - PyTest Mocking  (0) 2024.08.02
3 실습1 PyTest 세팅  (0) 2024.07.22
3 테스트 코드 PyTest  (0) 2024.06.01
2 실습11 ORM 적용 - DELETE API  (0) 2024.06.01

관련글

티스토리툴바