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

5 실습10 Redis 컨테이너 실행 & Redis 연결 (docker)

by yororing 2025. 1. 24.

앞 단계 참조 링크:

 

00 개요

  • Caching(캐싱)을 실습하기 위해 docker를 사용하여 Redis 컨테이너를 띄우는 작업을 실습할 것

1. Redis란 (Remote Dictionary Server)

 

01 Redis 띄우기 (docker 사용)

0. docker 사용 방법

1. docker 컨테이너 상태/정보 확인

$ docker container ls [OPTIONS]

# 또는
$ docker container list
$ docker container ps
$ docker ps         # 실행 중인 컨테이너 확인
$ docker ps
CONTAINER ID   IMAGE       COMMAND                  CREATED          STATUS          PORTS                               NAMES
bde6f7c5824c   mysql:8.0   "docker-entrypoint.s…"   2 hours ago      Up 2 hours      0.0.0.0:3306->3306/tcp, 33060/tcp   todos
  • docker ps 입력 시 bde6f7로 시작하는 컨테이너가 mysql:8.0 이미지를 이용하여 NAME=todos라는 이름으로 동작 중임을 확인
  • 우리는 docker에 Redis 컨테이너를 생성 후 총 2개 (mysql, Redis)의 컨테이너를 사용해서 todo 앱을 실행할 것

2. Redis 컨테이너 띄우기

$ docker run -p 6379:6379 --name redis -d --rm redis

1) 옵션 설명

  • -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) 이렇게 작성하는데, 뒤에 버전을 명시하지 않을 경우 가장 최신 버전을 자동으로 사용

2) 위 명령어 실행 시 설명

  • 위의 명령어 실행 시 현재 redis 이미지가 없기에 우선 docker hub에서 이미지를 다운로드함
$ docker run -p 6379:6379 --name redis -d --rm redis
Unable to find image 'redis:latest' locally
latest: Pulling from library/redis
7ce705000c39: Pull complete 
cc82b694f012: Pull complete 
b95871a2f28d: Pull complete 
194ef0d9d22c: Pull complete 
e99cf8129078: Pull complete 
0505c4e376ef: Pull complete 
4f4fb700ef54: Pull complete 
8829b28bc638: Pull complete 
Digest: sha256:ca65ea36ae16e709b0f1c7534bc7e5b5ac2e5bb3c97236e4fec00e3625eb678d
Status: Downloaded newer image for redis:latest
345de51abc6672562d82c924ea5cbb13fa5a1e4acf98ab8c601f2c5e7c4b200c
  • 이미지 다운로드 완료 시 방금 만든 redis 라는 docker 컨테이너를 동작시킴
  • 마지막으로 출력되는 id (여기선 '345de51abc6672562d82c924ea5cbb13fa5a1e4acf98ab8c601f2c5e7c4b200c') 가 로컬 컴퓨터에서 동작하는 docker 컨테이너의 id임

3. docker 컨테이너 상태/정보 확인

$ docker ps
CONTAINER ID   IMAGE       COMMAND                  CREATED          STATUS          PORTS                               NAMES
345de51abc66   redis       "docker-entrypoint.s…"   10 minutes ago   Up 10 minutes   0.0.0.0:6379->6379/tcp              redis
bde6f7c5824c   mysql:8.0   "docker-entrypoint.s…"   2 hours ago      Up 2 hours      0.0.0.0:3306->3306/tcp, 33060/tcp   todos
  • docker ps 입력 시 345de51로 시작하는 컨테이너가 redis 이미지를 이용하여 NAME=redis라는 이름으로 추가되어 동작 중임을 확인

4. docker 컨테이너 로그 확인

$ docker container logs [OPTIONS] CONTAINER

# 또는
$ docker logs
$ docker logs redis
1:C 23 Jan 2025 08:57:56.292 * oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
1:C 23 Jan 2025 08:57:56.292 * Redis version=7.4.2, bits=64, commit=00000000, modified=0, pid=1, just started
1:C 23 Jan 2025 08:57:56.292 # Warning: no config file specified, using the default config. In order to specify a config file use redis-server /path/to/redis.conf
1:M 23 Jan 2025 08:57:56.294 * monotonic clock: POSIX clock_gettime
1:M 23 Jan 2025 08:57:56.297 * Running mode=standalone, port=6379.
1:M 23 Jan 2025 08:57:56.298 * Server initialized
1:M 23 Jan 2025 08:57:56.298 * Ready to accept connections tcp
  • 맨 마지막 줄에 "Ready to accept connections tcp"가 출력됨, 즉, redis가 정상적으로 우리의 커넥션을 받을 수 있는 상태임을 확인
  • 우리는 여기서 docker를 통해 redis에 직접 들어가 보는 대신 python 라이브러리를 통해 redis 명령어를 직접 내려볼 것

