REAL Python – FastAPI] – “풀스택 블로그 개발하기 – 회원가입 API 구현하기”
REAL Python – FastAPI] – “풀스택 블로그 개발하기 – 회원가입 API 구현하기”
테스트 코드 작성하기
테스트 케이스가 없다면 모든 변경이 잠정적인 버그다. 아키텍처가 아무리 유연하더라도, 설계를 아무리 잘 나눴더라도, 테스트 케이스가 없으면 개발자는 변경을 주저한다. 버그가 숨어들까 두렵기 때문이다.
클린 코드, p157
이번 시간에는 이전 시간에 작성했던 코드의 문제점을 살펴보고, 리팩토링한 후 회원가입을 구현합니다. 먼저 무언가 변경을 만들기 전, 우리의 API 를 테스트하는 시간을 가집시다. 그래야 안심하고 아키텍처와 설계를 개선할 수 있기 때문입니다.
pytest 설치하기
먼저, 테스트를 위해 유용한 파이썬 패키지인 pytest 를 설치하겠습니다. 아래의 명령어를 터미널에 입력해 pytest 를 설치하세요.
poetry add pytest -G test
그리고, 테스트 코드를 위해 루트 디렉토리 아래에 tests/ 패키지를, tests/ 패키지 안에 users 패키지를 만듭니다. 해당 패키지 아래에는 새 test_routers 파일을 만들어주세요.
그리고, 아래의 코드를 입력합니다.
from fastapi.testclient import TestClient
from fastapi import status
from app import app
client = TestClient(app)
def test_read_user_list() -> None:
response = client.get("/users")
assert response.status_code == status.HTTP_200_OK
간단해 보이는 위의 코드에서 우리는 몇 가지를 알 수 있습니다!
FastAPI에서 제공하는TestClient를 사용해, 우리의 웹 애플리케이션 서버에 요청을 보낼 수 있는 테스트용 클라이언트를 만들었습니다.- 7줄에서 새로운 테스트 함수를 정의했습니다.
/users경로에HTTP GET요청을 보냈을 때, 해당 요청에 대한 응답의 상태 코드가 200이어야 테스트가 통과한다고 적혀 있습니다.
그리고, 터미널에 pytest . 을 입력해 보세요. 자동으로 앞에 test_ 가 붙은 함수들을 찾아, 테스트를 실행해줄 겁니다.
엇, 그런데 테스트에 실패했습니다. 응답에 대한 상태 코드로 200을 기대했는데, 404가 반환되었다네요.
“사용자 목록을 나타내는 API” 는 URL 에 users 라는 자원이 들어가는 게 더 알맞아 보입니다. routers.py 에서 아래와 같이 함수를 수정해 주면 되겠어요.
그리고, 테스트를 다시 실행해 보세요. 성공할 겁니다.
하지만.. 위에서는 조금 문제가 있습니다.
종속성 오버라이드하며 테스트하기
위의 문제는, /users 를 테스트할 때에 실제 데이터베이스 에 접근한다는 점입니다. 현재 API 는 단지 “조회” 용이지만, 나중에 “삭제” 나 “생성” 과 같은 테스트를 할 때에는 실제 데이터베이스에 있는 정보를 삭제하거나 생성해 버릴 수도 있겠죠. 이를 분리해야 합니다.
먼저, 이를 위해 pytest 플러그인 몇 가지를 설치해야 합니다.
poetry add pytest-asyncio -G test, poetry add pytest-env -G test 를 터미널에 입력하여 설치해 주세요.
그리고, pyproject.toml 에 아래의 내용을 입력합니다. 아래의 내용은 pytest 실행 시 환경 변수로 주입될 값입니다.
그리고, 몇 가지 복잡한 테스트를 위해서 tests/ 아래에 conftest.py 를 만듭니다. 이 곳에서는 테스트를 위한 세션 객체, 테스트 전 테이블 생성, 더미 회원 등을 미리 만드는 등 테스트에 필요한 공통 작업들이 정의되는 곳입니다.
from typing import AsyncGenerator, Sequence
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession
from api.users.models import User
from config.db import engine, Model, AsyncSessionLocal
@pytest_asyncio.fixture(autouse=True)
async def create_tables() -> AsyncGenerator[None, None]:
"""
Create tables before running tests.
Drop tables after running tests.
"""
async with engine.begin() as conn:
await conn.run_sync(Model.metadata.create_all)
yield
async with engine.begin() as conn:
await conn.run_sync(Model.metadata.drop_all)
@pytest_asyncio.fixture
async def test_session() -> AsyncGenerator[AsyncSession, None]:
"""Create a new database session for a test."""
session = AsyncSessionLocal()
try:
yield session
finally:
await session.close()
@pytest_asyncio.fixture
async def one_test_user(test_session: AsyncSession) -> User:
test_user = User(
email="twicegoddessana1229@gmail.com",
username="twicegoddessana1229",
password="twicegoddessana1229",
)
test_session.add(test_user)
await test_session.commit()
return test_user
@pytest_asyncio.fixture
async def fifty_test_users(test_session: AsyncSession) -> Sequence[User]:
test_users = [
User(
email=f"test{i+1}@gmail.com",
username=f"test{i+1}",
password=f"test{i+1}",
)
for i in range(50)
]
test_session.add_all(test_users)
await test_session.commit()
return test_users
각각의 함수를 살펴보며 알아볼까요?
- 공통적으로 붙어있는
pytest_asyncio.fixture은pytest에게 “이것은async테스트를 위한 픽스쳐야!” 를 알려주는 역할을 합니다. create_tables는 테스트 전 필요한 테이블을 만들어주는 역할을 합니다. 하나의 테스트 함수마다, 해당 테스트를 위한 테이블이 만들어질 겁니다. 테스트가 끝나면, 명시적으로 모든 테이블을 드랍합니다.create_tables에서 주목할 점은,autouse=True를 사용하여 테스트 코드에서 명시적으로 요청하지 않아도 테이블을 만들게끔 했다는 것입니다.test_session은 테스트만을 위한 세션을 만들어주는 역할을 합니다. 환경 변수로 주입된 값을 사용하여 세션 객체를 만들어주는 역할을 합니다.in-memory SQLite에서 테스트를 수행할 수 있습니다.one_test_user는 테스트만을 위한 회원 하나를 만들어주는 역할을 합니다.fifty_test_users는 테스트만을 위한 회원 50명을 만들어주는 역할을 합니다.
회원 목록 API 테스트 코드 작성하기
그러면, 실제 테스트 코드를 아래와 같이 작성할 수 있습니다.
from typing import Sequence
import pytest
from fastapi.testclient import TestClient
from fastapi import status
from sqlalchemy.ext.asyncio import AsyncSession
from api.users.models import User
from app import app
client = TestClient(app)
@pytest.mark.asyncio
async def test_read_user_list_empty() -> None:
response = client.get("/users")
assert response.status_code == status.HTTP_200_OK
assert "data" in response.json()
assert len(response.json()["data"]) == 0
@pytest.mark.asyncio
async def test_read_user_list_with_one_user(one_test_user: User) -> None:
response = client.get("/users")
assert response.status_code == status.HTTP_200_OK
assert "data" in response.json()
assert len(response.json()["data"]) == 1
assert response.json()["data"][0]["email"] == one_test_user.email
@pytest.mark.asyncio
async def test_read_user_list_with_fifty_users(
fifty_test_users: Sequence[User]
) -> None:
response = client.get("/users")
assert response.status_code == status.HTTP_200_OK
assert "data" in response.json()
assert len(response.json()["data"]) == 50
assert response.json()["data"][0]["email"] == fifty_test_users[0].email
assert response.json()["data"][49]["email"] == fifty_test_users[49].email
각각의 코드는 아래의 부분들을 검증합니다.
- 각각
users에 요청을 보냅니다. - 요청의 HTTP 응답 코드가 200 OK 임을 확인합니다.
- 요청받은
JSON스키마에 “data” 라는 키가 있는지 확인합니다. - 데이터베이스에 있는 회원의 정보를 잘 나타내는지 값을 검증합니다.
그리고 테스트 코드를 pytest . 을 통해서 실행해 보세요. 아래와 같이 잘 통과되었다고 할 겁니다.
pytest뒤에 붙는.은 “현재 디렉토리” 를 의미합니다. 경로를 입력함으로서 원하는 파일만 테스트할 수 있습니다.
회원가입 API 구현하기
이전 시간 우리가 정했던 요구사항은 아래와 같았습니다.
- 먼저 회원가입은 “닉네임, 이메일, 비밀번호, 비밀번호 재입력” 을 입력함으로서 진행하는 것으로 정합시다.
- 닉네임과 이메일은 중복이 없어야 합니다.
- 닉네임은 특수문자가 없는 영어&숫자로만 이루어진 25자 이하의 문자열이어야 합니다.
- 이메일은 실제 존재하는 이메일이어야 합니다.
- 비밀번호는 입력 시 여덟 자 이상이어야 합니다.
- 비밀번호는 암호화되어 저장되어야 합니다.
- 회원 등급은 “에디터”, “관리자”, “독자” 세 가지가 있는 것으로 합시다.
- “에디터”는 블로그에 새로운 글을 발행하거나, 수 있습니다.
- “관리자” 는 저자의 역할을 포함해 회원 관리, 블로그 메타 정보를 수정하는 등의 작업을 수행할 수 있습니다.
- “독자” 는 댓글을 달거나, 우리 블로그를 구독하는 등의 작업을 수행할 수 있습니다.
위의 요구사항 하나하나를 API 스펙으로 바꿔 볼까요?
먼저 회원가입은 “닉네임, 이메일, 비밀번호, 비밀번호 재입력” 을 입력함으로서 진행하는 것으로 정합시다.
우리는 위의 상황에 대해서, 각각 알맞지 않은 데이터가 들어온다면 상태 코드 422 와 함께 알맞는 에러 메시지를 던져줘야 합니다. FastAPI 가 대신 처리해주는 부분이 있지만, 그걸 조금 간단히 바꿉시다. 아래와 같은 응답이 오게끔요:
{
"detail": [
{
"message": "some validation error",
"location": "password",
}
]
}회원 등급은 “에디터”, “관리자”, “독자” 세 가지가 있는 것으로 합시다.
회원가입을 할 때에는, 기본 등급으로 “독자” 를 선택하겠습니다. “에디터”, “관리자” 등은 추후 “관리자” 등급이 등업을 시켜 주는 것으로 하면 되겠어요.
그러면, 각각의 상황에 대해서 테스트를 작성해 볼 수 있겠습니다. 우려되는 부분에 대한 테스트를 먼저 작성하고, 테스트를 계속 돌려 보며 통과할 때까지 우리의 코드를 한번 다듬어 봅시다.
회원가입을 위한 테스트 코드 작성하기
아래의 함수들을 하나씩 test_routers.py 에 추가합니다.
회원가입 성공 시, 상태 코드와 응답 값을 검증하는 테스트 코드
@pytest.mark.asyncio
async def test_register_user_success_response() -> None:
response = client.post(
url="/users",
json={
"email": "test@gmail.com",
"username": "test",
"password": "password",
},
)
assert response.status_code == status.HTTP_201_CREATED
assert "id" in response.json()
assert "username" in response.json()
assert "email" in response.json()
assert "password" not in response.json()
- 위의 테스트 함수는 “회원가입이 성공했을 때, 필요한 정보만 노출하는지” 에 대한 역할을 합니다.
- 회원가입이 성공적으로 이루어졌다면, 서버에서 보내주는 상태 코드는 201 이어야 합니다.
회원가입 성공 시, 데이터베이스에 실제로 값이 저장되는지를 검증하는 테스트 코드
@pytest.mark.asyncio
async def test_user_register_success_adds_to_db(test_session: AsyncSession) -> None:
response = client.post(
url="/users",
json={
"email": "test@gmail.com",
"username": "test",
"password": "password",
},
)
assert response.status_code == status.HTTP_201_CREATED
results = await test_session.execute(select(User))
users = results.scalars().all()
assert len(users) == 1
assert users[0].email == "test@gmail.com"
assert users[0].username == "test"
- 위의 테스트 함수는 성공적으로 회원가입이 이루어진 이후, 실제 데이터베이스에 사용자의 정보가 잘 담겨있는지를 확인합니다.
클라이언트의 잘못된 값에, 적당한 상태 코드를 반환하는지에 대한 테스트 코드
@pytest.mark.asyncio
@pytest.mark.parametrize(
"invalid_data",
[
{
"email": "", # empty email
"username": "test",
"password": "password",
},
{
"email": "test@example.com",
"username": "", # empty username
"password": "password",
},
{
"email": "test@example.com",
"username": "test",
"password": "", # empty password
},
],
)
async def test_register_user_with_empty_fields(invalid_data: dict[str, str]) -> None:
response = client.post(
url="/users",
json=invalid_data,
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
- 위의 테스트 함수는 비어있는 필드에 대한 유효성 검사가 잘 이루어지는지를 검증합니다. 만약 셋 중 하나라도 비어있는 필드가 있다면, 상태 코드로 422 를 응답해 주어야 합니다.
@pytest.mark.parametrize코드를 통해서, 위의invalid_data리스트에 대한 모든 경우를 테스트할 수 있습니다.
중복된 이메일과 유저네임을 제대로 검증하는지에 대한 테스트 코드
@pytest.mark.asyncio
async def test_register_user_with_duplicate_email(one_test_user: User) -> None:
response = client.post(
url="/users",
json={
"email": one_test_user.email,
"username": "test",
"password": "password",
},
)
assert response.status_code == status.HTTP_409_CONFLICT
@pytest.mark.asyncio
async def test_register_user_with_duplicate_username(one_test_user: User) -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": one_test_user.username,
"password": "password",
},
)
assert response.status_code == status.HTTP_409_CONFLICT- 위의 두 테스트 함수는 회원가입 시 중복된 이메일이나 사용자명을 사용 중이라면, 그것에 대해서 409 상태 코드를 내려줘야 함을 검증합니다.
one_test_user는 이미conftest.py에서 만들어뒀었죠? 해당 이메일이나 사용자명과 함께 회원가입을 시도한다면, 서버는 그것을 “잘못된 요청” 이라고 단단히 일러줘야 합니다.
TIP: HTTP 409 Conflict 에 대해서 더 잘 알아보세요.
https://developer.mozilla.org/ko/docs/Web/HTTP/Status/409
너무 길거나 짧은 비밀번호와 유저네임을 제대로 검증하는지에 대한 테스트 코드
@pytest.mark.asyncio
async def test_register_user_with_short_password() -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": "test",
"password": "pass",
},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_register_user_with_short_username() -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": "t",
"password": "password",
},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_register_user_with_long_username() -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": "somefreakinglongusernamevalue",
"password": "password",
},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY- 비밀번호, 너무 짧은 사용자명, 너무 긴 사용자명은 422 응답을 내려주어야 합니다.
비밀번호가 제대로 암호화되어 저장되는지를 검증하는 테스트 코드
@pytest.mark.asyncio
async def test_password_is_hashed(test_session: AsyncSession) -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": "test",
"password": "password",
},
)
assert response.status_code == status.HTTP_201_CREATED
results = await test_session.execute(select(User))
users = results.scalars().all()
assert users[0].password != "password"
- 데이터베이스에 사용자 비밀번호가
password와 같이 원본 그대로 저장되어선 안 됩니다. 만에 하나 접근하면 안 되는 사람이 데이터베이스에 접근해도 비밀번호를 확인할 수 없게끔, 단방향 암호화된 형태로 저장되어 있어야 하죠. - 위의 테스트 코드에서는, 실제 회원가입을 진행한 후 데이터베이스에서 비밀번호를 가져와 그 값이 같지 않은지를 검증합니다.
.. 그러면 전체 테스트 코드는 아래와 같습니다.
from enum import Enum
from typing import Sequence
import pytest
from fastapi.testclient import TestClient
from fastapi import status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from api.users.models import User
from app import app
client = TestClient(app)
#############################
# Test cases for list users #
#############################
@pytest.mark.asyncio
async def test_read_user_list_empty() -> None:
response = client.get("/users")
assert response.status_code == status.HTTP_200_OK
assert "data" in response.json()
assert len(response.json()["data"]) == 0
@pytest.mark.asyncio
async def test_read_user_list_with_one_user(one_test_user: User) -> None:
response = client.get("/users")
assert response.status_code == status.HTTP_200_OK
assert "data" in response.json()
assert len(response.json()["data"]) == 1
assert response.json()["data"][0]["email"] == one_test_user.email
@pytest.mark.asyncio
async def test_read_user_list_with_fifty_users(
fifty_test_users: Sequence[User]
) -> None:
response = client.get("/users")
assert response.status_code == status.HTTP_200_OK
assert "data" in response.json()
assert len(response.json()["data"]) == 50
assert response.json()["data"][0]["email"] == fifty_test_users[0].email
assert response.json()["data"][49]["email"] == fifty_test_users[49].email
###########################
# Test cases for register #
###########################
@pytest.mark.asyncio
async def test_register_user_success_response() -> None:
response = client.post(
url="/users",
json={
"email": "test@gmail.com",
"username": "test",
"password": "password",
},
)
assert response.status_code == status.HTTP_201_CREATED
assert "id" in response.json()
assert "username" in response.json()
assert "email" in response.json()
assert "password" not in response.json()
@pytest.mark.asyncio
async def test_user_register_success_adds_to_db(test_session: AsyncSession) -> None:
response = client.post(
url="/users",
json={
"email": "test@gmail.com",
"username": "test",
"password": "password",
},
)
assert response.status_code == status.HTTP_201_CREATED
results = await test_session.execute(select(User))
users = results.scalars().all()
assert len(users) == 1
assert users[0].email == "test@gmail.com"
assert users[0].username == "test"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"invalid_data",
[
{
"email": "", # empty email
"username": "test",
"password": "password",
},
{
"email": "test@example.com",
"username": "", # empty username
"password": "password",
},
{
"email": "test@example.com",
"username": "test",
"password": "", # empty password
},
],
)
async def test_register_user_with_empty_fields(invalid_data: dict[str, str]) -> None:
response = client.post(
url="/users",
json=invalid_data,
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_register_user_with_duplicate_email(one_test_user: User) -> None:
response = client.post(
url="/users",
json={
"email": one_test_user.email,
"username": "test",
"password": "password",
},
)
assert response.status_code == status.HTTP_409_CONFLICT
@pytest.mark.asyncio
async def test_register_user_with_duplicate_username(one_test_user: User) -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": one_test_user.username,
"password": "password",
},
)
assert response.status_code == status.HTTP_409_CONFLICT
@pytest.mark.asyncio
async def test_register_user_with_short_password() -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": "test",
"password": "pass",
},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_register_user_with_short_username() -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": "t",
"password": "password",
},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_register_user_with_long_username() -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": "somefreakinglongusernamevalue",
"password": "password",
},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_password_is_hashed(test_session: AsyncSession) -> None:
response = client.post(
url="/users",
json={
"email": "test@test.com",
"username": "test",
"password": "password",
},
)
assert response.status_code == status.HTTP_201_CREATED
results = await test_session.execute(select(User))
users = results.scalars().all()
assert users[0].password != "password"
그리고 테스트를 실행해 볼까요?
역시, 테스트를 돌려 보면 구현하지 않았으니 당연히 에러가 발생합니다. 이전에 작성했던 “사용자 목록 조회 테스트” 빼고는 모두 실패하네요. 이를 하나하나 맞춰가 봅시다.
구현하기: 테스트 코드 통과시키기
위의 테스트 코드를 잘 살펴보면, 모조리 “예상되는 상태 코드는 ~~~ 인데, 실제로는 405 야! 그래서 실패했어” 입니다. 상태 코드 405는 허용되지 않는 메서드인데 그것을 서버에서 지원하지 않는다는 것을 의미합니다.
HTTP POST 메서드를 처리하는 라우터 작성하기
당연히 우리는 GET 메서드에 대한 요청을 처리하는 라우트만 작성했는데, 클라이언트에서 POST 요청을 보냈으니 서버는 그것을 어떻게 처리해야 할 건지 알 방도가 없다는 거죠. 서버에게 “POST 메서드는 이렇게 처리하는 거야” 를 알려주기 위해, users/routers.py 에 새로운 라우트 함수를 작성합니다.
@users_router.post("/users")
async def reigster_user(session: Annotated[AsyncSession, Depends(get_session)]):
pass
그리고 다시 테스트를 해 보면, 아래와 같이 405 관련 실패는 사라지지만, 여전히 상태 코드 관련 에러가 발생할 겁니다.
실패 시, 다양한 상태 코드 반환하기
이번에는 “성공했을 때는 201, 중복된 이메일과 사용자명으로 회원가입을 시도할 때는 409, 유효성 검증에 실패했을 때는 422여야 하는데 다 200이야!” 를 알려주고 있네요. 이처럼 성공하는 시나리오를 제외하고 추가적인 응답을 정의하려면, 아래와 같이 FastAPI 에서 제공하는 responses 를 지정함으로서 해결할 수 있습니다.
하지만 이를 위해서 준비해야 할 것이 있습니다. 우리가 Pyadntic 을 이용해 스키마를 작성했던 것이 기억나시나요? 우리는 그것을 사용함으로서 응답과 요청에 대한 형식을 지정할 수 있었죠. 그것을 작성해 봅시다.
Pydantic 으로 유효성 검증하고 응답 형식 정의하기
users/schemaspy 에 아래의 코드를 추가합니다.
class UserRegisterRequest(BaseModel):
username: str
email: EmailStr
password: str
class UserRegisterResponse(BaseModel):
id: int
username: str
email: EmailStr
is_active: bool
created_at: datetime
role: str
- 회원가입을 할 때에는
username, email, password를 지정합니다. - 회원가입 후 응답은
id, username등 필요한 필드를 노출합니다.
..그러면 라우팅 함수를 아래와 같이 바꿀 수 있습니다.
@users_router.post(
"/users",
response_model=UserRegisterResponse,
status_code=status.HTTP_201_CREATED,
)
async def reigster_user(session: Annotated[AsyncSession, Depends(get_session)]):
pass
이제 실제 로직을 작성하면 되겠어요.
먼저 POST 요청과 함께 서버는 클라이언트로부터 회원가입 데이터를 받아와야 합니다. 이를 위한 스키마는 미리 작성해 뒀었죠. FastAPI 에서는 해당 모델을 함수에서 사용할 수 있습니다.
위처럼, register_user 함수가 register_data 와 같은 추가적인 데이터를 인자로 받도록 합니다. 그리고 서버를 실행시켜, 문서를 확인해 보세요.
그러면, FastAPI 는 위처럼 모델을 분석하여 API 문서를 잘 생성해 줄뿐만 아니라, 유효성 검사도 수행해 줄 겁니다.
새로운 회원 저장하기
요청 스키마, 응답 모델을 모두 정의했으면 아래처럼 실제 데이터베이스에 접근해 새로운 사용자를 저장하는 로직을 작성할 수 있겠네요.
그러면 테스트를 해 봅시다.
몇몇 테스트 실패는 사라졌지만, 여전히 해결해야 할 문제가 여럿 남아 있네요.
- 맨 처음 나오는 두 줄의 에러는 “비어있는 값에 대해서 422 응답을 해 줘야 하는데, 그렇지 못했어!” 입니다.
- 세 번째, 네 번째 에러는
SQLite에서 발생시키는 에러입니다. 유일 제약 조건이 걸린 컬럼에 같은 필드를 넣으려고 시도하니, “동일한 이메일 혹은 회원이름이 있어!” 라고 에러를 발생시킵니다. - 그 다음에 해결해야 할 것은 너무 짧거나 긴 이름, 비밀번호에 대한 유효성 검사입니다. 유효성 검사가 제대로 이뤄지지 않고 있네요.
- 마지막 에러는 비밀번호가 암호화되지 않고, 그대로 저장되고 있음을 보여주는 테스트 실패입니다. 사용자가 입력한 비밀번호 그대로가 아니어야 합니다.
먼저 닉네임 길이, 이메일 길이, 비밀번호 길이에 대한 유효성 검사를 위해 from pydantic import Field 를 통해 Field 를 가져온 다음, 위처럼 추가 제약 조건을 설정해 줍시다.
테스트를 해 보면, 위처럼 유효성 검사 관련 에러는 잘 처리되네요. 이제 중복되는 이메일과 사용자명을 통해서 가입하지 않도록 합시다. 아래처럼 먼저 존재하는지 확인하고, 존재한다면 409 에러를 발생시키면 되겠어요.
@users_router.post(
"/users",
response_model=UserRegisterResponse,
status_code=status.HTTP_201_CREATED,
)
async def reigster_user(
session: Annotated[AsyncSession, Depends(get_session)],
register_data: UserRegisterRequest,
) -> Any:
user = User(
username=register_data.username,
email=register_data.email,
password=register_data.password,
)
email_exists = await session.execute(
select(exists().where(User.email == register_data.email))
)
if email_exists.scalar():
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email already exists",
)
username_exists = await session.execute(
select(exists().where(User.username == register_data.username))
)
if username_exists.scalar():
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Username already exists",
)
session.add(user)
await session.commit()
return user
.. 테스트 코드를 돌리면, 테스트 코드가 통과할 겁니다. 단 한 가지 테스트 케이스만 빼구요. 아래의 테스트가 실패하는 이유는 사용자가 입력한 비밀번호가 데이터베이스에 그대로 저장되었기 때문입니다.
비밀번호 암호화하여 저장하기
이제 비밀번호를 안전한 방법으로 암호화하여 저장해 보면 되겠네요.
안전한 암호화를 위해, 암호화는 직접 구현하는 것보다는 커뮤니티의 검증을 받은 타사 라이브러리를 사용하는 것이 좋습니다. 터미널에 poetry install bcrypt 를 입력하여, 검증 라이브러리인 bycrypt 를 설치합니다.
그리고, 아래와 같이 코드를 수정합니다.
async def reigster_user(
session: Annotated[AsyncSession, Depends(get_session)],
register_data: UserRegisterRequest,
) -> Any:
# imprt bcrypt 필수!
# 암호화
password_hash = bcrypt.hashpw(
register_data.password.encode("utf-8"), bcrypt.gensalt()
).decode("utf-8")
# 암호화한 데이터를 사용해 새로운 사용자 인스턴스 생성
user = User(
username=register_data.username,
email=register_data.email,
password=password_hash,
)
# 나머지 코드들..그리고, 테스트를 돌려 보세요. 성공할 겁니다.
드디어 성공입니다. 요구사항에 맞춘, 테스트 코드를 통과하는 회원가입을 구현했습니다.
아마 여기까지 구현하신 분들은 코드의 품질에 대해서 불만을 가지고 계셨을 지도 모르겠습니다. 아마 “routers.py 에서의 하나의 함수가 너무나 많은 책임을 가지고 있는 것이 아닌가?” 가 주된 불평의 이유일 것입니다. 다음 시간에는, 이렇게 작성된 테스트 코드로 수시로 우리의 코드를 검증하며, 코드를 조금 더 큰 애플리케이션에 걸맞도록 리팩토링하는 시간을 가지겠습니다.