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

5 실습12 OTP 인증 API

by yororing 2025. 2. 13.

앞 단계 참조 링크:

00 개요

  • 생성된 OTP를 인증하는 실습 진행

01 OTP 인증

1. schema/request.py 수정

  • Request Body로 email 및 otp를 받을 VerifyOTPRequest 클래스 추가
...
class VerifyOTPRequest(BaseModel):    # 추가
    email: str
    otp: int

2. api/users.py 수정

  • verify_otp_handler() 추가
  • 기능:
    • 1. access_token 사용 (즉, 회원가입 한, 인증된 사용자만 요청 가능)
    • 2. request body로 email, otp 받음
    • 3. request로 받은 otp와 redis.get(email)로 받은 otp 비교하여 이메일 인증
    • 4. 인증 완료 시 user에 email 저장 (실습에서는 구현 안 하고 return 값에서 user가 잘 조회되는지 확인만 할 것)
# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from src.schema.request import SignUpRequest, LogInRequest, CreateOTPRequest, VerifyOTPRequest  # 추가
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('/email/otp/verify')          # 추가
def verify_otp_handler(
    request: VerifyOTPRequest,                      # request body로 email, otp 받음 (request가 get_access_token보다 위로 와야 에러 안 남)
    access_token: str = Depends(get_access_token),   # access_token 사용 (즉, 회원가입 한, 인증된 사용자만 요청 가능)
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends(),
    ):

    # request로 받은 otp와 redis.get(email)로 받은 otp 비교하여 이메일 인증
    otp: str | None = redis_client.get(request.email)   # key가 잘못됬거나 만료 시 None 반환
    if not otp:
        raise HTTPException(status_code=400, detail='Bad Request')
    if request.otp != int(otp):
        raise HTTPException(status_code=400, detail='Bad Request')
    username: str = user_service.decode_kwt(access_token=access_token)
    user: User | None = user_repo.get_user_by_username(username)
    if not user:
        raise HTTPException(status_code=404, detail='User Not Found')
    return UserSchema.from_orm(user)                # User가 잘 조회되는지 확인 (user에 email 저장하는 실습은 아직 진행 안 함)

 

 

02 Swagger UI에서 확인

1. Docker 실행 (Docker Desktop, 터미널)

  • Docker Desktop 열어서 todos 컨테이너 실행 (세모 버튼 클릭)
  • 터미널에서 redis 컨테이너 실행 (입력)
# docker run -p 6379:6379 --name redis -d --rm redis
  • 설명:
    • 참조: 2025.01.24 - [Web 개발/FAST API (인프런 강의 내용)] - 5 실습10 Redis 컨테이너 실행 & Redis 연결 (docker)
    • -p 로컬호스트port번호:docker컨테이너port번호
      • : 포트를 매핑해주는 옵션
      • Redis는 기본적으로 6379번 포트 사용
      • -p 옵션으로 6379:6379을 매핑 해주면 로컬 호스트의 6379번 포트와 docker 컨테이너 안의 6379번 포트가 연결이 됨
      • 그래서 우리가 6379번 포트로 어떤 데이터베이스 명령을 요청하게 되면 그게 docker 안까지 전달이 되어 Redis에  적용됨
    • -d
      • : 'detatch' 옵션
      • 컨테이너가 백그라운드에서 동작하게 하도록 하는 옵션
    • --rm
      • : 'remove' 옵션, redis 컨테이너가 종료되었을 때 컨테이너를 자동으로 삭제하는 옵션
      • 이 옵션은 테스트나 단발성 작업에 유용, 불필요하게 많은 컨테이너가 docker 시스템에 남지 않도록 관리하는 데 도움을 줌
    • --name docker컨테이너명
      • : 동작시킬 docker 컨테이너의 이름을 지정해주는 옵션
      • 동작시킬 docker 컨테이너의 이름을 redis라고 지정
    • redis (맨 마지막)
      • : 컨테이너를 동작시킬 때 사용할 docker 이미지 지정해주는 옵션
      • 원래는 이미지명:버전 (예, mysql:8.0) 이렇게 작성하는데, 뒤에 버전을 명시하지 않을 경우 가장 최신 버전을 자동으로 사용
  • todos 및 redis 컨테이너 실행 중인지 확인
# docker ps

CONTAINER ID   IMAGE       COMMAND                  CREATED              STATUS              PORTS                               NAMES
5ef94a21b717   redis       "docker-entrypoint.s…"   About a minute ago   Up About a minute   0.0.0.0:6379->6379/tcp              redis
bde6f7c5824c   mysql:8.0   "docker-entrypoint.s…"   3 weeks ago          Up 23 minutes       0.0.0.0:3306->3306/tcp, 33060/tcp   todos

