[udemy] Try FastAPI Test Driven Development 2024 - 학습 정리

- 강좌 정보 : https://www.udemy.com/course/try-fastapi-api-test-driven-development/

 

1. vscode 확장 - Ruff

   1) 빠르고 강력한 Python Linter & Formatter

      - Python용 초고속 Linter 및 Formatter로, Rust로 작성되어 기존 도구들보다 훨씬 빠른 성능을 제공한다.

 

   2) Ruff의 주요 특징

      1. 빠른 속도

         - 기존 Python 기반 도구들보다 10~100배 이상 빠름
         - Rust로 작성되어 다중 스레드를 활용 가능

      2. Lint & Format 지원

         - Flake8, pyflakes, pycodestyle, mccabe 등 다양한 룰을 지원
         - ruff format을 통해 Black 스타일의 코드 포맷팅 가능

      3. 다양한 플러그인 내장

         - isort (import 정렬)
         - pyupgrade (자동 코드 업그레이드)
         - eradicate (사용하지 않는 코드 제거)
         - flake8-bugbear, flake8-comprehensions, pandas-vet 등 지원

      4. 자동 수정 기능

         - ruff check --fix 명령으로 자동 코드 수정 가능

         - Lint 오류를 수정하는 동시에 코드 스타일도 개선 가능

      5. 설정 파일 지원

         - pyproject.toml, .ruff.toml, ruff.json 등 다양한 설정 방식 제공
         - 프로젝트별로 세부 설정 가능

 

   3) 설치 및 사용법

      1. 설치

pip install ruff  # pip 설치

 

      2. 기본 사용법

ruff check .  # 현재 디렉토리 내 코드 검사
ruff check --fix  # 자동 수정 적용

 

      3. Formatter 사용

ruff format .  # 코드 자동 포맷팅

 

      4. 설정 파일 (pyproject.toml) 예제

[tool.ruff]
line-length = 88
select = ["E", "F", "I"]
ignore = ["E501"]  # 특정 규칙 무시

 

   4) Ruff vs 기존 도구 (Flake8, Black, isort) 비교

      - ✅ = 지원, ❌ = 미지원, 🚀 = 매우 빠름

기능	Ruff	Flake8	Black	isort
Lint	✅	✅	❌	❌
Format	✅	❌	✅	✅
자동 수정	✅	❌	일부	✅
속도	🚀 빠름	🐢 느림	🐢 느림	🐢 느림
Rust 기반	✅	❌	❌	❌

 

   5) Ruff을 써야 하는 이유

      - Flake8 + isort + Black을 하나로 통합 가능
      - Python보다 Rust 기반이라 압도적으로 빠름
      - Lint & Format을 한 번에 적용 가능
      - 설정이 간편하고 사용법이 직관적

 

2. Docker Compose의 environment와 env_file 정리

   - Docker Compose에서 컨테이너 환경 변수를 설정하는 방법은 크게 두 가지가 있다.

 

   1) environment 옵션 (직접 정의)

      - docker-compose.yml 파일에서 직접 환경 변수를 정의하는 방법

services:
  app:
    image: myapp:latest
    environment:
      - DEBUG=true
      - DATABASE_URL=postgres://user:password@db:5432/mydb

 

      - ✅ 장점
         - 설정이 직관적이고 즉시 적용 가능
         - 개별 컨테이너마다 환경 변수를 따로 설정 가능
      - ❌ 단점
         - 환경 변수가 많아지면 docker-compose.yml이 지저분해질 수 있음
         - .env 파일처럼 쉽게 변경하거나 관리하기 어려움

 

   2) env_file 옵션 (.env 파일 사용)

      - 환경 변수를 별도의 .env 파일에 정의하고 docker-compose.yml에서 불러오는 방법

 

      - 예제 (.env 파일)

DEBUG=true
DATABASE_URL=postgres://user:password@db:5432/mydb
SECRET_KEY=supersecret

 

      - 예제 (docker-compose.yml에서 env_file 사용)

services:
  app:
    image: myapp:latest
    env_file:
      - .env

 

      - ✅ 장점
         - .env 파일을 사용해 환경 변수를 쉽게 관리 가능
         - 보안성을 위해 코드 저장소(Git)에 .env 파일을 추가하지 않을 수도 있음
         - 여러 서비스에서 같은 환경 변수를 공유할 때 유용
      - ❌ 단점
         - .env 파일에 민감한 정보(비밀번호, API 키 등)를 저장하면 보안 위험이 있을 수 있음
         - → 해결책: .env 파일을 .gitignore에 추가하여 Git에 커밋하지 않도록 설정

 

   3) .env 파일을 사용하여 postgresql과 app의 환경 변수를 설정하는 방법

      - FastAPI + PostgreSQL을 사용하는 프로젝트에서 환경 변수를 .env 파일로 관리하고 docker-compose.yml에서 env_file을 활용하여 설정

 

      - .env 파일 생성

POSTGRES_USER=myuser
POSTGRES_PASSWORD=mypassword
POSTGRES_DB=mydatabase
DEBUG=true

 

      - docker-compose.yml에서 env_file 적용

version: '3.8'

services:
  db:
    image: postgres:15
    env_file:
      - .env
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  app:
    image: my-fastapi-app:latest
    depends_on:
      - db
    env_file:
      - .env
    environment:
      - DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB}
    ports:
      - "8000:8000"

volumes:
  postgres_data:

 

3. Docker 라이브러리 사용

   - 별도 정리된 포스트 : https://devspoon.tistory.com/317

파이썬 기반 docker 라이브러리 정리

   - 강의 내 프로젝트 파일

