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

5 실습7 로그인 API 구현

by yororing 2024. 11. 16.

앞 단계 참조 링크:

00 개요

  • 로그인 처리하는 API 를 구현하는 실습 진행

01 로그인 API 로직 구현

0. 구현해야 할 기능 정리:

  • 1. request body로 username과 password 입력 받기
  • 2. DB에서 password 조회
  • 3. DB의 password (해싱됨)과 입력 받은 password (아직 해싱 안 됨)을 bcrypt.checkpw을 통해 검증 
  • 4. JWT 생성 (검증된 사용자에게 주기 위함)
  • 5. JWT 반환

1. schema의 requests.py에 내용 추가

  • 사용자로부터 받을 LogInRequest() 모델 추가
# /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

 

2. repository.py에 내용 추가

  • DB에서 사용자 정보를 조회하기 위해 UserRepository 클래스에 get_user_by_username라는 메소드 추가
    • → def get_user_by_username (self, username: str) -> User | None:
  • 반환값으로 password를 돌려줌
    • → return self.session.scalar(select(ToDo).where(User.username == username))
    • scalar를 사용하여, 입력 받은 User의 username이 DB의 user 테이블의 username과 동일한 조건 하에 해당 사용자의 데이터를 반환, 동일한 username이 없을 경우 None 반환 
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용

...

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

3. service/user.py에 내용 추가

  • DB에 저장된 hash화된 password와 입력 받은 plain한 password를 검증하는 로직 구현하는 verify_password() 함수 정의
    • → def verify_password(self, plain_password: str, hashed_password: str) -> bool:
  • 반환값으로 "UTF-8" 인코딩을 적용한 plain_password와 hashed_password를 bcrypt의 checkpw를 사용하여 일치 여부를 돌려줌
    • → return bcrypt.checkpw(plain_password.encode(self.encoding), hashed_password.encode(self.encoding))
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
import bcrypt
import logging

class UserService:
    encoding: str = "UTF-8"

...

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

 

4. python-jose 설치 및 service/user.py에 내용 더 추가

  • JWT 토큰을 생성하기 위해 python-jose라는 패키지 설치 및 jwt import 필요
    • → from jose import jwt
    • jose (JSON Web Signature and Encryption) 패키지는 JWT를 생성하는 기능을 제공 
# pip install python-jose
  • username을 string로 받아 string으로 JWT를 반환하는 create_jwt 메서드 정의
    • → def create_jwt(self, payload: dict) -> str:
    • JWT를 생성하기 위해선 secret key 필요
    • 터미널에서 다음 명령어를 실행하여 랜덤한 비밀키 (secret key)를 얻을 수 있음
      • a0ef6b27d5050d0040867160134b69bd673f640dd1db8db8b0ee0166114f4cfa
openssl rand -hex 32

  • 이 키를 복사하여 secret_key라는 속성으로 지정, 사용할 알고리즘도 지정, 이와 같이 jwt import, 또한 payload에서 토큰 유효 기간 설정하는 데에 사용될 datetime 및 timedelta로 import하기 
    • → secret_key: str = "a0ef6b27d5050d0040867160134b69bd673f640dd1db8db8b0ee0166114f4cfa"
    • 굳이 이렇게 생성 안하고 아무 문자 자리 넣어줘도 가능하긴 함!
    • 토큰 생성하는 데에 알고리즘도 필요, 보통 웹 토큰에서는 hs256 사용함
    • → jwt_algorithm: str = "hs256"
    • → from jose import jwt
    • → from datetime import datetime, timedelta
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
from datetime import datetime, timedelta  # 추가
from jose import jwt                      # 추가
import bcrypt
import logging

class UserService:
    encoding: str = "UTF-8"
    secret_key: str = "a0ef6b27d5050d0040867160134b69bd673f640dd1db8db8b0ee0166114f4cfa" # 추가
    jwt_algorithm: str = "hs256"	 # 추가
...
  • 반환값으로 jwt의 encode라는 메소드를 사용하여 payload, secret key, algorithm (hs256)을 인자로 전달해줘서 생성된  JWT 토큰 을 전달하기 
    • → return jwt.encode({"sub": username, "exp": datetime.now() + timedelta(days=1) }, self.secret_key, algorithm=self.jwt_algorithm)
    • hs256: Web Token 생성에 사용되는 알고리즘
    • "sub" (subject)라는 키로 username 전달: username에게 할당되는 토큰임을 의미, 사용자를 식별할 수 있는 식별자 역할 
    • "exp" (expired)라는 키에는 현재 시간 + 하루 (i.e., datetime.now() + timedelta(days=1)) 전달: 토큰의 유효기간을 현재 시간으로부터 하루임을 의미 
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
from datetime import datetime, timedelta        # 추가
from jose import jwy                            # 추가
import bcrypt
import logging

class UserService:
    encoding: str = "UTF-8"
    secret_key: str = "a0ef6b27d5050d0040867160134b69bd673f640dd1db8db8b0ee0166114f4cfa"    # 추가
    jwt_algorithm: str = "HS256"                                                            # 추가

...

    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
        )

5. schema/response.py에 내용 추가

  • jwt를 반환할 것으로서 response 객체 만들어주기: BaseModel을 상속받음 
    • → class JWTResponse(BaseModel):
  • access_token을 반환함
    • → access_token: str
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List

...

class JWTResponse(BaseModel):	# 추가
    access_token: str