2. uvicorn 으로 app 실행 (터미널)

  • 터미널에서 다음을 실행
  • 경로: todos에서 실행
# uvicorn src.main:app --reload --port 8000

3. 브라우저에 127.0.0.1:8000/docs 입력

 

4. 로그인 해서 access token 받기

  • 전에 만들어놓은 username: admin, password: password 계정으로 로그인 시도
  • POST: /users/log-in > Try it out > Request body: {"username": "admin", "password": "password"} 입력 > Execute 클릭

  • 정상적으로 로그인 된 것 확인 (Response: 200)

  • access_token 복사

5. access_token으로 Authorize 수행

{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTczOTU2MjUwM30.CdXpFvO1rnZvyC0XVs2CCI54T54S6F0U8yS6Lnp3U-Y"
}
  • 위로 올라가 Authorize 클릭

  • HTTPBearer (http, Bearer)의 Value에 복사한 access_token 입력 > Authorize 클릭

  • 정상적으로 승인된 것을 확인 > Close

 

6. OTP 생성 (발급) API 검증하기

  • POST: /users/email/otp > Try it out > Request body: {"email": "admin@fastapi.com"} 입력 > Execute 클릭
    • "admin@fastapi.com" 는 아무 이메일 입력해도 됨

  • 정상적으로 OTP (랜덤한 4자리 숫자)가 생성된 것을 확인 (i.e., 9796)

 

7. OTP 인증 API 검증하기

  • POST: /users/email/otp/verify > Try it out > Request body: {"email": "admin@fastapi.com", "otp": 9796} 입력 > Execute 클릭

  • 정상적으로 인증된 것 확인 (Response: 200)

  • 정상적으로 생성된 OTP 외에 다른 번호를 입력할 경우 (예, 'otp': 1111) 요청이 잘못 되었다는 오류 (i.e., 400 Bad Request) 발생 확인

 

 

 

이때까지의 코드들: 15개

  • /c/Users/관리자/Desktop/projects/todos/src/cache.py
    • Note:
      • 강의에서는 수정하지 않았지만, 개인 컴퓨터로 실습 중 오류 발생 (unexpected argument: decode_response) 해서 해당 argument 지우고 실습 진행함
      • 강의에서 사용한 redis 버전과 내 개인 컴퓨터에 다운로드한 redis 버전이 달라서 생기는 오류로 추정
      • decode_response 인자 값을 안 줘도 강의 실습 진행하는 데 문제 없었음
  • /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 (수정 - redis 버전 문제)

│     ├──  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 수정 - 강의에서는 수정하지 않았지만, 개인 실습 중 redis 버전 문제로 수정
# /c/Users/관리자/Desktop/projects/todos/src/cache.py 내용
import redis

# 아래와 같이 할 경우 다음 에러 발생: TypeError: Redis.__init__() got an unexpected keyword argument 'decode_response'
# redis_client = redis.Redis(
#     host='127.0.0.1', port=6379, db=0, encoding='UTF-8', decode_response=True
# )

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

 

# /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, VerifyOTPRequest  # 추가
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에 전송하는 로직을 실습에선 간략하게 이렇게만 구현할 것

@router.post('/email/otp/verify')           ######## 추가
def verify_otp_handler(
    request: VerifyOTPRequest,                      # request body로 email, otp 받음 (request가 get_access_token보다 위로 와야 에러 안 남)
    access_token: str = Depends(get_access_token),   # access_token 사용 (즉, 회원가입 한, 인증된 사용자만 요청 가능)
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends(),
    ):

    # request로 받은 otp와 redis.get(email)로 받은 otp 비교하여 이메일 인증
    otp: str | None = redis_client.get(request.email)   # key가 잘못됬거나 만료 시 None 반환
    if not otp:
        raise HTTPException(status_code=400, detail='Bad Request')
    if request.otp != int(otp):
        raise HTTPException(status_code=400, detail='Bad Request')
    username: str = user_service.decode_jwt(access_token=access_token)
    user: User | None = user_repo.get_user_by_username(username)
    if not user:
        raise HTTPException(status_code=404, detail='User Not Found')
    return UserSchema.from_orm(user)                # User가 잘 조회되는지 확인 (user에 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 CreateOTPRequest(BaseModel):
    email: str

class VerifyOTPRequest(BaseModel):  # 추가
    email: str
    otp: int

 

 

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

 

 

# /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 실습11 OTP 생성 API  (0) 2025.02.10
5 실습10 Redis 컨테이너 실행 & Redis 연결 (docker)  (0) 2025.01.24
5 기능 고도화 Caching  (1) 2024.12.25
5 실습9 JWT 사용 인증 구현 및 DB 데이터 조회  (0) 2024.11.26

관련글

티스토리툴바