import os
import time

import docker


def is_container_ready(container):
    """
    주어진 Docker 컨테이너가 실행 중인지 확인하는 함수입니다.
    
    :param container: 확인할 Docker 컨테이너 객체
    :return: 컨테이너가 실행 중이면 True, 그렇지 않으면 False
    """
    container.reload()  # 컨테이너의 상태를 새로 고침
    return container.status == "running"  # 컨테이너 상태가 'running'인지 확인


def wait_for_stable_status(container, stable_duration=3, interval=1):
    """
    컨테이너가 안정적인 상태가 될 때까지 대기하는 함수입니다.
    
    :param container: 안정성을 확인할 Docker 컨테이너 객체
    :param stable_duration: 안정성을 확인할 시간 (초)
    :param interval: 상태 확인 간격 (초)
    :return: 컨테이너가 안정적인 상태가 되면 True, 그렇지 않으면 False
    """
    start_time = time.time()  # 시작 시간 기록
    stable_count = 0  # 안정적인 상태 카운트 초기화
    while time.time() - start_time < stable_duration:  # 지정된 시간 동안 반복
        if is_container_ready(container):  # 컨테이너가 준비되었는지 확인
            stable_count += 1  # 안정적인 상태 카운트 증가
        else:
            stable_count = 0  # 안정적이지 않으면 카운트 초기화

        if stable_count >= stable_duration / interval:  # 안정적인 상태가 일정 수치에 도달하면
            return True  # True 반환

        time.sleep(interval)  # 지정된 간격만큼 대기
    return False  # 안정적인 상태가 되지 않으면 False 반환


def start_database_container():
    """
    데이터베이스 컨테이너를 시작하는 함수입니다.
    
    :return: 시작된 컨테이너 객체
    """
    client = docker.from_env()  # Docker 클라이언트 생성
    scripts_dir = os.path.abspath("./scripts")  # 스크립트 디렉토리의 절대 경로
    container_name = "test-db"  # 컨테이너 이름 정의

    try:
        existing_container = client.containers.get(container_name)  # 기존 컨테이너 가져오기
        print(f"Container '{container_name} exists. Stopping and removing...")  # 컨테이너 존재 메시지 출력
        existing_container.stop()  # 기존 컨테이너 중지
        existing_container.remove()  # 기존 컨테이너 제거
        print((f"Container '{container_name} stopped and removed"))  # 중지 및 제거 메시지 출력
    except docker.errors.NotFound:
        print(f"Container '{container_name}' does not exist.")  # 컨테이너가 존재하지 않으면 메시지 출력

    # 컨테이너 구성 정의
    container_config = {
        "name": container_name,
        "image": "postgres:16.1-alpine3.19",  # 사용할 Docker 이미지
        "detach": True,  # 백그라운드에서 실행
        "ports": {"5432": "5434"},  # 포트 매핑
        "environment": {
            "POSTGRES_USER": "postgres",  # PostgreSQL 사용자
            "POSTGRES_PASSWORD": "postgres",  # PostgreSQL 비밀번호
        },
        "volumes": [f"{scripts_dir}:/docker-entrypoint-initdb.d"],  # 초기화 스크립트 볼륨
        "network_mode": "fastapi-development_dev-network",  # 네트워크 모드 설정
    }

    # 컨테이너 시작
    container = client.containers.run(**container_config)  # 컨테이너 실행

    while not is_container_ready(container):  # 컨테이너가 준비될 때까지 대기
        time.sleep(1)

    if not wait_for_stable_status(container):  # 안정적인 상태가 아닐 경우 예외 발생
        raise RuntimeError("Container did not stabilize within the specified time")

    return container  # 시작된 컨테이너 반환

 

4. Configuring Alembic for Multi-Database

   1) alembic.ini에 테스트 db 설정

      - [alembic] 기본 섹션 헤더 안에 각각의 db 섹션 헤더를 만든다.

      - url은 env.py에서 .env를 읽어 어플리케이션이 실행될 때 정의하도록 한다.

[alembic]

[devdb]

script_location = migrations
prepend_sys_path = .
version_path_separator = os
sqlalchemy.url = 

[testdb]

script_location = migrations
prepend_sys_path = .
version_path_separator = os
sqlalchemy.url = 
...

 

   2) alembic의 env.py 설정

      - alembic.ini의 url은 env.py에서 .env를 읽어 어플리케이션이 실행될 때 정의하도록 한다.

...
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Set the database URLs based on environment variables or any other source
url_db1 = os.environ.get("DEV_DATABASE_URL")
url_db2 = os.environ.get("TEST_DATABASE_URL")

# Modify the database URLs in the Alembic config
config.set_section_option("devdb", "sqlalchemy.url", url_db1)
config.set_section_option(
    "testdb", "sqlalchemy.url", os.environ.get("TEST_DATABASE_URL")
)

target_metadata = models.Base.metadata
...

 

   3) alembic의 command를 이용한 지정된 리비전으로 데이터베이스 업그레이드

import alembic.config
from alembic import command


def migrate_to_db(
    script_location, alembic_ini_path="alembic.ini", connection=None, revision="head"
):
    """
    데이터베이스로 마이그레이션을 수행하는 함수입니다.
    
    :param script_location: 마이그레이션 스크립트의 위치
    :param alembic_ini_path: Alembic 설정 파일의 경로 (기본값: "alembic.ini")
    :param connection: 데이터베이스 연결 객체 (기본값: None)
    :param revision: 마이그레이션할 리비전 (기본값: "head")
    """
    config = alembic.config.Config(alembic_ini_path)  # Alembic 설정 객체 생성
    if connection is not None:  # 연결 객체가 제공된 경우
        config.config_ini_section = "testdb"  # 테스트 데이터베이스 섹션 설정
        command.upgrade(config, revision)  # 지정된 리비전으로 데이터베이스 업그레이드
        # "head"는 가장 최신의 리비전을 의미

 

   4) docker 라이브러리를 기반으로 컨테이너 제어 함수 정의

