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

5 실습11 OTP 생성 API

by yororing 2025. 2. 10.

앞 단계 참조 링크:

00 개요

  • 회원가입 API에 OTP (One-Time Password) API를 추가할 것
  • 현재는 사용자 입력 (username, password)으로 회원가입 및 로그인 기능 존재
  • 앞으로는 회원가입 후, 사용자 입력 (email)으로 이메일을 받아 인증 (OTP 적용)을 거친 후 DB에 저장할 것
  • 추후 저장된 이메일로 알림 전송 기능이 가능할 것
  • 계획:
    • POST API
      • OTP 생성: /users/email/otp
        • key: email, value: 1234, exp: 3 min 값들을 받아 OTP 생성
      • 이메일 인증 및 DB 저장: /users/email/otp/verify
        • email, OTP 값들을 받아 이메일 인증 후 USER 테이블에 email 저장

01 OTP 생성

1. schema/request.py 수정

  • Request Body로 email을 받을 CreateOTPRequest 클래스 추가
...
class CreateOTPReqeust(BaseModel):
    email: str

2. service.user.py 수정

  • OTP 생성하기 위해 create_otp(self) 함수 추가
    • 파이썬 random 모듈 사용하여 OTP 생성
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
from datetime import datetime, timedelta
from jose import jwt
import bcrypt
import logging
import random				# 추가

class UserService:
...
    @staticmethod
    def create_otp() -> int:		# 추가
        return random.randint(1000, 9999)	# 추가
  • 설명:
    • @staticmethod:
      • 인스턴스 속성을 사용할 필요가 없기 때문에 static 메소드 적용, 이로 인해 아랫줄 코드에 'def create_otp(self)' 안에 self 작성 불필요해짐 → def create_otp() -> int: 로 충분함
    • def create_otp() -> int: 
      • int를 반환하는 함수 정의
    • return random.randint(1000, 9999):
      • random 모듈의 randint 함수를 사용하여 1000부터 9999사이의 랜덤한 숫자 반환

3. src 디렉토리에 cache.py 파일 추가

  • redis client 객체 추가
# /c/Users/관리자/Desktop/projects/todos/src/cache.py 내용
import redis

redis_client = redis.Redis(
    host='127.0.0.1', port=6379, db=0, encoding='UTF-8', decode_response=True
)

# note: redis 5.2.1 버전에서는 decode_response 인자가 unexpected argument ERROR 발생시켜서 제거했음
# 다음 글의 소스코드 참조: 5 실습12 OTP 인증 API - cache.py

4. api/users.py 수정

  • create_otp_handler() 추가
  • 기능:
    • 1. access_token 사용 (즉, 회원가입 한, 인증된 사용자만 이메일 인증 가능)
    • 2. request body로 email 받음
    • 3. OTP 생성 (랜덤한 4자리 숫자)
    • 4. redis에 저장 (key: email, value: OTP, exp=3min)
    • 5. OTP를 email에 전송 
      • 실습에서는 email에 직접 전송하지 않고 return 값으로 otp를 반환할 것
# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from src.schema.request import SignUpRequest, LogInRequest, CreateOTPRequest # 추가
from src.schema.response import UserSchema, JWTResponse
from src.service.user import UserService
from src.database.orm import User
from src.database.repository import UserRepository
from src.security import get_access_token       # 추가
from src.cache import redis_client              # 추가

...

@router.post('/email/otp')                      # 추가
def create_otp_handler(
    request: CreateOTPRequest,                  # request body로 email 받음 (request가 get_access_token보다 위로 와야 에러 안 남)
    _: str = Depends(get_access_token),         # access_token 사용 (즉, 회원가입 한, 인증된 사용자만 이메일 인증 가능)
                                                    # 헤더에 access_token이 있어야지만 여기를 통과하지만, access_token을 여기 로직에서 사용할 것은 아니라서 지금은 _로 줄 것
                                                    # 즉, 헤더에 있는지 검증만 하고 실제로 이 값을 사용하지 않을 것
    user_service: UserService = Depends()
    ):
    otp: int = user_service.create_otp()        # OTP 생성 (랜덤한 4자리 숫자)
    redis_client.set(request.email, otp)        # redis에 저장 (key: email, value: OTP, exp=3min)
    redis_client.expire(request.email, 3 * 60)  # 초 단위로 전달해야됨
    return {'otp': otp}                         # OTP를 email에 전송하는 로직을 실습에선 간략하게 이렇게만 구현할 것

 

 

