[프로젝트] 51. Docker 활용 주문 API 자동화 툴 구축

Docker를 사용하여 HTTP 요청을 자동화하는 툴을 구축하였습니다. 주요 기능은 사용자 등록, 로그인, 주문 생성, 결제 시도를 자동으로 수행하는 것입니다. Python 스크립트는 http_request_tool.py로 명명되며, 동시에 여러 HTTP 요청을 보내는 데 사용됩니다.

 

Dockerfile 구성

Dockerfile은 Python 환경을 구축하고 필요한 의존성을 설치하는 데 사용됩니다. 또한, 작성한 Python 스크립트를 컨테이너 내부에 복사하고 실행합니다.

# Dockerfile
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# CMD [ "python", "./http_request_tool.py" ]

# 회원가입이 이미 돼있는경우
CMD ["python", "http_request_tool.py", "--skip_registration"]

이 Dockerfile은 Python 3.9 slim 이미지를 기반으로 하며, 필요한 Python 패키지를 requirements.txt를 통해 설치합니다. 마지막으로, http_request_tool.py 스크립트를 실행합니다. 사용자 등록 과정을 건너뛰고 싶은 경우 --skip_registration 옵션을 사용합니다.

 

Python 스크립트 (http_request_tool.py)

http_request_tool.py 스크립트는 requests 라이브러리를 사용하여 HTTP 요청을 보내고, ThreadPoolExecutor를 사용하여 동시에 여러 요청을 처리합니다. 사용자 등록, 로그인, 주문 생성, 결제 시도 등의 과정을 자동화합니다.

import requests
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
import time
import sys

# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# API 엔드포인트 설정
JOIN_URL = "http://host.docker.internal:8081/api/auth/join"
LOGIN_URL = "http://host.docker.internal:8081/api/auth/login"
ORDERS_URL = "http://host.docker.internal:8084/api/orders"
PAY_URL = "http://host.docker.internal:8084/api/orders/pay/"

# 요청 타임아웃 설정 (초 단위)
REQUEST_TIMEOUT = 15

# 배치 처리 설정
BATCH_SIZE = 100
BATCH_DELAY = 5  # 배치 사이의 지연 시간 (초)

# 고유한 사용자 정보 생성 함수
def generate_user_info(user_id):
    return {
        "email": f"user_{user_id}@example.com",
        "password": "123123",
        "name": f"user_{user_id}"
    }

# 회원가입 함수
def register_user(user_info):
    try:
        response = requests.post(JOIN_URL, json=user_info, timeout=REQUEST_TIMEOUT)
        if response.status_code in [200, 201]:
            logging.info(f"Registration successful for {user_info['email']}")
            return True
        else:
            logging.warning(f"Registration failed for {user_info['email']}: {response.status_code}")
            return False
    except requests.exceptions.RequestException as e:
        logging.error(f"Registration error for {user_info['email']}: {e}")
        return False

# 로그인 함수
def login(user_info):
    try:
        response = requests.post(LOGIN_URL, json=user_info, timeout=REQUEST_TIMEOUT)
        if response.status_code == 200:
            token = response.json()['data']['token']
            logging.info(f"Login successful for {user_info['email']}")
            return token
        else:
            logging.warning(f"Login failed for {user_info['email']}: {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        logging.error(f"Login error for {user_info['email']}: {e}")
        return None

# 결제 화면 진입 함수
def enter_payment_screen(token):
    headers = {"Authorization": f"Bearer {token}"}
    data = [{"itemId": 15, "count": 1}]
    try:
        response = requests.post(ORDERS_URL, headers=headers, json=data, timeout=REQUEST_TIMEOUT)
        if response.status_code == 200:
            response_data = response.json()
            orderId = response_data.get('data')  # 'data' 필드에서 orderId 추출
            logging.info(f"Entering payment screen successful, orderId: {orderId}")
            return orderId
        else:
            logging.warning(f"Entering payment screen failed: {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        logging.error(f"Entering payment screen error: {e}")
        return None
        
# 결제 시도 함수
def attempt_payment(token, orderId):
    headers = {"Authorization": f"Bearer {token}"}
    try:
        response = requests.post(f"{PAY_URL}{orderId}", headers=headers, timeout=REQUEST_TIMEOUT)
        if response.status_code == 200:
            logging.info(f"Payment successful for order {orderId}")
            return True
        else:
            logging.warning(f"Payment failed for order {orderId}: {response.status_code}")
            return False
    except requests.exceptions.RequestException as e:
        logging.error(f"Payment error for order {orderId}: {e}")
        return False

# 메인 함수
def main(skip_registration):
    users_count = 10000  # 사용자 수를 10,000으로 설정
    batches = users_count // BATCH_SIZE

    with ThreadPoolExecutor(max_workers=50) as executor:
        for batch in range(batches):
            start_index = batch * BATCH_SIZE
            end_index = start_index + BATCH_SIZE
            user_infos = [generate_user_info(user_id) for user_id in range(start_index + 1, end_index + 1)]

            if not skip_registration:
                # 회원가입 배치 처리
                register_futures = [executor.submit(register_user, user_info) for user_info in user_infos]
                logging.info(f"Batch {batch+1}/{batches}: Registration started.")
                for future in as_completed(register_futures):
                    pass  # 회원가입 결과 처리

            # 로그인 배치 처리
            logging.info(f"Batch {batch+1}/{batches}: Login started.")
            login_futures = {executor.submit(login, user_info): user_info for user_info in user_infos}
            tokens = [future.result() for future in as_completed(login_futures) if future.result()]

            # 주문 생성 및 결제 시도 배치 처리
            logging.info(f"Batch {batch+1}/{batches}: Order and payment started.")
            order_futures = [executor.submit(enter_payment_screen, token) for token in tokens]
            orderIds = [future.result() for future in as_completed(order_futures) if future.result()]

            pay_futures = [executor.submit(attempt_payment, tokens[i], orderId) for i, orderId in enumerate(orderIds)]
            for future in as_completed(pay_futures):
                pass  # 결제 시도 결과 처리

            logging.info(f"Batch {batch+1}/{batches} completed.")
            if batch < batches - 1:
                logging.info(f"Waiting for {BATCH_DELAY} seconds before next batch...")
                time.sleep(BATCH_DELAY)  # 다음 배치까지 지연

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="HTTP Request Automation Tool")
    parser.add_argument("--skip_registration", action="store_true", help="Skip the registration process")
    args = parser.parse_args()

    main(args.skip_registration)

이 스크립트는 명령줄 인수 --skip_registration을 사용하여 사용자 등록 과정을 선택적으로 건너뛸 수 있습니다. ThreadPoolExecutor는 동시에 여러 작업을 실행하여 처리 성능을 향상시킵니다.

 

requirements.txt

requirements.txt 파일에는 스크립트 실행에 필요한 Python 패키지 목록이 포함됩니다.

# requirements.txt

requests

 

실행 방법

1. Docker 이미지 빌드

docker build -t http_request_tool .

2. Docker 컨테이너 실행:

docker run http_request_tool

사용자 등록 과정을 건너뛰려면:

docker run http_request_tool --skip_registration

 

이 도구를 사용하여 주문 API를 자동화하고, 복잡한 시나리오에서의 API 성능과 안정성을 테스트할 수 있습니다.