import os
import time

import docker


def is_container_ready(container):
    """
    주어진 Docker 컨테이너가 실행 중인지 확인하는 함수입니다.
    
    :param container: 확인할 Docker 컨테이너 객체
    :return: 컨테이너가 실행 중이면 True, 그렇지 않으면 False
    """
    container.reload()  # 컨테이너의 상태를 새로 고침
    return container.status == "running"  # 컨테이너 상태가 'running'인지 확인


def wait_for_stable_status(container, stable_duration=3, interval=1):
    """
    컨테이너가 안정적인 상태가 될 때까지 대기하는 함수입니다.
    
    :param container: 안정성을 확인할 Docker 컨테이너 객체
    :param stable_duration: 안정성을 확인할 시간 (초)
    :param interval: 상태 확인 간격 (초)
    :return: 컨테이너가 안정적인 상태가 되면 True, 그렇지 않으면 False
    """
    start_time = time.time()  # 시작 시간 기록
    stable_count = 0  # 안정적인 상태 카운트 초기화
    while time.time() - start_time < stable_duration:  # 지정된 시간 동안 반복
        if is_container_ready(container):  # 컨테이너가 준비되었는지 확인
            stable_count += 1  # 안정적인 상태 카운트 증가
        else:
            stable_count = 0  # 안정적이지 않으면 카운트 초기화

        if stable_count >= stable_duration / interval:  # 안정적인 상태가 일정 수치에 도달하면
            return True  # True 반환

        time.sleep(interval)  # 지정된 간격만큼 대기
    return False  # 안정적인 상태가 되지 않으면 False 반환


def start_database_container():
    """
    데이터베이스 컨테이너를 시작하는 함수입니다.
    
    :return: 시작된 컨테이너 객체
    """
    client = docker.from_env()  # Docker 클라이언트 생성
    scripts_dir = os.path.abspath("./scripts")  # 스크립트 디렉토리의 절대 경로
    container_name = "test-db"  # 컨테이너 이름 정의

    try:
        existing_container = client.containers.get(container_name)  # 기존 컨테이너 가져오기
        print(f"Container '{container_name} exists. Stopping and removing...")  # 컨테이너 존재 메시지 출력
        existing_container.stop()  # 기존 컨테이너 중지
        existing_container.remove()  # 기존 컨테이너 제거
        print((f"Container '{container_name} stopped and removed"))  # 중지 및 제거 메시지 출력
    except docker.errors.NotFound:
        print(f"Container '{container_name}' does not exist.")  # 컨테이너가 존재하지 않으면 메시지 출력

    # 컨테이너 구성 정의
    container_config = {
        "name": container_name,
        "image": "postgres:16.1-alpine3.19",  # 사용할 Docker 이미지
        "detach": True,  # 백그라운드에서 실행
        "ports": {"5432": "5434"},  # 포트 매핑
        "environment": {
            "POSTGRES_USER": "postgres",  # PostgreSQL 사용자
            "POSTGRES_PASSWORD": "postgres",  # PostgreSQL 비밀번호
        },
        "volumes": [f"{scripts_dir}:/docker-entrypoint-initdb.d"],  # 초기화 스크립트 볼륨
        "network_mode": "fastapi-development_dev-network",  # 네트워크 모드 설정
    }

    # 컨테이너 시작
    container = client.containers.run(**container_config)  # 컨테이너 실행

    while not is_container_ready(container):  # 컨테이너가 준비될 때까지 대기
        time.sleep(1)

    if not wait_for_stable_status(container):  # 안정적인 상태가 아닐 경우 예외 발생
        raise RuntimeError("Container did not stabilize within the specified time")

    return container  # 시작된 컨테이너 반환

 

   5) pytest를 위한 fixtures.py에서 db_session 정의

      - start_database_container 함수를 이용하여 컨테이너 실행

      - create_engine 함수를 이용하여 데이터베이스 연결을 위한 환경

      - migrate_to_db 함수를 이용하여 마이그레이션 실행

def db_session():
    container = start_database_container()

    engine = create_engine(os.getenv("TEST_DATABASE_URL"))

    with engine.begin() as connection:
        migrate_to_db("migrations", "alembic.ini", connection)

    SessionLocal = sessionmaker(autocommit=False, autoflush=True, bind=engine)

    yield SessionLocal

    # container.stop()
    # container.remove()
    engine.dispose()

 

5. table의 __table_args__ 정의

   - table 정의

class Category(Base):
    __tablename__ = "category"

    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(String(100), nullable=False)
    slug = Column(String(120), nullable=False)
    is_active = Column(Boolean, nullable=False, default=False, server_default="False")
    level = Column(Integer, nullable=False, default="100", server_default="100")
    parent_id = Column(Integer, ForeignKey("category.id"), nullable=True)

    __table_args__ = (
        CheckConstraint("LENGTH(name) > 0", name="category_name_length_check"),
        CheckConstraint("LENGTH(slug) > 0", name="category_slug_length_check"),
        UniqueConstraint("name", "level", name="uq_category_name_level"),
        UniqueConstraint("slug", name="uq_category_slug"),
    )

 

   - 상위 테이블이 생성될 때에는 아래와 같은 제약조건이 정의된 쿼리가 생성된다.