이때까지의 코드들: 15개

  • /c/Users/관리자/Desktop/projects/todos/src/cache.py (추가)
  • /c/Users/관리자/Desktop/projects/todos/src/main.py
  • /c/Users/관리자/Desktop/projects/todos/src/security.py
  • /c/Users/관리자/Desktop/projects/todos/src/api/todo.py
  • /c/Users/관리자/Desktop/projects/todos/src/api/user.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/connection.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/orm.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/service/user.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_todos_api.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_users_api.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py

 

이때까지의 디렉토리

TODOS

├── Include

├── Lib

├── Scripts

├── src

│     ├── api

│     │      ├── todo.py

│     │      └── user.py (수정)

│     ├── database

│     │      ├── connection.py

│     │      ├── orm.py

│     │      └── repository.py

│     ├── schema

│     │      ├── request.py (수정)

│     │      └── response.py

│     ├── service

│     │      └── user.py (수정)

│     ├── tests

│     │      ├── __init__.py

│     │      ├── conftest.py

│     │      ├── test_main.py

│     │      ├── test_todos_api.py

│     │      └── test_users_api.py

│     ├──  cache.py (추가)

│     ├──  main.py

│     └──  security.py

└── pyvenv.cfg

 

현재 DB 상태

mysql> select * from todo;
+----+-------------------+---------+---------+
| id | contents          | is_done | user_id |
+----+-------------------+---------+---------+
|  1 | FastAPI Section 0 |       1 |       1 |
|  2 | FastAPI Section 1 |       1 |       1 |
|  3 | FastAPI Section 2 |       0 |       1 |
|  4 | FastAPI Section 0 |       1 |       2 |
|  5 | FastAPI Section 1 |       0 |       2 |
|  6 | Algorithm 0       |       1 |       3 |
+----+-------------------+---------+---------+

mysql> select * from user;
+----+----------+--------------------------------------------------------------+
| id | username | password                                                     |
+----+----------+--------------------------------------------------------------+
|  1 | admin    | $2b$12$3B0G59vaOdMYD2fGS5O0Yeu60ElzAacYRpvf8uShlcYw7oo2OfQau | -> 'password' 해시
|  2 | pearl    | nothasedpassword                                             |
|  3 | dayoonz  | $2b$12$dZqBc4.mOJEZDyxD2z4YUe1zMN6702Rq.1meBcfiQr1x238Svc5CO | -> 'helloworld' 해시
+----+----------+--------------------------------------------------------------+

 

소스 코드

  • cache.py 추가
# /c/Users/관리자/Desktop/projects/todos/src/cache.py 내용
import redis

redis_client = redis.Redis(
    host='127.0.0.1', port=6379, db=0, encoding='UTF-8', decode_response=True
)

# note: redis 5.2.1 버전에서는 decode_response 인자가 unexpected argument ERROR 발생시켜서 제거했음
# 다음 글의 소스코드 참조: 5 실습12 OTP 인증 API - cache.py

 

# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI
from src.api import todo, user

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

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

 

# /c/Users/관리자/Desktop/projects/todos/src/security.py 내용
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

def get_access_token(
    auth_header: HTTPAuthorizationCredentials | None = Depends(HTTPBearer(auto_error=False)),
) -> str:
    if auth_header is None:
        raise HTTPException(
            status_code=401,
            detail='Not Authorized',
        )
    return auth_header.credentials

 

# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from typing import List	

from src.database.repository import ToDoRepository, ToDo, UserRepository, User
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest
from src.security import get_access_token
from src.service.user import UserService

router = APIRouter(prefix='/todos')

