앞 단계 참조 링크:
00 개요
- 목적: PyTest의 fixture 기능 적용하기 - fixture 생성
- PyTest의 Fixture이란
- 테스트 코드 안에서 반복적으로 사용하는 데이터 또는 객체가 있을 경우 해당 데이터/객체를 재사용할 수 있도록 도와주는 PyTest의 기능
- 사실,, 지금까지 이 fixture의 기능을 사용하고 있었음.......!!!!!!!!!
- 앞 강의에서 mocker를 import 하지 않았음에도 불구하고 pytest-mock 라이브러리 안에 mocker라는 fixture가 정의되어 있기에 pytest-mock 라이브러리를 설치만 해도 mocker를 사용할 수 있게 된 것!
01 fixture 생성 및 사용하기
1. test_main.py를 예시로 보기
- test_main.py에 보면 TestClient()라는 클라이언트 객체를 파일마다 만들어서 사용하고 있음
- 이 클라이언트 같은 경우 fixture로 등록하면 다른 파일들이 추가된다고 할 때 그 파일들에서도 만들어진 fixture 사용이 가능하게될 것!
- 즉, 파일마다 이런 객체를 만드는 대신 해당 객체를 fixture로 설정하면 여러 파일 간 재사용 가능!
2. conftest.py 파일 생성 및 작성
- 경로: /c/Users/관리자/Desktop/projects/todos/src/tests/
- 위의 경로 안에 conftest.py라는 이름의 새 파일 생성하기
- 다음과 같이 작성하기:
# /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py 내용
import pytest
from fastapi.testclient import TestClient
from src.main import app
@pytest.fixture
def client():
return TestClient(app=app)
- 설명:
- import pytest
- pytest의 fixture라는 데코레이터 사용하기 위해 참조
- from fastapi.testclient import TestClient
- TestClient 라는 객체를 만들기 위해 참조
- from src.main import app
- TestClient 객체 생성 시 우리가 만든 FastAPI 객체인 app을 app으로 전달해주기 위해 참조
- @pytest.fixture
- 바로 밑에서 만들 client (i.e., def client(): ) 를 fixture로 사용하기 위해 pytest의 fixture라는 데코레이터 추가!
- def client():
- return TestClient(app=app)
- 함수 형태로 client라는 이름을 가진 fixture를 만드는 것!
- 반환값으로 TestClient(app=app) 객체를 주어 client 라는 fixture 자체가 TestClient(app=app)이 되도록 설정하는 것
- 여기서 app은 우리가 main.py에서 정의한 FastAPI 클래스임! (우리가 app이라는 이름으로 정의하였음)
- 이렇게 fixture를 생성하면 PyTest가 이 client()라는 함수를 fixture로 인식하여 우리가 테스트 코드 안에서 global하게 사용할 수 있게 됨!!
3. test_main.py 수정
- 이제 더 이상 TestClient(app=app)을 정의해주지 않아도 되기에 (fixture로 등록 되었기에) test_main.py를 다음과 같이 수정:
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용
# from fastapi.testclient import TestClient # 삭제 - 더 이상 이 파일 안에서 불필요
from src.schema.response import ToDoSchema
# from src.main import app # 삭제 - 더 이상 이 파일 안에서 불필요
from src.database.orm import ToDo
# client = TestClient(app=app) # 삭제 - conftest.py에서 fixture로 정의해 주었기에
# 필요 없어짐
def test_health_check(client): # 인자로 client 추가
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"ping": "pong"}
def test_get_todos(client, mocker): # 인자로 client 추가
# order = ASC
mocker.patch("src.main.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},
]
}
- Note: fixture 사용할 때 끝에 "()" 사용하지 않음!
- 예) test_main.py에서 "def test_health_check(client):", "response = client.get("/")" 등 함수로 정의되었어도 "()"가 안 붙음!
이때까지의 코드들: 8개
- /c/Users/관리자/Desktop/projects/todos/src/main.py
- /c/Users/관리자/Desktop/projects/todos/src/database/orm.py
- /c/Users/관리자/Desktop/projects/todos/src/database/connection.py
- /c/Users/관리자/Desktop/projects/todos/src/database/repository.py
- /c/Users/관리자/Desktop/projects/todos/src/schema/request.py
- /c/Users/관리자/Desktop/projects/todos/src/schema/response.py
- /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py (수정 - 삭제, 추가)
- /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py (생성)
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends
from sqlalchemy.orm import Session
from typing import List
from src.database.connection import get_db
from src.database.repository import get_todos, get_todo_by_todo_id, create_todo, update_todo, delete_todo
from src.database.orm import ToDo
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest
app = FastAPI()
# 첫 화면 API
@app.get("/")
def health_check_handler():
return {"ping": "pong"}
# GET Method 사용하여 전체 조회 API
@app.get("/todos", status_code=200)
def get_todos_handler(
order: str | None = None,
session: Session = Depends(get_db),
) -> ToDoListSchema:
todos: List[ToDo] = get_todos(session=session)
if order and order == "DESC":
# return todos[::-1]
return ToDoListSchema(
todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
)
# return todos
return ToDoListSchema(
todos=[ToDoSchema.from_orm(todo) for todo in todos]
)
# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(
todo_id: int,
session: Session=Depends(get_db),
) -> ToDoSchema:
todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
if todo:
return ToDoSchema.from_orm(todo)
raise HTTPException(status_code=404, detail="ToDo Not Found")
# POST Medthod 사용하여 todo 생성 API
@app.post("/todos", status_code=201)
def create_todo_handler(
request: CreateToDoRequest,
session: Session = Depends(get_db),
) -> ToDoSchema:
todo: ToDo = ToDo.create(request=request) # id=None
todo: ToDo = create_todo(session=session, todo=todo) # id=int
return ToDoSchema.from_orm(todo)
# PATCH Method 사용하여 is_done 값 수정 API
@app.patch("/todos/{todo_id}", status_code=200)
def update_todo_handler(
todo_id: int,
is_done: bool = Body(..., embed=True),
session: Session = Depends(get_db),
):
todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
if todo:
# update - is_done값이 True이면 todo.done() 실행, False이면 todo.undone() 실행
todo.done() if is_done else todo.undone()
todo: ToDo = update_todo(session=session, todo=todo)
return ToDoSchema.from_orm(todo)
raise HTTPException(status_code=404, detail="ToDo Not Found")
# DELETE Method 사용하여 todo 아이템 삭제 API
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo_handler(
todo_id: int,
session: Session = Depends(get_db),
):
todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
if not todo:
raise HTTPException(status_code=404, detail="ToDo Not Found")
delete_todo(session=session, todo_id=todo_id)
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
from src.schema.request import CreateToDoRequest
Base = declarative_base()
# ToDo 클래스 모델링 한 것
class ToDo(Base):
__tablename__ = 'todo'
id = Column(Integer, primary_key=True, index=True)
contents = Column(String(256), nullable=False)
is_done = Column(Boolean, nullable=False)
def __repr__(self):
return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
@classmethod
def create(cls, request: CreateToDoRequest) -> "ToDo":
return cls(
# id값은 DB에 의해 자동으로 결정 되서 별도로 지정안해줌
contents=request.contents,
is_done=request.is_done,
)
def done(self) -> "ToDo":
# ToDo의 is_done 값을 True로 변경 후 ToDo 반환
self.is_done = True
return self
def undone(self) -> "ToDo":
# ToDo의 is_done 값을 False로 변경 후 ToDo 반환
self.is_done = False
return self
# /c/Users/관리자/Desktop/projects/todos/src/database/connection.py 내용
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "mysql+pymysql://root:todos@127.0.0.1:3306/todos"
engine = create_engine(DATABASE_URL, echo=True)
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
session = SessionFactory()
try:
yield session
finally:
session.close()
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from typing import List
from src.database.orm import ToDo
def get_todos(session: Session) -> List[ToDo]:
return list(session.scalars(select(ToDo)))
def get_todo_by_todo_id(session: Session, todo_id: int) -> ToDo | None:
return session.scalar(select(ToDo).where(ToDo.id == todo_id))
def create_todo(session: Session, todo: ToDo) -> ToDo:
session.add(instance=todo)
session.commit()
session.refresh(instance=todo)
return todo
def update_todo(session: Session, todo: ToDo) -> ToDo:
session.add(instance=todo)
session.commit()
session.refresh(instance=todo)
return todo
def delete_todo(session: Session, todo_id: ToDo) -> None:
session.execute(delete(ToDo).where(ToDo.id == todo_id))
session.commit()
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List
class ToDoSchema(BaseModel):
id: int
contents: str
is_done: bool
class Config:
from_attributes = True
orm_mode = True # 이거 추가 해야지 ConfigError 안남
class ToDoListSchema(BaseModel):
todos: List[ToDoSchema]
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용
from pydantic import BaseModel
class CreateToDoRequest(BaseModel):
contents: str
is_done: bool
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용
# from fastapi.testclient import TestClient # 삭제
from src.schema.response import ToDoSchema
# from src.main import app # 삭제
from src.database.orm import ToDo
# client = TestClient(app=app) # 삭제
def test_health_check(client): # 인자로 client 추가
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"ping": "pong"}
def test_get_todos(client, mocker): # 인자로 client 추가
# order = ASC
mocker.patch("src.main.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},
]
}
# /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py 내용
import pytest
from fastapi.testclient import TestClient
from src.main import app
@pytest.fixture
def client():
return TestClient(app=app)
