[십자말풀이] PyTest를 활용한 인증, 퍼즐 서비스 테스트코드 작성

작업 레포지토리 : https://github.com/fnzksxl/word-puzzle

예비군 동미참 다녀오느라 업로드가 조금 느렸습니다..
PyTest 라이브러리를 활용해 이 때까지 개발했던 서비스 코드들에 대한 테스트 코드를 작성해보겠습니다.

서비스 코드에 몇 가지 변경사항이 있으니, 그대로 따라해보실 분들께서는 레포지토리의 test-puzzle_auth 브랜치를 참고해주시길 바랍니다.

1. Pytest 세팅

pip install pytest-asyncio

pytest-asyncio 라이브러리를 설치하면 비동기로 테스트를 진행할 수 있습니다.

테스트 폴더 구조

ROOT
ㄴ test/
  ㄴ __init__.py
  ㄴ conftest.py
    ㄴ domain1/
      ㄴ conftest.py
      ㄴ ...
    ㄴ domain2/
      ㄴ conftest.py
      ㄴ ...
    
    ...

위와 같은 형식으로 테스트 코드를 추가해나가겠습니다.

Test 환경 설정

테스트 할 때 실사용중인 DB에서 진행하면 안 되겠죠?
그리고 각 테스트 케이스에 독립성을 부여하기 위해 Test DB를 따로 사용하겠습니다.

# app/config.py

# Test Settings, Settings Class에 추가
TESTING = True if os.getenv("TESTING") == "true" else False
TEST_DB_NAME = os.getenv("TEST_DB_NAME")

# app/database.py

engine = create_engine(
    "mysql+pymysql://{username}:{password}@{host}:{port}/{name}".format(
        username=settings.DB_USERNAME,
        password=settings.DB_PASSWORD,
        host=settings.DB_HOST,
        port=settings.DB_PORT,
        name=settings.TEST_DB_NAME if settings.TESTING else settings.DB_NAME,
    )
)

테스트 중인지 아닌지를 서버세팅에 추가하고, 그 값에 따라 어떤 DB를 사용할 지 나누어주었습니다.

# test/conftest.py

@pytest_asyncio.fixture(scope="session")
def app():
    if not settings.TESTING:
        raise SystemError("Testing Environment setting must be 'TRUE'")

    return main.app


@pytest_asyncio.fixture
async def session():
    db = next(get_db())
    try:
        yield db
    finally:
        db.close()


@pytest_asyncio.fixture
async def client(app):
    async with AsyncClient(app=app, base_url="http://test/api/v1") as ac:
        models.Base.metadata.drop_all(bind=engine, tables=tables_to_drop)
        models.Base.metadata.create_all(bind=engine)
        yield ac

1. 테스트 설정이 아니라면 pytest를 실행시켜도 SystemError가 발생
2. session fixture로 test db에 접근
3. client fixture로 각 테스트케이스마다 DB를 초기화 시켜 독립성 유지

app fixture의 scope가 "session"으로 pytest를 실행할 때 최초 호출 후 캐시에서 불러오기 때문에 한 번만 실행된다는 점이 눈여겨볼만한 요소입니다.

2. Puzzle Test Case 작성

퍼즐 생성 테스트

퍼즐 생성 엔드포인트를 테스트하기 위해서는 WordInfo(단어정보) 테이블에 데이터가 들어가 있어야합니다.

하지만 우리가 위에서 작성한 conftest의 client fixture로 테스트를 진행하면 늘 DB가 초기화 된 상태로 진행되기 때문에 데이터가 들어있지 않습니다.

가장 간단해보이는 방법으로는 wordinfo fixture를 만들어 Test DB에 단어 데이터를 넣어주는 수가 있습니다만, 이 방법을 사용할 시에는 삽입해주는 데이터 양에 따라서 테스트 케이스 하나에 많은 비용이 들어갈 것 입니다.

단어 데이터를 모킹해주는 방법은 어떨까요?
테이블의 구조와 로직을 생각해보면 퍼즐 생성 후 저장하는 과정에서
WordInfo.id가 외래키로 걸려있으므로 테이블에 데이터를 삽입하는 과정이 불가피합니다.

그래서 어떻게 시간을 줄일거냐?

나름대로의 꼼수를 부려봤습니다.
현재 로직을 생각해보면 WordInfo 테이블에 단어 데이터를 정제하면서 삽입한 이후에 데이터의 수정을 요구하지 않습니다.

따라서, WordInfo 테이블에 데이터 삽입을 한번만 실행하고
client에서 WordInfo의 데이터를 드랍시키지 않는 방법을 떠올렸습니다.