CREATE TABLE category (
    id INTEGER NOT NULL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    slug VARCHAR(120) NOT NULL,
    is_active BOOLEAN NOT NULL DEFAULT FALSE,
    level INTEGER NOT NULL DEFAULT 100,
    parent_id INTEGER,
    CONSTRAINT category_name_length_check CHECK (LENGTH(name) > 0),
    CONSTRAINT category_slug_length_check CHECK (LENGTH(slug) > 0),
    CONSTRAINT uq_category_name_level UNIQUE (name, level),
    CONSTRAINT uq_category_slug UNIQUE (slug)
);

 

6. default와 server_default의 차이점

   1) default:

      - 클라이언트 측 기본값:

         - default는 SQLAlchemy(Python) 레이어에서 기본값을 설정한다.

 

      - 동작 방식:

         - 새로운 레코드를 추가할 때, 해당 컬럼에 값이 제공되지 않으면, SQLAlchemy가 지정된 default 값을 사용하여 INSERT 문을 생성한다.

from sqlalchemy import Column, Integer, String, func
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String, default='Anonymous')

 

      - 위의 예시에서, name 컬럼에 값이 제공되지 않으면, SQLAlchemy는 'Anonymous'를 기본값으로 사용하여 INSERT 문을 생성한다.

 

   2) server_default:

      - 서버 측 기본값:

         - server_default는 데이터베이스 레벨에서 기본값을 설정한다.

      - 동작 방식:

         - 테이블 생성 시, 해당 컬럼에 대한 기본값이 데이터베이스 스키마에 정의된다. 이후 해당 컬럼에 값이 제공되지 않으면, 데이터베이스가 지정된 기본값을 자동으로 적용한다.

from sqlalchemy import Column, Integer, String, func, text
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    created_at = Column(
        DateTime,
        server_default=func.now()
    )

 

      - 위의 예시에서, created_at 컬럼에 값이 제공되지 않으면, 데이터베이스는 NOW() 함수를 사용하여 현재 시간을 기본값으로 설정한다.

 

   3) 주요 차이점:

      - 적용 위치:

         - default는 SQLAlchemy(Python) 레이어에서 적용되며, server_default는 데이터베이스 레벨에서 적용된다.

         - 스키마 반영: default는 데이터베이스 스키마에 기본값을 설정하지 않지만, server_default는 테이블 생성 시 스키마에 기본값을 포함시킨다.

         - 기존 데이터 처리: 새로운 컬럼을 추가할 때, server_default를 사용하면 기존 레코드에 대해 데이터베이스가 자동으로 기본값을 설정할 수 있다. 반면, default는 새로운 레코드에만 적용된다.

 

   4) 주의사항:

      - server_default를 설정할 때, 값은 문자열로 지정해야 한다.

      - default와 server_default를 동시에 설정하면, SQLAlchemy가 default 값을 사용하여 INSERT 문을 생성하므로, 데이터베이스의 server_default는 적용되지 않을 수 있다.

 

7. postgresql의 :: 연산자

   - :: 연산자는 형 변환을 수행하는 데 사용된다. 이는 특정 데이터 값을 원하는 데이터 타입으로 변환할 때 활용된다.

 

   - 문자열을 정수로 변환:

SELECT '123'::integer;

 

   - 날짜 형식의 문자열을 DATE 타입으로 변환:

SELECT '2025-02-11'::DATE;

 

   - enum을 사용해 열거형으로 정의된 stock_status

...
stock_status = Column(
    Enum("oos", "is", "obo", name="status_enum"),
    nullable=False,
    server_default="oos",
)
...

 

   - pytest에서 database의 테이블 정보를 가져와 비교하는 경우 

      - columns["stock_status"]["default"] == "'oos'::status_enum"는 stock_status 컬럼의 기본값이 'oos' 문자열을 status_enum이라는 사용자 정의 열거형(enum) 타입으로 변환한 값과 일치하는지 확인하는 것이다. 즉, 'oos'라는 문자열을 status_enum 타입으로 캐스팅하여 해당 컬럼의 기본값으로 설정했음을 의미한다.

def test_model_structure_default_values(db_inspector):
    table = "product"
    columns = {columns["name"]: columns for columns in db_inspector.get_columns(table)}

    assert columns["is_digital"]["default"] == "false"
    assert columns["is_active"]["default"] == "false"
    assert columns["stock_status"]["default"] == "'oos'::status_enum"

 

8. created_at, updated_at 정의

   - SQLAlchemy의 text() 함수는 원시 SQL 문을 직접 작성하여 데이터베이스와 상호작용할 수 있도록 하는 기능을 제공한다. 이를 통해 ORM이나 SQL 표현식 언어를 사용하지 않고도 복잡한 SQL 쿼리나 특정 데이터베이스 기능을 활용할 수 있다.

created_at = Column(
        DateTime, server_default=text("CURRENT_TIMESTAMP"), nullable=False
    )
updated_at = Column(
    DateTime,
    server_default=text("CURRENT_TIMESTAMP"),
    onupdate=sqlalchemy.func.now(),
    nullable=False,
)

 

9. set을 이용한 비교

   - set을 사용하면 요소의 순서가 달라도 같은 집합으로 인식된다.

set_1 = set(["category_id"])
set_2 = set(["category_id"])
set_3 = set(["seasonal_id"])
set_4 = set(["seasonal_id", "category_id"])
set_5 = set(["category_id", "seasonal_id"])

print(set_1 == set_2)  # True
print(set_3 == set_4)  # False
print(set_4 == set_5)  # True  (순서가 달라도 동일한 집합으로 인식)

 

   1) 프로젝트 코드 분석

      - table 컬럼