# GET Method 사용하여 username의 todo 조회 API          # 변경 (원래: GET Method 사용하여 전체 조회 API)
@router.get("", status_code=200)
def get_todos_handler(
    access_token: str = Depends(get_access_token),
    order: str | None = None,
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends(),
) -> ToDoListSchema:
    username: str = user_service.decode_jwt(access_token=access_token)      # 이 username을 통해 사용자 조회
    user: User | None = user_repo.get_user_by_username(username=username)   # 이 user_id를 통해 User(테이블)를 불러와 User에 담긴 ToDo 조회
    if not user: # 회원 탈퇴, etc. 경우
        raise HTTPException(status_code=404, detail='User Not Found')
    todos: List[ToDo] = user.todos                      # User에 있는 ToDo들을 todos에 담음
    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)
def get_todo_handler(
    todo_id: int,
    todo_repo: ToDoRepository = Depends(),
    ) -> ToDoSchema:
    todo:ToDo | None = todo_repo.get_todo_by_todo_id(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)
def create_todo_handler(
    request: CreateToDoRequest,
    todo_repo: ToDoRepository = Depends(),
    ) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)          # id=None
    todo: ToDo = todo_repo.create_todo(todo=todo)      # id=int
    return ToDoSchema.from_orm(todo)

# PATCH Method 사용하여 is_done 값 수정 API
@router.patch("/{todo_id}", status_code=200)
def update_todo_handler(
    todo_id: int,
    is_done: bool = Body(..., embed=True),
    todo_repo: ToDoRepository = Depends(),
    ):
    todo:ToDo | None = todo_repo.get_todo_by_todo_id(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 = todo_repo.update_todo(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)
def delete_todo_handler(
    todo_id: int,
    todo_repo: ToDoRepository = Depends(),
    ):
    todo:ToDo | None = todo_repo.get_todo_by_todo_id(todo_id = todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="ToDo Not Found")
    todo_repo.delete_todo(todo_id=todo_id)

 

  • api/user.py 수정
# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from src.schema.request import SignUpRequest, LogInRequest, CreateOTPRequest # 추가
from src.schema.response import UserSchema, JWTResponse
from src.service.user import UserService
from src.database.orm import User
from src.database.repository import UserRepository
from src.security import get_access_token       # 추가
from src.cache import redis_client              # 추가

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler(
    request: SignUpRequest,
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends()
    ):
    # 해싱된 비밀번호 생성
    hashed_password: str = user_service.hash_password(
        plain_password = request.password
    )
    # 사용자 생성
    user: User = User.create(
        username = request.username,
        hashed_password = hashed_password
    )
    # 사용자를 DB에 저장
    user: User = user_repo.save_user(user=user) # 이 시점에서 사용자의 id 값 = int로 들어감, 실제 DB에 사용자가 생성됨

    # user(id, username) 값 반환
    return UserSchema.from_orm(user)

@router.post("/log-in")
def user_log_in_handler(
    request: LogInRequest,
    user_repo: UserRepository = Depends(),
    user_service: UserService = Depends(),
    ):
    # 사용자 정보 DB 조회
    user: User | None = user_repo.get_user_by_username(username=request.username)
    if not user:
        raise HTTPException(status_code=404, detail = "User Not Found")
    
    # 비밀번호 검증
    verified: bool = user_service.verify_password(
        plain_password=request.password,
        hashed_password=user.password
    )
    if not verified:
        raise HTTPException(status_code=401, detail = "Not Authorized")
    
    # jwt 생성
    access_token: str = user_service.create_jwt(username=user.username)
    
    # jwt 반환
    return JWTResponse(access_token=access_token)

@router.post('/email/otp')                      # 추가
def create_otp_handler(
    request: CreateOTPRequest,                  # request body로 email 받음 (request가 get_access_token보다 위로 와야 에러 안 남)
    _: str = Depends(get_access_token),         # access_token 사용 (즉, 회원가입 한, 인증된 사용자만 이메일 인증 가능)
                                                    # 헤더에 access_token이 있어야지만 여기를 통과하지만, access_token을 여기 로직에서 사용할 것은 아니라서 지금은 _로 줄 것
                                                    # 즉, 헤더에 있는지 검증만 하고 실제로 이 값을 사용하지 않을 것
    user_service: UserService = Depends()
    ):
    otp: int = user_service.create_otp()        # OTP 생성 (랜덤한 4자리 숫자)
    redis_client.set(request.email, otp)        # redis에 저장 (key: email, value: OTP, exp=3min)
    redis_client.expire(request.email, 3 * 60)  # 초 단위로 전달해야됨
    return {'otp': otp}                         # OTP를 email에 전송하는 로직을 실습에선 간략하게 이렇게만 구현할 것

 

 

# /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,         # 제거: DB에서 쿼리가 계속 발생하면 로그를 보기 힘들기 때문
                       )
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/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
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)
    user_id = Column(Integer, ForeignKey("user.id"))
    
    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