Pytest를 시작했을 때 TEST DB에 WordInfo 데이터 유무에 따라
데이터 삽입을 결정하도록 하는 로직을 그림으로 표현해봤습니다.
아래 코드는 위 그림을 구현한 것입니다.

pip install pandas

단어 데이터를 csv로 관리하고 DB에 추가하기 위해 pandas 라이브러리를 설치해줬습니다.

test/conftest.py

@pytest_asyncio.fixture(scope="session")
async def inserted_wordinfo_into_db(app):
    with engine.connect() as connection:
        result = connection.execute(text("SELECT COUNT(*) FROM wordinfo"))
        row_count = result.scalar()

        if row_count > 494047:
            connection.execute(text("DELETE FROM wordinfo"))
            connection.execute(text("ALTER TABLE wordinfo AUTO_INCREMENT = 1"))
        if row_count == 0:
            df = pd.read_csv("test/example/wordinfo_backup.csv")
            df.to_sql("wordinfo", con=engine, index=False, if_exists="append")


@pytest_asyncio.fixture
async def client(app, inserted_wordinfo_into_db):
    async with AsyncClient(app=app, base_url="http://test/api/v1") as ac:
        tables_to_drop = [
            table for table in models.Base.metadata.sorted_tables if table.name != "wordinfo"
        ]

        models.Base.metadata.drop_all(bind=engine, tables=tables_to_drop)
        models.Base.metadata.create_all(bind=engine)
        yield ac

단어 데이터 확인 fixture의 scope를 session으로 설정해 이 후 다른 테스트 케이스에서 client를 불러도 데이터 확인 로직이 한 번만 실행하도록 구성했습니다.

또한, client fixture에서 drop하는 테이블에서 WordInfo 테이블만 제외시켜 데이터를 보존하는 동시에 각 테스트 케이스의 독립성을 유지해줬습니다.

이제 테스트 케이스를 작성해봅시다.

# test/puzzle/test_puzzle.py

@pytest.mark.asyncio
async def test_create_puzzle(client):
    r = await client.get("/puzzle")
    data = r.json()

    assert r.status_code == 200
    assert type(data.get("map", None)) == list


@pytest.mark.asyncio
async def test_create_puzzle_error_with_size(client):
    r = await client.get("/puzzle?size=11")

    assert r.status_code == 422

터미널창에서 pytest -v를 입력해 확인해본 결과 잘 나왔습니다.

퍼즐 조회 테스트

퍼즐 조회 테스트를 하기 위해서는 퍼즐 데이터가 존재해야겠죠?
puzzle fixture를 작성해줍시다.

# test/puzzle/conftest.py

@pytest_asyncio.fixture
async def puzzle(session):
    with open("test/example/example_map.json", "r", encoding="utf-8") as f:
        json_data = json.load(f)
    _map = json_data.get("map", None)
    _answers = json_data.get("desc", None)

    map_row = models.Puzzle(puzzle=_map)
    session.add(map_row)
    session.flush()

    insert_data = [
        {"puzzle_id": map_row.id, "word_id": desc["id"], "num": desc["num"]} for desc in _answers
    ]

    session.bulk_insert_mappings(models.PuzzleAnswer, insert_data)
    session.commit()

퍼즐 샘플이 담긴 json 파일을 불러와 TEST DB에 저장하는 fixture입니다.

@pytest.mark.asyncio
async def test_get_puzzle(client, puzzle):
    r = await client.get("/puzzle/1")

    assert r.status_code == 200


@pytest.mark.asyncio
async def test_get_puzzle_error_with_wrong_id(client):
    r = await client.get("/puzzle/2")

    assert r.status_code == 404

3. Auth Test Case 작성

일반 로그인

# test/auth/conftest.py

@pytest_asyncio.fixture
async def user(session):
    user_dict = {
        "email": "test@test.com",
        "nickname": "testname",
    }
    raw_password = "test1234"
    salt_value = bcrypt.gensalt()
    password = bcrypt.hashpw(raw_password.encode(), salt_value)

    user_dict.update({"password": password})

    user_row = models.User(**user_dict)
    session.add(user_row)
    session.commit()

    return user_row

conftest에 user fixture를 만들어줍시다.

# test/auth/test_auth.py

@pytest.mark.asyncio
async def test_general_register(client, session):
    body = {
        "email": "test@test.com",
        "password": "test1234",
        "nickname": "testname",
    }
    r = await client.post("/auth/general-register", data=json.dumps(body))

    assert r.status_code == 200

    user = session.query(models.User).filter(models.User.email == body.get("email")).first()

    assert body.get("email") == user.email
    assert body.get("nickname") == user.nickname

@pytest.mark.asyncio
async def test_general_login(client, user):
    body = {"email": "test@test.com", "password": "test1234"}
    r = await client.post("/auth/general-login", data=json.dumps(body))

    assert r.status_code == 200
    
