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

4 실습1 Refactoring - FastAPI Router

by yororing 2024. 8. 8.

앞 단계 참조 링크:

 

00 개요

  • 앞서 <섹션 1: REST API 실습>, <섹션 2: ORM을 사용하여 데이터베이스에 연결>, < 섹션 3: 테스트 코드 적용> 에서 우리는 API를 개발한 후 Swagger UI를 통해 실제 API를 호출해보고 테스트 코드를 적용해서 검증 과정을 마침
  • 섹션 4: 프로젝트를 refactoring 하면서 여러가지 웹 개발 관련된 주요 개념, FastAPI의 routing 기능 및 FastAPI routing의 prefix 기능을 알아볼 것

01 API 파일들 분리

  • 프로젝트가 커지면 main.py 안에 많은 코드들이 추가될 것이므로 지금부터 어떤 Domain 또는 Resource별로 파일 분리해주기
  • 현재 main.py 보면 health_check_handler()를 제외한 나머지는 todo 관련 API들임
  • → main.py에서 API 관련된 코드들 분리: src 디렉토리 안에 api 디렉토리 생성 및 api 디렉토리 안에 todo.py 파일 생성
$ pwd
/c/Users/관리자/Desktop/projects/todos/src

$ mkdir api
$ touch api/todo.py

 

02 Router 기능 사용

1. api/todo.py 작성

  • api 관련된 함수들을 api/todo.py로 옮기기 및 추가
  • 추가 내용:
    • fastapi에서 APIRouter 클래서 참조
    • APIRouter()를 담은 router 변수 생성
    • 모든 app 데코레이터를 router로 변환
      • 예) @app.get(...) 을 @router.get(...)으로 변환
      • → router라는 클래스를 통해 api들을 연결해주는 것
# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
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

router = APIRouter()


# GET Method 사용하여 전체 조회 API
@router.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 ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

# GET Method 사용하여 단일 조회 API
@router.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
@router.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
@router.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
@router.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)

2. main.py 수정

  • main.py에서 todo.py에 작성한 router를 연결해줘야 정상적으로 API들이 연동될 수 있음
  • 수정 내용:
    • 원래 있던 API 관련 함수들 제거 (todo.py로 옮겨 놨음)
    • from api import todo
    • app.include_router(todo.router)
      • FastAPI()의 include_router() 메소드 사용 
      • include_router()에 todo의 router 객체를 연결
      • → main.py의 "/" (root) 디렉토리 아래에 health_check_handler() API가 있을 것이고 그 아래에 todo.router들이 연결되는 구조가 되어, 이전과 동일하게 path들을 사용할 수 있게 되는 것 
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI
from src.api import todo

app = FastAPI()
app.include_router(todo.router)

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

3. test_main.py 수정

  • 작성한 코드들이 참조하는 모듈의 경로 수정: 우리가 검증하는 API 함수들이 더 이상 main.py에 있지 않고 todo.py에 있기 때문에  main이라고 작성된 부분을 알맞게 수정하기
  • main 부분을 api.todo로 수정
# /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.api.todo.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.api.todo.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.api.todo.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의 spy 기능 사용
    mocker.patch(
        "src.api.todo.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.api.todo.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.api.todo.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.api.todo.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_delete_todo(client, mocker):
    # 상태코드 204
    mocker.patch(
        "src.api.todo.get_todo_by_todo_id",        # mocking 적용할 함수 및 반환값 설정
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    mocker.patch(
        "src.api.todo.delete_todo",
        return_value = None,
    )
    response = client.delete("/todos/1")    # path에 하위 서브 path 적어주기
    assert response.status_code == 204      # status_code만 주고 response.json()는 assert 안 해봐도 됨! 반환값이 없을 것이기에
    
    # 상태코드 404
    mocker.patch(
        "src.api.todo.get_todo_by_todo_id",        # mocking 적용할 함수 및 반환값 설정
        return_value = None,
    )   
    response = client.delete("/todos/1")
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}

1) pytest 실행

  • src 디렉토리로 이동하여 $ pytest 실행해보기
더보기

$ pwd
/c/Users/관리자/Desktop/projects/todos/src
$ pytest
====================== test session starts ======================
platform win32 -- Python 3.12.1, pytest-8.3.2, pluggy-1.5.0
rootdir: C:\Users\관리자\Desktop\projects\todos\src
plugins: anyio-4.3.0, hypothesis-6.108.5, mock-3.14.0
collected 6 items