...
category_id = Column(Integer, ForeignKey("category.id"), nullable=False)
seasonal_id = Column(Integer, ForeignKey("seasonal_event.id"), nullable=True)
...

 

      - pytest 테스트 함수

def test_model_structure_foreign_key(db_inspector):
    table = "product"
    foreign_keys = db_inspector.get_foreign_keys(table)
    product_foreign_key = next(
        (
            fk
            for fk in foreign_keys
            if set(fk["constrained_columns"]) == {"category_id"}
            or set(fk["constrained_columns"]) == {"seasonal_id"}
        ),
        None,
    )
    assert product_foreign_key is not None

 

      - set(fk["constrained_columns"]) 구조

# category_id
{
    "name": "fk_product_category",
    "constrained_columns": ["category_id"],
    "referred_table": "category",
    "referred_columns": ["id"]
}

# seasonal_id
{
    "name": "fk_product_seasonal_event",
    "constrained_columns": ["seasonal_id"],
    "referred_table": "seasonal_event",
    "referred_columns": ["id"]
}

 

      - set(fk["constrained_columns"])의 결과

         - 각 외래 키에 대해 set(fk["constrained_columns"])를 호출하면, 각 컬럼이 집합(set)으로 변환된다.

         - set()은 순서에 관계없이 유일한 값들을 반환한다.

# 첫 번째 외래 키 (category_id)
set(fk["constrained_columns"])  # {'category_id'}

# 두 번째 외래 키 (seasonal_id)
set(fk["constrained_columns"])  # {'seasonal_id'}

 

      - set()을 사용하는 이유

         1. 복합 외래 키 처리:

            - constrained_columns는 하나의 컬럼뿐만 아니라 여러 개의 컬럼을 외래 키로 가질 수도 있다.

"constrained_columns": ["category_id", "seasonal_id"]

 

            - set()은 리스트의 순서를 무시하고, 컬럼을 집합으로 다루기 때문에 복합 외래 키도 정상적으로 처리할 수 있다.

set(["category_id", "seasonal_id"])  # {'category_id', 'seasonal_id'}

 

         2. 중복 제거:

            - 만약 constrained_columns에 동일한 컬럼이 중복으로 들어있을 경우 set()을 사용하면 중복을 자동으로 제거할 수 있다.

"constrained_columns": ["category_id", "category_id"]

 

            - 이 리스트는 set()을 사용하면 중복된 "category_id"가 하나로 정리된다.

set(["category_id", "category_id"])  # {'category_id'}

 

10. Faker 사용

   - faker를 이용한 데이터 생성 함수

from faker import Faker

faker = Faker()

def get_random_category_dict(id_: int = None):
    return {
        "id": id_ or faker.random_int(1, 1000),
        "name": faker.word(),
        "slug": faker.slug(),
        "is_active": faker.boolean(),
        "level": faker.random_int(1, 20),
        "parent_id": None,
    }

 

   - 테스트 함수에서 사용

...
from app.models import Category
from tests.factories.models_factory import get_random_category_dict

def test_unit_delete_category_successfully(client, monkeypatch):
    category_dict = get_random_category_dict()
    category_instance = Category(**category_dict)
    ...

 

11. fastapi.testclient의 TestClient

   - TestClient는 FastAPI 애플리케이션의 엔드포인트를 테스트하기 위한 유용한 도구이다.

   - 이 클래스는 FastAPI의 TestClient 모듈에서 제공되며, HTTP 요청을 애플리케이션에 보내고 응답을 검사하는 데 사용된다.

 

   1) 주요 기능:

      - HTTP 요청 메서드 지원:
         - get(), post(), put(), delete(), options() 등 다양한 HTTP 메서드를 사용하여 요청을 보낼 수 있다.
      - 요청과 응답 검증:
         - 각 요청 메서드는 URL 및 선택적으로 JSON, 쿼리 매개변수, 헤더 등의 데이터를 포함할 수 있다.
         - 응답을 검사하고 상태 코드, JSON 데이터, 헤더 등을 확인할 수 있다.
      - 테스트 환경 설정:

         - 테스트에서 사용할 애플리케이션을 TestClient 인스턴스에 설정할 수 있다.
         - 일반적으로 테스트 시작 전에 애플리케이션을 설정하고, 테스트 후에는 정리(clean-up)하는 패턴을 따른다.
      - 예제 코드:

         - 아래는 간단한 FastAPI 애플리케이션을 TestClient를 사용하여 테스트하는 예제이다.

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

 