@pytest.mark.asyncio
async def test_check_duplicated_email(client, user):
    email = "test@test.com"
    r = await client.get(f"/auth/duplicated?email={email}")
    data = r.json()

    assert r.status_code == 200
    assert data.get("is_duplicated")

일반 회원가입과 로그인 테스트 및 중복 이메일 테스트 코드입니다.
이제 응답에 딸려온 토큰이 잘 생성되었는지 테스트 해봅시다.

# test/auth/conftest.py

@pytest_asyncio.fixture
async def token(user):
    jwt_service = JWTService()
    user_dict = user.as_dict()
    user_dict.pop("password")
    user_dict.pop("created_at")
    user_dict.pop("updated_at")
    access_token = jwt_service.create_access_token(user_dict)

    return access_token

user fixture를 주입받아 token을 생성하는 fixture입니다.

# test/auth/test_auth.py

@pytest.mark.asyncio
async def test_get_user_by_token(client, token):
    cookies = {"access": token}
    r = await client.get("/auth/get-user", cookies=cookies)
    data = r.json()

    assert r.status_code == 200
    assert data.get("email", None) == "test@test.com"

토큰 정보로 유저 정보를 반환하는 테스트 코드입니다.

소셜 로그인

소셜 로그인은 일반 로그인과 다르게 외부 API를 이용하는 로직이 있습니다.
이는 모킹을 이용해 테스트해보겠습니다.

파이썬에서 기본으로 제공하는 unittest 라이브러리의 patch 기능을 사용할 수 있습니다.

테스트 할 엔드포인트의 함수를 모킹하여 원하는 동작을 하도록 바꿔주는 기능이 patch입니다.

# test/auth/mock.py

async def mock_post_token_request(*args, **kwargs):
    return {"access_token": "mock_access_token"}

async def mock_get_userinfo(*args, **kwargs):
    return {"email": "test@test.com", "name": "testuser"}

위 함수는 구글 API에서 Access Token을 받아오는 메소드,
아래 함수는 구글 API에서 유저정보를 받아오는 메소드를 모킹할 함수입니다.

@patch("app.api.v1.auth.service.GoogleOAuthService.get_token", new=mock.mock_post_token_request)
@patch("app.api.v1.auth.service.GoogleOAuthService.get_userinfo", new=mock.mock_get_userinfo)

테스트 코드에 위와 같은 데코레이터를 붙여주면 모킹 기능이 작동합니다.

# test/auth/conftest.py

@pytest_asyncio.fixture
async def oauth_google_user(session):
    user = models.User(email="test@test.com", nickname="test")
    session.add(user)
    session.flush()

    oauth_info = models.OAuth(user_id=user.id, email=user.email, provider="google")
    session.add(oauth_info)
    session.commit()

소셜 로그인 유저 fixture를 작성한 뒤 테스트 코드도 작성해봅시다.

# test/auth/test_auth.py

@pytest.mark.asyncio
@patch("app.api.v1.auth.service.GoogleOAuthService.get_token", new=mock.mock_post_token_request)
@patch("app.api.v1.auth.service.GoogleOAuthService.get_userinfo", new=mock.mock_get_userinfo)
async def test_google_register(client, session):
    code = "dummy code"
    r = await client.get(f"/auth/oauth-register/google/callback?code={code}")

    assert r.status_code == 200

    user = session.query(models.User).filter(models.User.email == "test@test.com").first()

    assert "test" == user.nickname

    oauth_user = session.query(models.OAuth).filter(models.OAuth.id == user.id).first()

    assert oauth_user.email == user.email


@pytest.mark.asyncio
@patch("app.api.v1.auth.service.GoogleOAuthService.get_token", new=mock.mock_post_token_request)
@patch("app.api.v1.auth.service.GoogleOAuthService.get_userinfo", new=mock.mock_get_userinfo)
async def test_google_register_failed_by_duplicated_email(client, user):
    code = "dummmy code"
    r = await client.get(f"/auth/oauth-register/google/callback?code={code}")
    data = r.json()

    assert r.status_code == 400
    assert data.get("detail") == "중복된 이메일입니다."


@pytest.mark.asyncio
@patch("app.api.v1.auth.service.GoogleOAuthService.get_token", new=mock.mock_post_token_request)
@patch("app.api.v1.auth.service.GoogleOAuthService.get_userinfo", new=mock.mock_get_userinfo)
async def test_google_login(client, oauth_google_user):
    code = "dummmy code"
    r = await client.get(f"/auth/oauth-register/google/callback?code={code}")
    data = r.json()

    assert r.status_code == 200
    assert data.get("email") == "test@test.com"
    assert data.get("nickname") == "test"

테스트가 잘 되는 군요.