02 Redis 사용하기 (python 콘솔에서 진행)

0. redis 설치하기

  • redis 라이브러리가 설치되지 않았을 경우 설치해주기
  • 가상환경 (# source Scripts/activate - windows, # source venv/bin/activate - macOS)을 실행시킨 후 진행
$ pip3 install --upgrade redis
Collecting redis
  Downloading redis-5.2.1-py3-none-any.whl.metadata (9.1 kB)
Downloading redis-5.2.1-py3-none-any.whl (261 kB)
Installing collected packages: redis
Successfully installed redis-5.2.1

1. python 콘솔 열기

  • 가상환경 (# source Scripts/activate - windows, # source venv/bin/activate - macOS)을 실행시킨 후 다음 명령어를 터미널에 입력하여 python 콘솔 열기
$ python3     
Python 3.12.3 (v3.12.3:f6650f9ad7, Apr  9 2024, 08:18:47) [Clang 13.0.0 (clang-1300.0.29.30)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>>

2. redis import하기

>>> import redis

3. redis client 생성

  • redis를 사용하기 위해 client 생성
>>> redis_client = redis.Redis(host='127.0.0.1', 
                               port=6379, 
                               db=0, 
                               encoding='UTF-8',
                               decode_responses=True)

 

  • redis.Redis(host='127.0.0.1', port=6379, db=0)
    • host: 호스트로 127.0.0.1 (로컬호스트)을 연결
    • port: 6379번 사용
    • db: 데이터베이스의 이름을 0으로 지정, 0부터 시작해서 사용할 것
    • encoding: 인코딩 지정
    • decode_responses: redis는 원래 byte 형태로 데이터 저장하는데, 인코딩을 지정 후 True 옵션을 줄 경우 redis에서 데이터를 가져올 때 byte 대신 python에서 제공하는 데이터 타입으로 자동변환 돼서 쉽게 사용 가능

4. redis client에 키-값 저장

>>> redis_client.set('key', 'value')    # 'key' 키에 해당하는 'value' 값 저장
True

5. redis client에서 키로 값 조회

>>> redis_client.get('key')    # 'key'키로 해당하는 값 불러오기
'value'

6. redis client에 만료 기능 설정

>>> redis_client.expire('key', 10)   # 'key'키를 10초 간만 유지시키는 기능
True
>>> redis_client.get('key')        # 바로 'key'키로 값 조회 시 'value'값이 출력됨
'value'
>>> redis_client.get('key')        # 10초 후 값 조회 시 미출력 (만료됨)
>>>

  • 우리는 이 만료 기능을 사용할 것:
    • key를 이메일로, value를 무작위의 4자리 숫자 (OTP, One-Time Password)로 설정하여 여기에 만료시간 (예, 5분) 설정하여 redis에 저장
    • 해당 이메일을 소유한 사용자가 만료시간 (5분) 안에 인증을 하지 않을 경우 자동으로 만료되는 형태로 OTP 기능을 구현할 것

이때까지의 코드들: 14개

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

│     └──  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' 해시
+----+----------+--------------------------------------------------------------+

 

 

코드

# /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(),
    # todo_repo: ToDoRepository = 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] = todo_repo.get_todos()         # 삭제
    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)
# /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,         # 제거: 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
# /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
        )
    
    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
# /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          # 201로 변경
    assert response.json() == {"id": 1, "username": username}     # 반환값으로 UserRepository.save_user 검증

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

5 실습12 OTP 인증 API  (0) 2025.02.13
5 실습11 OTP 생성 API  (0) 2025.02.10
5 기능 고도화 Caching  (1) 2024.12.25
5 실습9 JWT 사용 인증 구현 및 DB 데이터 조회  (0) 2024.11.26
5 실습8 로그인 API 테스트  (1) 2024.11.25

관련글

티스토리툴바