12. Monkeypatching

   1) 개요

      - Monkeypatching은 실행 중에 기존 클래스, 함수, 모듈, 객체의 동작을 수정하거나 확장하는 기법이다.

         - 원래 소스 코드를 수정하지 않고 런타임에서 동적으로 변경할 수 있으므로, 테스트 환경에서 외부 의존성(예: 네트워크 호출, 파일 시스템 접근, 환경 변수 등)을 격리하거나 임시로 수정할 때 유용하다.

         - 단, 너무 남발할 경우 코드의 가독성과 유지보수성이 떨어질 수 있으므로 신중하게 사용해야 합니다.

 

   2) 파이썬에서의 monkeypatch 정의 및 용도

      1. 정의

         - Monkeypatching은 실행 중인 프로그램의 객체(함수, 클래스, 모듈 등)에 대해 "덮어쓰기"를 수행하여 다른 동작을 하도록 변경하는 것을 말한다.
         - 예를 들어, 테스트 시 외부 API 호출 대신 모의(Mock) 데이터를 반환하도록 함수를 임시로 변경할 수 있다.

 

      2. 사용 용도

         - 테스트 격리: 실제 외부 자원(네트워크, DB 등)에 접근하지 않고도 원하는 동작을 검증할 수 있다.
         - 버그 수정(Hot Fix): 3자 라이브러리의 단기적 문제를 임시로 해결할 때 사용한다.
         - 동적 확장: 기존 라이브러리에 기능을 추가하거나 수정할 때 활용할 수 있다.

 

   3) pytest에서의 monkeypatch 사용법

      - pytest는 monkeypatch라는 내장 fixture를 제공하여 테스트 함수 내에서 안전하게 객체의 속성, 딕셔너리 항목, 환경 변수 등을 수정할 수 있도록 한다.

 

      - 주요 메서드 :
         - monkeypatch.setattr(target, name, value, raising=True)
            - 대상 객체의 속성을 원하는 값으로 설정한다.
         - monkeypatch.delattr(target, name, raising=True)
            - 대상 객체의 속성을 삭제한다.
         - monkeypatch.setitem(mapping, key, value)
            - 딕셔너리 등의 항목 값을 수정한다.
         - monkeypatch.delitem(mapping, key, raising=True)
            - 딕셔너리 항목을 삭제한다.
         - monkeypatch.setenv(name, value, prepend=None)
            - 환경 변수를 설정한다.
         - monkeypatch.delenv(name, raising=True)
            - 환경 변수를 삭제한다.
         - monkeypatch.syspath_prepend(path)
            - sys.path에 경로를 추가한다.

 

         - 이와 같은 메서드는 테스트가 종료되면 자동으로 원래 상태로 복원되므로 안전하게 사용할 수 있다.

 

   4) 코드 예제

      1. 함수 수정 예제

         - 아래는 pathlib.Path.home()을 monkeypatching하여 항상 특정 경로를 반환하도록 하는 예제이다.

from pathlib import Path

def get_ssh_dir():
    # 실제 홈 디렉토리에 .ssh를 붙여 경로를 생성
    return Path.home() / ".ssh"

def test_get_ssh(monkeypatch):
    # 임의의 경로를 반환하는 함수 정의
    def fake_home():
        return Path("/abc")
    
    # Path.home 메서드를 fake_home으로 대체
    monkeypatch.setattr(Path, "home", fake_home)
    
    # get_ssh_dir() 호출 시 fake_home이 호출되어 '/abc/.ssh'가 반환됨
    result = get_ssh_dir()
    assert result == Path("/abc/.ssh")

 

      2. 환경 변수 수정 예제

         - 다음은 환경 변수를 설정하고 삭제하는 예제이다.

import os
import pytest

def get_user_lower():
    username = os.getenv("USER")
    if username is None:
        raise OSError("USER 환경 변수가 설정되지 않음")
    return username.lower()

def test_get_user_lower(monkeypatch):
    # 환경 변수 USER를 "TestingUser"로 설정
    monkeypatch.setenv("USER", "TestingUser")
    assert get_user_lower() == "testinguser"

def test_missing_user(monkeypatch):
    # 환경 변수 USER 삭제
    monkeypatch.delenv("USER", raising=False)
    with pytest.raises(OSError):
        get_user_lower()

 

      3. 딕셔너리 항목 수정 예제

def get_connection_string(config):
    # 간단한 연결 문자열 생성 예제
    return f"User={config['user']};DB={config['database']}"

def test_connection_string(monkeypatch):
    # 테스트용 기본 설정 딕셔너리
    default_config = {"user": "admin", "database": "prod_db"}
    
    # monkeypatch를 사용하여 값을 변경
    monkeypatch.setitem(default_config, "user", "test_user")
    monkeypatch.setitem(default_config, "database", "test_db")
    
    conn_str = get_connection_string(default_config)
    assert conn_str == "User=test_user;DB=test_db"

 

13. mark.parametrize와 monkeypatch를 이용한 좋은 테스트 코드

   1) 테스트 코드

@pytest.mark.parametrize(
    "existing_category, category_data, expected_detail",
    [
        (True, get_random_category_dict(), "Category name and level already exists"),
        (True, get_random_category_dict(), "Category slug already exists"),
    ],
)
def test_unit_create_new_category_existing(
    client, monkeypatch, existing_category, category_data, expected_detail
):
    def mock_check_existing_category(db, category_data):
        if existing_category:
            raise HTTPException(status_code=400, detail=expected_detail)

    monkeypatch.setattr(
        "app.routers.category_routes.check_existing_category",
        mock_check_existing_category,
    )

    monkeypatch.setattr("sqlalchemy.orm.Query.first", mock_output())
    body = category_data.copy()
    body.pop("id")
    response = client.post("api/category/", json=body)

    assert response.status_code == 400

    if expected_detail:
        assert response.json() == {"detail": expected_detail}

 

   2) pytest.mark.parametrize의 설정과 동작

      - @pytest.mark.parametrize 데코레이터는 테스트 함수를 다양한 인자 조합으로 여러 번 실행할 수 있도록 해준다. 이를 통해 여러 테스트 케이스를 간결하게 작성할 수 있다.

 

      - 첫 번째 인자: "existing_category, category_data, expected_detail"는 테스트 함수에 전달될 파라미터 이름들을 문자열로 지정한다. 각 이름은 쉼표로 구분된다.

      - 두 번째 인자: 리스트 안에 튜플 형태로 각 테스트 케이스에 사용할 인자 값을 정의한다.
         - (True, get_random_category_dict(), "Category name and level already exists")
         - (True, get_random_category_dict(), "Category slug already exists")

 

      - 여기서 각 튜플은 다음과 같은 의미를 가진다.

         - existing_category: True로 설정되어, 기존 카테고리가 존재함을 나타낸다.
         - category_data: get_random_category_dict() 함수를 호출하여 생성된 임의의 카테고리 데이터이다.
         - expected_detail: 예상되는 에러 메시지로, 각각 다른 내용을 가진다.

