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

3 실습7 테스트 코드 - PATCH API

by yororing 2024. 8. 5.

앞 단계 참조 링크:

 

00 개요

  • 목적: ToDo를 업데이트/변경하는 (main.py에) update_todo_handler()로 정의된 POST API에 테스트 코드 추가하는 실습 진행

01 PATCH API에 테스트 코드 추가하기

1. main.py 확인

  • 테스트 코드 적용할 API 확인하기:
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용

...

# 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)  # mocking 적용 부분
    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")

...
  • mocking 적용할 부분:
    • todo: ToDo | None = get_todo_by_todo_id(session=session, todo_id=todo_id) → repository.py의 get_todo_by_todo_id() 함수
    • todo.done() if is_done else todo.undone() 
    • todo: ToDo = update_todo(session=session, todo=todo)
  • 정보:
    • patch 요청 사용
    • 경로: "/todos/{todo_id}"
    • 함수명: 'update_todo_handler'
    • path로 전달 받는 인자: todo_id
    • 요청 값 (request): is_done
      • → todo_id에 해당한는 ToDo를 get 요청하여 todo 변수에 저장 
      • → ToDo를 is_done값에 맞춰 업데이트
      • → update_todo()를 통해 실제 DB에 반영
    • 정상적인 반환값: ToDoSchema.from_orm(todo)
    • 예외 처리: todo_id값을 잘못 받은 경우 404 예외 처리


2. test_main.py에 쓰기

  • def test_update_todo()를 test_main.py의 끝부분에 다음을 추가:
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용

...

# PATCH Method 사용하여 is_done 값 수정 API 검증    # 추가
def test_update_todo(client, mocker):
    # 상태코드 200
    
    # get_todo_by_todo_id 검증
    mocker.patch(
        "src.main.get_todo_by_todo_id",
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    
    # done() 또는 undone() 검증
    # done = mocker.patch.object(ToDo, "done") # ToDo 객체의 done() 메소드 호출
    undone = mocker.patch.object(ToDo, "undone") # ToDo 객체의 undone() 메소드 호출
    
    # update_todo 검증
    mocker.patch(
        "src.main.update_todo",        # mocking 적용할 함수 및 반환값 설정
        return_value = ToDo(id=1, contents="todo", is_done=False),   # done 경우 is_done = True, undone 경우 is_done=False 주기
    )
    
    response = client.patch("/todos/1", json={"is_done": False})   # done 경우 "is_done": True, undone 경우 "is_done": False 주기
    
    # done() 또는 undone() 검증 - done 또는 undone이 한 번 호출되었는지 확인. 아닐 경우 AssertionError 발생
    # done.assert_called_once_with()
    undone.assert_called_once_with()
    
    assert response.status_code == 200
    assert response.json() == {"id":1, "contents": "todo", "is_done": False }
    
    # 상태코드 404
    mocker.patch(
        "src.main.get_todo_by_todo_id",        # mocking 적용할 함수 및 반환값 설정
        return_value = None,
    )   
    response = client.patch("/todos/1", json={"is_done": True})
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}
  • 설명:
    • def test_update_todo(client, mocker):
      • test_update_todo() 함수 정의 시 client fixture와 mocker fixture 사용
    • 200 처리:
      • #  get_todo_by_todo_id 검증
      • mocker.patch("src.main.get_todo_by_todo_id", return_value=ToDo(id=1, contents="todo", is_done=True),)
        • mocker.patch("mocking적용함수", return_value="반환값") 함수를 통해 mocking을 적용할 함수와 반환값 설정
      • # done() 또는 undone() 검증
      • undone = mocker.patch.object(ToDo, "undone")
        • mocker.patch.obejct(mocking대상_클래스객체, "mocking대상_클래스_메소드")를 통해 mocking하려는 부분 설정
      • # update_todo 검증
      • mocker.patch("src.main.update_todo", return_value=ToDo(id=1, contents="todo", is_done=False),)
        • mocker.patch("mocking적용함수", return_value="반환값") 함수를 통해 mocking을 적용할 함수와 반환값 설정
        • return_value가 "is_done" 값으로 False로 줬을 시 ToDo(...is_done=False)로 업데이트 되야 됨을 설정 
      • response = client.patch("/todos/1", json={"is_done": False})
        • patch API 요청
        • 요청값: json 형식의 "is_done"값을 False로 설정하여 False로 업데이트 해주도록 설정
      • undone.assert_called_once_with()
        • assert_called_once_with() (→PyTest의 mocker에 정의된 메소드)를 통해 undone()이 위의 patch API 요청 이후에 호출 되었는지를 검증하는 것
        • 만약 위의 patch API의 response 값으로 "is_done":True를 줄 경우, request가 True이기 때문에 undone()이 아닌 done()이 호출 될 것이므로 (→undone을 한 번도 호출이 안 될 것) 이 줄에서 AssertionError 발생할 것  
        • → $ pytest 검증 시 다음 메세지 터미널에 출력될 것:
          • "... FAILED tesets/test_main.py::test_update_todo - AssertionError: Expected 'undone' to be called once. Called 0 times. ========== 1 failed in 0.07s =========="
      • assert response.status_code == 200
        • 200 상태 코드 반환 검증 확인
      • assert response.json() == {"'id":1, "contents":"todo", "is_done":False} 
        • 최종 response값은 is_done=False 업데이트 후인 것을 반영하여 "is_done":False로 설정 
    • 404 예외 처리: 
      • mocker.patch("src.main.get_todo_by_todo_id", return_value=None,)
        • mocker에 patch로 요청
      • response = client.patch("/todos/1", json={"is_done": True})
        • request body로 is_done값을 아무 값으로 요청하여 ToDo 요청 (True 또는 False 아무거나 줘도 상관 없)
      • assert response.status_code == 404
        • 바라는 상태 코드 = 404
      • assert response.json() == {"detail": "ToDo Not Found"}
        • 바라는 return값 = {"detail": "ToDo Not Found"}