tests\test_main.py ......                                  [100%]

======================= 6 passed in 0.29s =======================

  • 모든 테스트 코드가 정상적으로 동작하는 것 확인
  • 즉, api router가 잘 바뀐 것을 확인!

03 Router의 Prefix 기능 사용

  • APIRouter()에는 prefix라는 기능이 있음
  • prefix를 사용하여 API마다 반복되는 path가 있을 경우 APIRouter() 안에 prefix 키워드를 설정하여 반복되는 url prefix 제거 가능 
  • APIRouter(prefix="/반복되는 path") 하면 해당 <반복되는 path>로 시작하는 API들의 url 경로에서 해당 <반복되는 path> 제거해주면 됨

1. main.py 수정

  • 수정 내용:
    • router = APIRouter(prefix="/todos")
    • "~" 안에 있는 /todos 들 찾아서 제거
# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
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

router = APIRouter(prefix='/todos')         # prefix 추가


# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)            # "" 안에 /todos 제거됨
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 ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

# GET Method 사용하여 단일 조회 API
@router.get("/{todo_id}", status_code=200)            # "" 안에 /todos 제거됨
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
@router.post("", status_code=201)            # "" 안에 /todos 제거됨
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
@router.patch("/{todo_id}", status_code=200)            # "" 안에 /todos 제거됨
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
@router.delete("/{todo_id}", status_code=204)            # "" 안에 /todos 제거됨
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)

1) pytest 실행

  • src 디렉토리로 이동하여 $ pytest 실행하여 정상작동하는지 검증해보기
더보기

관리자@DESKTOP-THDM2MN MINGW64 ~/Desktop/projects/todos/src
$ pytest
======================================== test session starts ================================
platform win32 -- Python 3.12.1, pytest-8.3.2, pluggy-1.5.0
rootdir: C:\Users\관리자\Desktop\projects\todos\src
plugins: anyio-4.3.0, hypothesis-6.108.5, mock-3.14.0
collected 6 items

tests\test_main.py ......                                                                      [100%]

========================================= 6 passed in 0.31s ================================

 

 

이때까지의 코드들: 9개

  • /c/Users/관리자/Desktop/projects/todos/src/main.py (수정 - 제거 및 추가)
  • /c/Users/관리자/Desktop/projects/todos/src/api/todo.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
from src.api import todo

app = FastAPI()
app.include_router(todo.router)

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

 

# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
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

router = APIRouter(prefix='/todos')


# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)            # "" 안에 /todos 제거됨
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 ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

# GET Method 사용하여 단일 조회 API
@router.get("/{todo_id}", status_code=200)            # "" 안에 /todos 제거됨
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
@router.post("", status_code=201)            # "" 안에 /todos 제거됨
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
@router.patch("/{todo_id}", status_code=200)            # "" 안에 /todos 제거됨
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
@router.delete("/{todo_id}", status_code=204)            # "" 안에 /todos 제거됨
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)
  • todo.py 추가

 

# /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.api.todo.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.api.todo.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.api.todo.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의 spy 기능 사용
    mocker.patch(
        "src.api.todo.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.api.todo.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.api.todo.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.api.todo.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_delete_todo(client, mocker):
    # 상태코드 204
    mocker.patch(
        "src.api.todo.get_todo_by_todo_id",        # mocking 적용할 함수 및 반환값 설정
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    mocker.patch(
        "src.api.todo.delete_todo",
        return_value = None,
    )
    response = client.delete("/todos/1")    # path에 하위 서브 path 적어주기
    assert response.status_code == 204      # status_code만 주고 response.json()는 assert 안 해봐도 됨! 반환값이 없을 것이기에
    
    # 상태코드 404
    mocker.patch(
        "src.api.todo.get_todo_by_todo_id",        # mocking 적용할 함수 및 반환값 설정
        return_value = None,
    )   
    response = client.delete("/todos/1")
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}
  • test_main.py 수정

 

# /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)

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

4 실습3 Refactoring - Repository Pattern 레포지토리 패  (0) 2024.08.08
4 실습2 Refactoring - Dependency Injection 의존성 주입  (0) 2024.08.08
3 실습8 테스트 코드 - DELETE API  (0) 2024.08.07
3 실습7 테스트 코드 - PATCH API  (0) 2024.08.05
3 실습6 테스트 코드 - POST API  (0) 2024.08.05

관련글

티스토리툴바