# User 클래스 모델링 한 것
class User(Base):
    __tablename__ = 'user'
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(256), unique=True, nullable=False) # unique=True는 내가 넣은 것임
    password = Column(String(256), nullable=False)
    todos = relationship("ToDo", lazy="joined")
    
    @classmethod
    def create(cls, username: str, hashed_password: str) -> "User":
        return cls(
            username=username,
            password=hashed_password,
        )

 

 

# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from typing import List
from fastapi import Depends

from src.database.orm import ToDo, User
from src.database.connection import get_db

class ToDoRepository:
    def __init__(self, session:Session = Depends(get_db)):
        self.session = session 
        
    def get_todos(self,) -> List[ToDo]:
        return list(self.session.scalars(select(ToDo)))

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

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

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

    def delete_todo(self, todo_id: ToDo) -> None:
        self.session.execute(delete(ToDo).where(ToDo.id == todo_id))
        self.session.commit()

class UserRepository:
    def __init__(self, session: Session = Depends(get_db)):
        self.session = session
    
    def get_user_by_username(self, username: str) -> User | None:
        return self.session.scalar(select(User).where(User.username == username))
    
    def save_user(self, user: User) -> User:
        self.session.add(instance=user)
        self.session.commit()
        self.session.refresh(instance=user)
        return user

 

 

  • request.py 수정
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용
from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
    contents: str
    is_done: bool
    
class SignUpRequest(BaseModel):
    username: str
    password: str

class LogInRequest(BaseModel):
    username: str
    password: str

class CreateOTPReqeust(BaseModel):  # 추가
    email: str

 

 

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

class UserSchema(BaseModel):
    id: int
    username: str
    
    class Config:
        from_attributes = True
        # orm_mode = True # 이거 추가 해야지 ConfigError 안남

class JWTResponse(BaseModel):
    access_token: str

 

 

  • service/user.py 수정
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
from datetime import datetime, timedelta
from jose import jwt
import bcrypt
import logging
import random           # 추가

class UserService:
    encoding: str = "UTF-8"
    secret_key: str = "a0ef6b27d5050d0040867160134b69bd673f640dd1db8db8b0ee0166114f4cfa"
    jwt_algorithm: str = "HS256"
    
    def hash_password(self, plain_password: str) -> str:
        '''
        input: plain_password (str)
        output: hashed_password (str)
        function: receives plain-text password and returns hashed password
        '''
        logging.info("hash_password called with: %s", plain_password)
        hashed_password: bytes = bcrypt.hashpw(
            plain_password.encode(self.encoding),
            salt=bcrypt.gensalt()
        )
        return hashed_password.decode(self.encoding)
    
    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        '''
        input: plain_password (str)
        output: hashed_password (str)
        function: receives plain-text password and hashed password, returns bool whether they are the same password or not
        '''
        return bcrypt.checkpw(
            plain_password.encode(self.encoding),
            hashed_password.encode(self.encoding)
        )
        
    def create_jwt(self, username: dict) -> str:
        return jwt.encode(
            {"sub": username,
             "exp": datetime.now() + timedelta(days=1) },
            self.secret_key,
            algorithm=self.jwt_algorithm
        )
    
    def decode_jwt(self, access_token: str) -> str:
        payload: dict = jwt.decode(
            access_token, self.secret_key, algorithms=[self.jwt_algorithm]
        )
        # expire: 원래는 여기서 토큰의 만료 기간을 검증하는 부분 필요하나 우리는 거기까지 안 할 것임
        return payload['sub'] # username
    
    @staticmethod                                   # 추가
	def create_otp() -> int:                    # 추가
        return random.randint(1000, 9999)           # 추가

 

 

 

