■ pytest 모듈을 사용해 FastAPI 애플리케이션에서 단위 테스트를 하는 방법을 보여준다.
※ 본 예제 코드는 [FastAPI 클래스 : JWT 인증 애플리케이션 만들기 (MongoDB 연동)] 자료에서 사용된 예제 코드에 단위 테스트 코드를 추가한 것이다.
※ 추가된 단위 테스트 관련 예제 코드부터 먼저 소개를 하고 [FastAPI 클래스 : JWT 인증 애플리케이션 만들기 (MongoDB 연동)] 자료에서 사용된 예제 코드를 표시했다.
[추가 단위 테스트 예제 코드]
▶ pytest.ini
|
1 2 3 4 |
[pytest] asyncio_mode = auto |
▶ test/__init__.py
|
1 2 3 |
▶ test/conftest.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
import asyncio import httpx import pytest from application_setting import ApplicationSetting from main import fastAPI from model.event import Event from model.user import User @pytest.fixture(scope = "session") def event_loop(): # 함수명 변경시 에러가 발생한다. loop = asyncio.get_event_loop() yield loop loop.close() async def initializeDatabase(): applicationSetting = ApplicationSetting() await applicationSetting.initializeDatabase() @pytest.fixture(scope = "session") async def defaultClient(): await initializeDatabase() async with httpx.AsyncClient(app = fastAPI, base_url = "http://127.0.0.1:8000") as client: yield client await User.find_all().delete() await Event.find_all().delete() |
※ 상기 파일명으로 설정해야 한다.
▶ test/test_user.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 |
import httpx import pytest @pytest.mark.asyncio async def testUserSignup(defaultClient : httpx.AsyncClient) -> None: payloadDictionary = { "email" : "testuser@packt.com", "password" : "testpassword", } headerDictionary = { "accept" : "application/json", "Content-Type": "application/json" } response = await defaultClient.post("/user/signup", json = payloadDictionary, headers = headerDictionary) responseDictionary = { "message" : "User created successfully" } assert response.status_code == 200 assert response.json() == responseDictionary @pytest.mark.asyncio async def testSignUp(defaultClient : httpx.AsyncClient) -> None: payloadDictionary = { "username" : "testuser@packt.com", "password" : "testpassword" } headerDictionary = { "accept" : "application/json", "Content-Type" : "application/x-www-form-urlencoded" } response = await defaultClient.post("/user/signin", data = payloadDictionary, headers= headerDictionary) assert response.status_code == 200 assert response.json()["token_type"] == "Bearer" |
▶ test/test_event.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
import httpx import pytest from auth.jwt_helper import ceateAccessToken from model.event import Event @pytest.fixture(scope = "module") async def accessToken() -> str: return ceateAccessToken("testuser@packt.com") @pytest.fixture(scope = "module") async def mockEvent() -> Event: newEvent = Event( creator = "testuser@packt.com", title = "FastAPI Book Launch", image = "https://linktomyimage.com/image.png", description = "We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!", tagList = ["python", "fastapi", "book", "launch"], location = "Google Meet" ) await Event.insert_one(newEvent) yield newEvent @pytest.mark.asyncio async def testCreateEvent(defaultClient : httpx.AsyncClient, accessToken : str) -> None: payloadDictionary = { "creator" : "testuser@packt.com", "title" : "FastAPI Book Launch", "image" : "https://linktomyimage.com/image.png", "description" : "We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!", "tagList" : ["python","fastapi","book","launch"], "location" : "Google Meet", } headerDictionary = { "Content-Type" : "application/json", "Authorization" : f"Bearer {accessToken}" } response = await defaultClient.post("/event/new", json = payloadDictionary, headers = headerDictionary) responseDictionary = {"message" : "Event created successfully"} assert response.status_code == 200 assert response.json() == responseDictionary print(response) @pytest.mark.asyncio async def testGetEventList(defaultClient : httpx.AsyncClient, mockEvent : Event) -> None: response = await defaultClient.get("/event/") assert response.status_code == 200 assert response.json()[0]["creator"] == str(mockEvent.creator) @pytest.mark.asyncio async def testGetEvent(defaultClient : httpx.AsyncClient, mockEvent : Event) -> None: url = f"/event/{str(mockEvent.id)}" response = await defaultClient.get(url) assert response.status_code == 200 assert response.json()["creator"] == mockEvent.creator assert response.json()["title"] == str(mockEvent.title) @pytest.mark.asyncio async def testGetEventCount(defaultClient : httpx.AsyncClient) -> None: response = await defaultClient.get("/event/") events = response.json() assert response.status_code == 200 assert len(events) == 2 @pytest.mark.asyncio async def testUpdateEvent(defaultClient : httpx.AsyncClient, mockEvent : Event, accessToken : str) -> None: payloadDictionary = { "title" : "Updated FastAPI event" } headerDictionary = { "Content-Type" : "application/json", "Authorization" : f"Bearer {accessToken}" } url = f"/event/{str(mockEvent.id)}" response = await defaultClient.put(url, json = payloadDictionary, headers = headerDictionary) assert response.status_code == 200 assert response.json()["title"] == payloadDictionary["title"] @pytest.mark.asyncio async def test_delete_event(defaultClient : httpx.AsyncClient, mockEvent : Event, accessToken : str) -> None: headerDictionary = { "Content-Type" : "application/json", "Authorization" : f"Bearer {accessToken}" } url = f"/event/{mockEvent.id}" response = await defaultClient.delete(url, headers = headerDictionary) responseDictionary = {"message" : "Event deleted successfully."} assert response.status_code == 200 assert response.json() == responseDictionary @pytest.mark.asyncio async def test_get_event_again(defaultClient : httpx.AsyncClient, mockEvent : Event) -> None: url = f"/event/{str(mockEvent.id)}" response = await defaultClient.get(url) assert response.status_code == 404 |
▶ requirements.txt
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 |
anyio==3.5.0 asgi-lifespan==1.0.1 asgiref==3.5.0 attrs==21.4.0 bcrypt==3.2.2 beanie==1.11.0 certifi==2021.10.8 cffi==1.15.0 charset-normalizer==2.0.12 click==8.0.4 coverage==6.3.3 cryptography==36.0.2 dnspython==2.2.1 ecdsa==0.17.0 email-validator==1.1.3 fastapi==0.77.1 greenlet==3.0.3 h11==0.12.0 httpcore==0.14.7 httpx==0.22.0 idna==3.3 iniconfig==1.1.1 Jinja2==3.0.3 MarkupSafe==2.1.0 motor==2.5.0 multidict==6.0.2 packaging==21.3 passlib==1.7.4 pluggy==1.0.0 py==1.11.0 pyasn1==0.4.8 pycparser==2.21 pydantic==1.9.0 pymongo==3.12.0 pyparsing==3.0.9 pytest==7.1.2 pytest-asyncio==0.18.3 python-dotenv==0.20.0 python-jose==3.3.0 python-multipart==0.0.5 rfc3986==1.5.0 rsa==4.8 six==1.16.0 sniffio==1.2.0 SQLAlchemy==1.4.32 sqlalchemy2-stubs==0.0.2a20 sqlmodel==0.0.6 starlette==0.19.1 toml==0.10.2 tomli==2.0.1 typing_extensions==4.1.1 uvicorn==0.17.6 yarl==1.7.2 |
[FastAPI 클래스 : JWT 인증 애플리케이션 만들기 (MongoDB 연동) 예제 코드]
▶ .env
|
1 2 3 4 |
DATABASE_URL=mongodb://localhost:27017/testdb SECRET_KEY=pass1234567 |
※ testdb : MongoDB 데이터베이스명
※ pass1234567 : JWT 비밀키
▶ application_setting.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
from typing import Optional from pydantic import BaseSettings from model.event import Event from model.user import User from database.db_helper import DBHelper class ApplicationSetting(BaseSettings): DATABASE_URL : Optional[str] = None SECRET_KEY : Optional[str] = None async def initializeDatabase(self): dbHelper = DBHelper() await dbHelper.initialize(self.DATABASE_URL, [Event, User]) class Config: env_file = ".env" |
※ .env : 애플리케이션 설정 파일명
▶ auth/hash_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 |
from passlib.context import CryptContext cryptContext = CryptContext(schemes=["bcrypt"], deprecated = "auto") class HashHelper: def createHash(self, password : str): return cryptContext.hash(password) def verifyHash(self, plain_password : str, hashed_password : str): return cryptContext.verify(plain_password, hashed_password) |
▶ auth/jwt_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 |
import time from datetime import datetime from jose import jwt, JWTError from fastapi import HTTPException, status from application_setting import ApplicationSetting applicationSetting = ApplicationSetting() def ceateAccessToken(user : str): payload = { "user" : user, "expires" : time.time() + 3600 } token = jwt.encode(payload, applicationSetting.SECRET_KEY, algorithm = "HS256") return token def verifyAccessToken(token : str): try: decodedTokenDictionary = jwt.decode(token, applicationSetting.SECRET_KEY, algorithms=["HS256"]) expiresTimeStamp = decodedTokenDictionary.get("expires") if expiresTimeStamp is None: raise HTTPException(status_code = status.HTTP_400_BAD_REQUEST, detail = "No access token supplied") if datetime.utcnow() > datetime.utcfromtimestamp(expiresTimeStamp): raise HTTPException(status_code = status.HTTP_403_FORBIDDEN, detail = "Token expired!") return decodedTokenDictionary except JWTError: raise HTTPException(status_code = status.HTTP_400_BAD_REQUEST, detail = "Invalid token") |
▶ auth/auth_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
from auth.jwt_helper import verifyAccessToken from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer oauth2PasswordBearer = OAuth2PasswordBearer(tokenUrl = "/user/signin") async def authenticate(token : str = Depends(oauth2PasswordBearer)) -> str: if not token: raise HTTPException(status_code = status.HTTP_403_FORBIDDEN, detail = "Sign in for access") decodedTokenDictionary = verifyAccessToken(token) return decodedTokenDictionary["user"] |
▶ model/user.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
from pydantic import BaseModel, EmailStr from beanie import Document class User(Document): email : EmailStr password : str class Collection: name = "user" class Config: json_schema_extra = { "example" : { "email" : "fastapi@packt.com", "password" : "strong!!!" } } class TokenResponse(BaseModel): access_token : str token_type : str |
▶ model/event.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
from typing import Optional, List from pydantic import BaseModel from beanie import Document class Event(Document): creator : Optional[str] title : str image : str description : str tagList : List[str] location : str class Collection: name = "event" class Config: json_schema_extra = { "example" : { "title" : "FastAPI BookLaunch", "image" : "https://linktomyimage.com/image.png", "description" : "We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!", "tagList" : ["python", "fastapi", "book", "launch"], "location" : "Google Meet" } } class EventUpdate(BaseModel): title : Optional[str] image : Optional[str] description : Optional[str] tagList : Optional[List[str]] location : Optional[str] class Config: json_schema_extra = { "example" : { "title" : "FastAPI BookLaunch", "image" : "https://linktomyimage.com/image.png", "description" : "We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!", "tagList" : ["python", "fastapi", "book", "launch"], "location" : "Google Meet" } } |
▶ database/db_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
from motor.motor_asyncio import AsyncIOMotorClient from beanie import init_beanie class DBHelper: def __init__(self): pass async def initialize(self, databaseURL, documentModelList): self.databaseURL = databaseURL self.documentModelList = documentModelList client = AsyncIOMotorClient(self.databaseURL) await init_beanie(database = client.get_default_database(), document_models = self.documentModelList) |
▶ database/document_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
from pydantic import BaseModel from beanie import PydanticObjectId class DocumentHelper: def __init__(self, modelMetaClass): self.modelMetaClass = modelMetaClass async def create(self, document): await document.create() return async def get(self, id : PydanticObjectId): document = await self.modelMetaClass.get(id) if document: return document return False async def getList(self): documentList = await self.modelMetaClass.find_all().to_list() return documentList async def update(self, id : PydanticObjectId, updateModel : BaseModel): documentID = id updateModelDictionary = updateModel.dict() updateModelDictionary = {k : v for k, v in updateModelDictionary.items() if v is not None} updateQuery = {"$set" : {field : value for field, value in updateModelDictionary.items()}} document = await self.get(documentID) if not document: return False await document.update(updateQuery) return document async def delete(self, id : PydanticObjectId): document = await self.get(id) if not document: return False await document.delete() return True |
▶ route/user.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordRequestForm from model.user import User, TokenResponse from auth.hash_helper import HashHelper from auth.jwt_helper import ceateAccessToken from database.document_helper import DocumentHelper userAPIRouter = APIRouter(tags = ["User"]) userDocumentHelper = DocumentHelper(User) hashHelper = HashHelper() @userAPIRouter.post("/signup") async def signUp(user : User) -> dict: existingUser = await User.find_one(User.email == user.email) if existingUser: raise HTTPException(status_code = status.HTTP_409_CONFLICT, detail = "User with email provided exists already.") hashedPassword = hashHelper.createHash(user.password) user.password = hashedPassword await userDocumentHelper.create(user) return {"message" : "User created successfully"} @userAPIRouter.post("/signin", response_model = TokenResponse) async def signIn(oauth2PasswordRequestForm : OAuth2PasswordRequestForm = Depends()) -> dict: existingUser = await User.find_one(User.email == oauth2PasswordRequestForm.username) if not existingUser: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "User with email does not exist.") if hashHelper.verifyHash(oauth2PasswordRequestForm.password, existingUser.password): accessToken = ceateAccessToken(existingUser.email) return { "access_token" : accessToken, "token_type" : "Bearer" } raise HTTPException(status_code = status.HTTP_401_UNAUTHORIZED, detail = "Invalid details passed.") |
▶ route/event.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
from typing import List from beanie import PydanticObjectId from fastapi import APIRouter, Depends, HTTPException, status from model.event import Event, EventUpdate from auth.auth_helper import authenticate from database.document_helper import DocumentHelper eventAPIRouter = APIRouter(tags = ["Event"]) eventDocumentHelper = DocumentHelper(Event) @eventAPIRouter.get("/", response_model = List[Event]) async def getEventList() -> List[Event]: eventList = await eventDocumentHelper.getList() return eventList @eventAPIRouter.get("/{id}", response_model = Event) async def getEvent(id : PydanticObjectId) -> Event: event = await eventDocumentHelper.get(id) if not event: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist") return event @eventAPIRouter.post("/new") async def createEvent(event : Event, user : str = Depends(authenticate)) -> dict: event.creator = user await eventDocumentHelper.create(event) return {"message" : "Event created successfully"} @eventAPIRouter.put("/{id}", response_model = Event) async def updateEvent(id : PydanticObjectId, eventUpdate : EventUpdate, user : str = Depends(authenticate)) -> Event: event = await eventDocumentHelper.get(id) if event.creator != user: raise HTTPException(status_code = status.HTTP_400_BAD_REQUEST, detail = "Operation not allowed") updatedEvent = await eventDocumentHelper.update(id, eventUpdate) if not updatedEvent: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist") return updatedEvent @eventAPIRouter.delete("/{id}") async def deleteEvent(id : PydanticObjectId, user : str = Depends(authenticate)) -> dict: event = await eventDocumentHelper.get(id) if not event: raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event not found") if event.creator != user: raise HTTPException(status_code = status.HTTP_400_BAD_REQUEST, detail = "Operation not allowed") event = await eventDocumentHelper.delete(id) return {"message" : "Event deleted successfully."} |
▶ main.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 |
import uvicorn from fastapi import FastAPI from fastapi.responses import RedirectResponse from fastapi.middleware.cors import CORSMiddleware from application_setting import ApplicationSetting from route.event import eventAPIRouter from route.user import userAPIRouter applicationSetting = ApplicationSetting() fastAPI = FastAPI() fastAPI.add_middleware( CORSMiddleware, allow_origins = ["*"], allow_credentials = True, allow_methods = ["*"], allow_headers = ["*"], ) fastAPI.include_router(userAPIRouter , prefix = "/user" ) fastAPI.include_router(eventAPIRouter, prefix = "/event") @fastAPI.on_event("startup") async def startUp(): await applicationSetting.initializeDatabase() @fastAPI.get("/") async def home(): return RedirectResponse(url="/event/") if __name__ == "__main__": uvicorn.run("main:fastAPI", host = "127.0.0.1", port = 8000, reload = True) |