@pytest.mark.parametrize(
    "existing_category, category_data, expected_detail",
    [
        (True, get_random_category_dict(), "Category name and level already exists"),
        (True, get_random_category_dict(), "Category slug already exists"),
    ],
)

 

   3) 테스트 함수의 파라미터 위치와 역할

      - client: 테스트 클라이언트로, HTTP 요청을 시뮬레이션하는 데 사용된다.
      - monkeypatch: 테스트 중 특정 객체나 함수를 임시로 대체하거나 수정할 수 있게 해주는 pytest의 픽스처이다.
      - existing_category: parametrize 데코레이터를 통해 전달된 부울 값으로, 기존 카테고리의 존재 여부를 나타낸다.
      - category_data: 테스트에 사용할 카테고리 데이터로, get_random_category_dict() 함수를 통해 생성된다.
      - expected_detail: 예상되는 에러 메시지로, 테스트의 예상 결과를 검증하는 데 사용된다.

def test_unit_create_new_category_existing(
    client, monkeypatch, existing_category, category_data, expected_detail
):
    ...

 

   4) mock_check_existing_category 함수와 관련된 코드의 동작

      - 인자:
         - db: 데이터베이스 세션 또는 연결 객체로 예상된다.
         - category_data: 카테고리 데이터로, 테스트 중 생성된 데이터를 받는다.

 

      - 동작:
         - existing_category가 True이면, HTTPException을 발생시켜 상태 코드 400과 expected_detail 메시지를 반환한다.
         - existing_category가 False인 경우, 아무 작업도 수행하지 않는다.

 

      - monkeypatch를 사용하여 실제 함수를 모의 함수로 대체한다

monkeypatch.setattr(
    "app.routers.category_routes.check_existing_category",
    mock_check_existing_category,
)

 

      - 이 코드는 app.routers.category_routes 모듈의 check_existing_category 함수를 mock_check_existing_category로 대체한다. 이를 통해 테스트 중 해당 함수 호출 시 실제 함수 대신 모의 함수가 호출된다.
      - 또한, 데이터베이스 쿼리의 결과를 제어하기 위해 다음과 같이 monkeypatch를 사용한다.

      - 이 코드는 SQLAlchemy의 Query 객체의 first 메서드를 mock_output 함수로 대체하여, 데이터베이스 쿼리의 결과를 제어한다.

monkeypatch.setattr("sqlalchemy.orm.Query.first", mock_output())

 

   5) 테스트의 전체적인 흐름

      - 파라미터 주입: pytest는 @pytest.mark.parametrize 데코레이터를 통해 정의된 각 인자 조합을 테스트 함수로 전달
      - 모의 함수 정의: mock_check_existing_category 함수를 정의하여, existing_category의 값에 따라 예외를 발생시키거나 아무 작업도 하지 않음
      - 함수 대체: monkeypatch를 사용하여 실제 check_existing_category 함수와 Query.first 메서드를 모의 함수로 대체
      - 테스트 실행:

         - category_data를 복사하고 "id" 키를 제거하여 요청 본문을 준비한다.
         - client를 사용하여 /api/category/ 엔드포인트에 POST 요청을 보낸다.
         - 응답 상태 코드가 400인지 확인한다.
         - 응답의 JSON 내용이 {"detail": expected_detail}과 일치하는지 확인한다.

 

14. Query에 monkeypatch를 연결하여 view 테스트를 하는 방법

def get_random_category_dict(id_: int = None):
    return {
        "id": id_ or faker.random_int(1, 1000),
        "name": faker.word(),
        "slug": faker.slug(),
        "is_active": faker.boolean(),
        "level": faker.random_int(1, 20),
        "parent_id": None,
    }

def mock_output(return_value=None):
    return lambda *args, **kwargs: return_value

def test_unit_get_all_categories_successfully(client, monkeypatch):
    category = [get_random_category_dict(i) for i in range(5)]
    monkeypatch.setattr("sqlalchemy.orm.Query.all", mock_output(category))
    response = client.get("api/category/")
    assert response.status_code == 200
    assert response.json() == category

 

15. logging.conf를 이용하여 멀티 로거 설정하기

   - 원하는 파일에 getLogger를 정의한다.

logger = logging.getLogger("app")

 

   - loggers 항목의 keys에 정의된 root는 logging.getLogger() 함수를 호출하여 명시적으로 이름을 지정하지 않고 로거를 생성할 때 사용되는 기본 로거이다. 같은 위치에 , 를 사용하여 app을 정의하고 logger_app과 handler_fileHandler_app을 정의한다.

[loggers]
keys=root, app

[handlers]
keys=consoleHandler, fileHandler, fileHandler_app

[formatters]
keys=simpleFormatter

[logger_root]
level=DEBUG
handlers=consoleHandler, fileHandler

[logger_app]
level=DEBUG
handlers=fileHandler_app
qualname=app

[handler_consoleHandler]
class=StreamHandler
level=DEBUG
formatter=simpleFormatter
args=(sys.stdout,)

[handler_fileHandler]
class=FileHandler
level=DEBUG
formatter=simpleFormatter
args=('dev.log',)

[handler_fileHandler_app]
class=FileHandler
level=DEBUG
formatter=simpleFormatter
args=('app.log',)