6. api/user.py에 내용 추가

  • 포함해야 할 기능: 
    • 1. request body로 username과 password 입력 받기
    • 2. DB에서 username으로 데이터 (password) 조회
    • 3. DB의 user의 password (해싱됨)과 입력 받은 password (아직 해싱 안 됨)을 bcrypt.checkpw을 통해 검증 
    • 4. JWT 생성 (검증된 사용자에게 주기 위함)
    • 5. JWT 반환
  • router를 통해 post 메소드 매핑하여 log-in이라는 path로 API 생성
    • → @router.post("/log-in")
  • 기능 1: user_log_in_handler()라는 핸들러 생성, request로는 위에 생성한 LogInRequest를 받음 (기능 1), 사용자 정보를 가져오기 위해 UserRepository를 받음 (기능 2, 방금 위에서 정의한 get_user_by_username을 사용할 것), 입력 받은 password와 DB의 password를 검증하게 위한 기능이 있는 UserService를 받음 (기능 3, 방금 위에서 정의한 verify_password을 사용할 것)
    • → def user_log_in_handler(request: LogInRequest, user_repo: UserRepository = Depends(), user_service: UserService = Depends(), ):
  • 기능 2:  위에 정의한 get_user_by_username을 사용하여 user라는 변수에 사용자 정보 받아와서 (기능 2) user라는 변수에 저장 
    • → user: User | None = user_repo.get_user_by_username(username=request.username)
    • 사용자 정보를 못 받아왔을 경우, 즉, username이 DB에 등록되지 않았을 경우 HTTPException을 통해 강제 에러 발생
      • → raise HTTPException(status_code=404, detail = "User Not Found")
  • 기능 3: 위에 정의한 verify_password를 사용하여 입력 받은 password와 DB에 저장된 해시된 password를 비교하여 동일한지 검증하여 (기능 3) verified라는 변수에 저장
    • → verified: bool = user_service.verify_password(plain_password=request.password, hashed_password=user.password)
    • 인증에 실패했을 경우, 즉, 비밀번호가 다를 경우 HTTPException을 통해 강제 에러 발생
      • → raise HTTPException(status_code=401, detail = "Not Authorized")
      • 인증 실패했을 경우 일반적으로 401 반환함 
  • 기능 4: 위에 정의한 create_jwt를 사용하여 DB의 username을 인자로 넣어 JWT(토큰)를 생성 (기능 4) 하여 access_token 변수에 저장 
    • → access_token: str = user_service.create_jwt(username=user.username)

 

  • 기능 5: 위에 정의한 JWTResponse를 사용하여 JWT(토큰) 반환 (기능 5)
    • → return JWTResponse(access_token=access_token)
# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from src.schema.request import SignUpRequest, LogInRequest  # 추가
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

router = APIRouter(prefix="/users")

...

@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)
  • 코드 설명:
    • request를 통해 request body로 username과 password를 입력받음
    • user 변수에 실제 사용자의 username을 사용하여 DB로부터 해당 사용자의 정보를 저장
    • user 변수의 값이 없으면, 즉, DB에 username이 등록되지 않았다면 404 에러 강제 발생
    • user 변수에 값이 할당되었다면, 즉, DB에 등록된 사용자라면 다음을 실행
    • 실제 사용자가 입력한 password(request.password)가 DB에 등록된 password(user.password)와 맞는지 확인하여 verified에 값을 넣음 (맞으면 True, 틀리면 False가 verified 변수에 할당됨)
    • verified 변수가 False라면, 즉, DB에 등록된 password와 다르다면 401 에러 강제 발생
    • verified 변수가 True라면, 즉, DB에 등록된 password와 같다면 다음을 실행
    • access_token 변수에 JWT 값 생성하여 저장
    • 생성한 access_token(JWT) 반환
  • 그러면 이제 Client는 이 access_token을 갖고, 우리가 expired를 하루로 설정해놨기에 최대 하루 동안 로그인을 유지할 수 있음, 로그인 한 상태로 우리가 생성한 API를 계속 호출할 수 있음!

 

이때까지의 코드들: 13개

  • /c/Users/관리자/Desktop/projects/todos/src/main.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/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/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

│      └──  main.py

└── pyvenv.cfg

 

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

from database.repository import ToDoRepository, ToDo
from schema.response import ToDoSchema, ToDoListSchema
from schema.request import CreateToDoRequest

router = APIRouter(prefix='/todos')

# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)
def get_todos_handler(
    order: str | None = None,
    todo_repo: ToDoRepository = Depends(),
) -> ToDoListSchema:
    todos: List[ToDo] = todo_repo.get_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)
# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from src.schema.request import SignUpRequest, LogInRequest  # 추가
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

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)
# /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/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
from schema.request import CreateToDoRequest

Base = declarative_base()

# ToDo 클래스 모델링 한 것
class ToDo(Base):
    __tablename__ = 'todo'
    
    id = Column(Integer, primary_key=True, index=True)
    contents = Column(String(256), nullable=False)
    is_done = Column(Boolean, nullable=False)
    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 database.orm import ToDo, User
from 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(ToDo).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
# /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
# /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

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
        )
# /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
from src.database.repository import ToDoRepository

# 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 사용하여 단일 조회 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          # 201로 변경
    assert response.json() == {"id": 1, "username": username}     # 반환값으로 UserRepository.save_user 검증

 

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

5 실습9 JWT 사용 인증 구현 및 DB 데이터 조회  (0) 2024.11.26
5 실습8 로그인 API 테스트  (1) 2024.11.25
5 기능 고도화 - 로그인 / 유저 인증 - JWT  (0) 2024.11.16
5 실습6 회원가입 API 테스트  (2) 2024.11.16
5 실습5 회원가입 API 구현  (0) 2024.10.18

관련글

티스토리툴바