# /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py 내용
import pytest
from fastapi.testclient import TestClient
from main import app

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

 

 

# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용

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

 

 

# /c/Users/관리자/Desktop/projects/todos/src/tests/test_todos_api.py 내용
from src.schema.response import ToDoSchema
from src.database.orm import ToDo, User 
from src.database.repository import ToDoRepository, UserRepository
from src.service.user import UserService

# GET Method 사용하여 전체 조회 API 검증
# def test_get_todos(client, mocker):
#     # order = ASC
#     mocker.patch.object(ToDoRepository, "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 사용하여 username의 todo 목록 조회 API 검증
def test_get_todos(client, mocker):
    access_token: str = UserService().create_jwt(username="test")
    headers = {'Authorization': f"Bearer {access_token}"}
    user = User(id=1, username="test", password="hashed")
    user.todos = [
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDo(id=2, contents="FastAPI Section 1", is_done=False),
    ]
    mocker.patch.object(
        UserRepository, "get_user_by_username", return_value=user
    )
    # order = ASC
    response = client.get('/todos', header=headers)
    assert response.status_code == 200

# GET Method 사용하여 단일 조회 API 검증
def test_get_todo(client, mocker):
    # 상태코드 200
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id"
        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.object(ToDoRepository, "get_todo_by_todo_id", 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.object(ToDoRepository, "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.object(ToDoRepository, "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.object(ToDoRepository, "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.object(ToDoRepository, "get_todo_by_todo_id",
        return_value = None,
    )   
    response = client.patch("/todos/1", json={"is_done": True})
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}
  
# DELTE Method API 검증  
def test_delete_todo(client, mocker):
    # 상태코드 204
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id",
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    mocker.patch.object(ToDoRepository, "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.object(ToDoRepository, "get_todo_by_todo_id",return_value = None,) 
    response = client.delete("/todos/1")
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}

 

 

# /c/Users/관리자/Desktop/projects/todos/src/tests/test_users_api.py 내용
from src.service.user import UserService
from src.database.orm import User
from src.database.repository import UserRepository

username = 'test11'
def test_user_sign_up(client, mocker):                  # mocker는 fixture
    hashed_password = mocker.patch.object(              # UserService.hash_password 메소드에 mocking 적용
        UserService,
        "hash_password",
        return_value="hashed"
    )
    
    user_create = mocker.patch.object(                  # User.create 메소드에 mocking 적용
        User,
        "create",
        return_value=User(id=None, username=username, password="hashed")
    )
    
    mocker.patch.object(                                # UserRepository.save_user 메소드에 mocking 적용
        UserRepository,
        "save_user",
        return_value=User(id=1, username=username, password="hashed")
    )
    
    body = {                                            # API에 넘겨줄 body 추가
        "username": username,
        "password": "plain"
    }
    response = client.post("users/sign-up", json=body)  # json 형식으로 body 넘겨줘서 POST 요청 보내기 추가
    print(hashed_password.call_args_list)
    hashed_password.assert_called_once_with(            # UserService.hash_password 호출 여부 검증
        plain_password="plain"
    )

    user_create.assert_called_once_with(                # User.create 호출 여부 검증
        username=username, hashed_password="hashed"
    )
    
    assert response.status_code == 201
    assert response.json() == {"id": 1, "username": username}     # 반환값으로 UserRepository.save_user 검증

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

5 실습13 회원가입 알림 - Background Task  (0) 2025.02.17
5 실습12 OTP 인증 API  (0) 2025.02.13
5 실습10 Redis 컨테이너 실행 & Redis 연결 (docker)  (0) 2025.01.24
5 기능 고도화 Caching  (1) 2024.12.25
5 실습9 JWT 사용 인증 구현 및 DB 데이터 조회  (0) 2024.11.26

관련글

티스토리툴바