[formatter_simpleFormatter]
format=%(asctime)s - %(levelname)s - %(message)s

 

16. 소프트웨어 테스트

   1) 컴포넌트 통합 테스트 (Component Integration Testing)

      - 컴포넌트 통합 테스트는 개별 모듈이나 유닛이 아닌, 여러 유닛이나 컴포넌트가 결합되어 하나의 모듈이나 서브시스템으로 작동할 때 이들의 상호 작용을 검증하는 테스트이다.

      - 이 단계에서는 각 컴포넌트가 단독으로는 정상적으로 동작하더라도, 통합 시 예상치 못한 문제가 발생하지 않는지를 확인한다.

 

      - 주요 특징:

         - 초점: 여러 유닛이나 컴포넌트 간의 인터페이스와 데이터 흐름을 검증한다.
         - 범위: 하나의 모듈이나 서브시스템 내에서의 통합에 중점을 둔다.
         - 방법: 하향식(Top-Down), 상향식(Bottom-Up), 샌드위치(Sandwich) 등 다양한 접근 방식을 사용하여 통합 테스트를 수행한다.

 

   2) 시스템 통합 테스트 (System Integration Testing)

      - 시스템 통합 테스트는 통합된 여러 모듈이나 서브시스템이 전체 시스템으로서 잘 작동하는지를 검증하는 단계이다.         - 이 테스트는 내부 구성 요소뿐만 아니라 외부 시스템과의 상호 작용도 포함하여, 시스템 간의 인터페이스와 데이터 교환이 예상대로 이루어지는지를 확인한다.

      - 주요 특징:

         - 초점: 내부 모듈 간 및 외부 시스템과의 상호 작용을 검증한다.
         - 범위: 전체 시스템 수준에서의 통합에 중점을 둔다.
         - 방법: 실제 운영 환경과 유사한 조건에서 테스트를 수행하여, 시스템 간의 통합이 원활한지 확인한다.

 

   3) 엔드 투 엔드 통합 테스트 (End-to-End Integration Testing)

      - 엔드 투 엔드 통합 테스트는 사용자의 관점에서 시스템의 전체 워크플로우를 검증하는 테스트이다.

      - 시스템의 시작부터 끝까지 모든 기능이 의도한 대로 작동하며, 실제 사용 시나리오에서 문제가 발생하지 않는지를 확인한다.

      - 주요 특징:

         - 초점: 사용자 관점에서의 전체 프로세스와 기능을 검증한다.
         - 범위: 시스템의 모든 구성 요소와 외부 시스템 간의 상호 작용을 포함한 전체 워크플로우를 다룬다.
         - 방법: 실제 사용자 시나리오를 기반으로 테스트 케이스를 작성하고, 시스템의 모든 계층(UI, API, 데이터베이스 등)을 포함하여 테스트를 수행한다.

 

   4) 비교 요약:

      - 컴포넌트 통합 테스트: 여러 유닛이나 컴포넌트의 상호 작용을 검증하여, 모듈 내 통합의 정확성을 확인한다.
      - 시스템 통합 테스트: 통합된 모듈이나 서브시스템이 전체 시스템으로서 올바르게 작동하는지를 검증한다.
      - 엔드 투 엔드 통합 테스트: 사용자 관점에서 시스템의 전체 워크플로우를 검증하여, 실제 사용 시나리오에서의 문제를 발견하고 해결한다.

 

 

17. ORM 쿼리 결과의 __table__를 이용하여 컬럼 이름과 값 추출하여 비교 테스트하는 방법

   1) 응답 데이터와 데이터베이스 항목 일치 여부 검증:

      1. create_category.__table__.columns

         - create_category는 데이터베이스에서 조회한 카테고리 객체이다.
         - __table__.columns는 이 객체가 속한 테이블의 모든 컬럼(필드) 정보를 담고 있는 SQLAlchemy 객체이다.

 

      2. 딕셔너리 컴프리헨션

{
    column.name: getattr(create_category, column.name)
    for column in create_category.__table__.columns
}

 

         - 이 부분은 각 컬럼을 순회하면서, 각 컬럼의 이름(column.name)을 키(key)로, 해당 컬럼에 해당하는 create_category 객체의 값을 getattr(create_category, column.name)를 사용해 가져와서 값을 할당한다.
         - 결과적으로, 데이터베이스에 저장된 해당 카테고리의 모든 필드와 값들을 담은 딕셔너리가 생성된다.

 

      3. assert response.json() == ...

         - response.json()은 API 호출 후 받은 응답의 JSON 데이터를 파이썬 딕셔너리 형태로 변환한 결과이다.
         - 이 딕셔너리 컴프리헨션에서 생성한 딕셔너리와 비교하여, API 응답에 포함된 데이터가 데이터베이스에 실제 저장된 값과 정확히 일치하는지를 검증한다.

 

def test_integrate_create_new_category_successful(client, db_session_integration):
    # Arrange: Prepare test data
    category_data = get_random_category_dict()
    category_id = category_data.pop("id")

    # Act: Make a POST request to create a new category
    response = client.post("api/category/", json=category_data)

    # Assert: Verify response
    assert response.status_code == 201

    # Assert: Verify the response and database state
    create_category = (
        db_session_integration.query(Category).filter_by(id=category_id).first()
    )
    assert create_category is not None

    # Assert: Verify response data matches database entry
    assert response.json() == {
        column.name: getattr(create_category, column.name)
        for column in create_category.__table__.columns
    }

 

 - reference : 

https://www.lambdatest.com/blog/monkey-patching-in-python/

What is Monkey Patching in Python: A Complete Tutorial With Examples | LambdaTest