3. 검증 2가지 방법

1) 모든 테스트 한 번에 실행

  • src 디렉토리에서 $ pytest 명령어 실행 
    • 실행 전 uvicorn이 띄워져있는지, docker가 실행중인지 확인해야 됨
$ pytest

# 또는 다른 경로에 있을 경우 tests 경로를 지정해서 실행
# 예시 - todos/src 경로에 있을 경우
$ pytest tests

  • 정상적으로 통과된 것을 확인!
    • 메세지: ================= 5 passed in 0.27s ===================

2) 1개의 테스트 실행

  • src 디렉토리에서 $pytest 이후에 원하는 함수 경로 지정하여 실행
$ pytest 상대경로::함수명
$ pytest tests/test_main.py::test_update_todo

  • 정상적으로 통과된 것을 확인!
    • 메세지: ================= 1 passed in 0.25s ===================

 

이때까지의 코드들: 8개

  • /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/tests/conftest.py
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends
from sqlalchemy.orm import Session
from typing import List	

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

app = FastAPI()

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

# GET Method 사용하여 전체 조회 API
@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 src.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 src.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 src.schema.response import ToDoSchema
from src.database.orm import ToDo

def test_health_check(client):      # client 는 내가 conftest.py에서 정의한 fixture
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}

# GET Method 사용하여 전체 조회 API 검증
def test_get_todos(client, mocker):
    # order = ASC
    mocker.patch("src.main.get_todos", return_value=[
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDoSchema(id=2, contents="FastAPI Section 2", is_done=False),
    ])
    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": False},
        ]
            
    }
    
    # order = DESC
    response = client.get("/todos?order=DESC")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id":2, "contents": "FastAPI Section 2", "is_done": False},
            {"id":1, "contents": "FastAPI Section 0", "is_done": True},
        ]
    }

# GET Method 사용하여 단일 조회 API 검증
def test_get_todo(client, mocker):
    # 상태코드 200
    mocker.patch(
        "src.main.get_todo_by_todo_id",        # mocking 적용할 함수 및 반환값 설정
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )   
    response = client.get("/todos/1")   # path에 하위 서브 path 적어주기
    assert response.status_code == 200
    assert response.json() == {"id":1, "contents": "todo", "is_done": True }
    
    # 상태코드 404
    mocker.patch(
        "src.main.get_todo_by_todo_id",        # mocking 적용할 함수 및 반환값 설정
        return_value = None,
    )   
    response = client.get("/todos/1")
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}

# POST Medthod 사용하여 todo 생성 API 검증
def test_create_todo(client, mocker):
    create_spy = mocker.spy(ToDo, "create")
    mocker.patch(
        "src.main.create_todo",
        return_value=ToDo(id=1, contents="todo", is_done=False),
    )
    
    body = {
        "contents": "test",
        "is_done": False,
    }
    response = client.post("/todos", json=body)
    
    assert create_spy.spy_return.id is None
    assert create_spy.spy_return.contents == "test"
    assert create_spy.spy_return.is_done is False
    
    assert response.status_code == 201
    assert response.json() == {"id":1, "contents":"todo", "is_done":False}


# PATCH Method 사용하여 is_done 값 수정 API 검증    # 추가
def test_update_todo(client, mocker):
    # 상태코드 200
    
    # get_todo_by_todo_id 검증
    mocker.patch(
        "src.main.get_todo_by_todo_id",
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    
    # done() 또는 undone() 검증
    # done = mocker.patch.object(ToDo, "done") # ToDo 객체의 done() 메소드 호출
    undone = mocker.patch.object(ToDo, "undone") # ToDo 객체의 undone() 메소드 호출
    
    # update_todo 검증
    mocker.patch(
        "src.main.update_todo",
        return_value = ToDo(id=1, contents="todo", is_done=False),   # done 경우 is_done = True, undone 경우 is_done=False 주기
    )
    
    response = client.patch("/todos/1", json={"is_done": False})   # done 경우 "is_done": True, undone 경우 "is_done": False 주기
    
    # done() 또는 undone() 검증 - done 또는 undone이 한 번 호출되었는지 확인. 아닐 경우 AssertionError 발생
    # done.assert_called_once_with()
    undone.assert_called_once_with()
    
    assert response.status_code == 200
    assert response.json() == {"id":1, "contents": "todo", "is_done": False }
    
    # 상태코드 404
    mocker.patch(
        "src.main.get_todo_by_todo_id",        # mocking 적용할 함수 및 반환값 설정
        return_value = None,
    )   
    response = client.patch("/todos/1", json={"is_done": True})
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}
# /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py 내용
import pytest
from fastapi.testclient import TestClient
from src.main import app

@pytest.fixture
def client():
	return TestClient(app=app)