23cb8291bc
- run_local_server.sh / run_local_locust.sh / run_local_pgwatch.sh (대화형, 메뉴 선택 방식) - loadtest: self-register 방식 locustfile (가상 유저가 on_start 에서 계정 생성 → login/me/refresh) - locust 스크립트가 부하용 공급사 보장·이전 부하계정/토큰 정리·LOAD_SUPPLIER_ID 주입을 자동 처리 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
"""인증 서버 부하 테스트 (self-register 방식, 사전 시드 불필요).
|
||
|
||
각 가상 유저가 on_start 에서 자기 계정을 생성(/create)하고 로그인한 뒤,
|
||
me/login/refresh/healthz 를 가중치대로 반복한다.
|
||
|
||
supplier 의 /create 는 supplier_id(소속 공급사)가 필수이므로, run_local_locust.sh 가
|
||
부하용 공급사(부하테스트공급사) 1건을 보장하고 그 supplier_id 를 LOAD_SUPPLIER_ID 로 넘긴다.
|
||
|
||
직접 실행 시:
|
||
LOAD_SUPPLIER_ID=<공급사 uuid> locust -f loadtest/locustfile.py --host http://localhost:9300
|
||
|
||
정리(테스트 후, self-register 로 쌓인 계정 삭제):
|
||
psql ... -c "DELETE FROM supplier.supplier_users WHERE id LIKE 'load_user_%';"
|
||
|
||
주의:
|
||
- /create, /login 은 bcrypt(CPU 바운드) + DB read/write 라 가장 무겁다.
|
||
- /me, /refresh 는 su_id DB 존재/활성 검증(read 2회)을 한다 — 순수 JWT 경로가 아니다.
|
||
- 논리 DB 가 USER/PARTNER 2개라 같은 negosium_db 에 엔진 풀이 4벌(R/W×2) 잡힌다.
|
||
SELECT count(*) FROM pg_stat_activity WHERE datname = 'negosium_db';
|
||
"""
|
||
|
||
import os
|
||
import random
|
||
|
||
from locust import HttpUser, between, events, task
|
||
|
||
LOAD_SUPPLIER_ID = os.environ.get("LOAD_SUPPLIER_ID", "")
|
||
LOAD_PW = "loadpw1234"
|
||
|
||
|
||
class AuthUser(HttpUser):
|
||
wait_time = between(0.5, 2.0)
|
||
|
||
def on_start(self):
|
||
# 유저마다 고유 계정을 생성(self-register)하고 로그인해 토큰을 확보한다.
|
||
self.login_id = f"load_user_{random.randint(0, 1_000_000_000)}"
|
||
self.access_token = None
|
||
self.refresh_token = None
|
||
with self.client.post(
|
||
"/v1/auth/create",
|
||
json={"supplier_id": LOAD_SUPPLIER_ID, "id": self.login_id, "pw": LOAD_PW},
|
||
name="POST /v1/auth/create",
|
||
catch_response=True,
|
||
) as resp:
|
||
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
|
||
resp.success()
|
||
else:
|
||
resp.failure(f"create failed: {resp.status_code} {resp.text[:120]}")
|
||
self._login()
|
||
|
||
def _login(self):
|
||
with self.client.post(
|
||
"/v1/auth/login",
|
||
json={"id": self.login_id, "pw": LOAD_PW},
|
||
name="POST /v1/auth/login",
|
||
catch_response=True,
|
||
) as resp:
|
||
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
|
||
body = resp.json()
|
||
self.access_token = body.get("access_token")
|
||
self.refresh_token = body.get("refresh_token")
|
||
resp.success()
|
||
else:
|
||
resp.failure(f"login failed: {resp.status_code} {resp.text[:120]}")
|
||
|
||
@task(5)
|
||
def me(self):
|
||
# 토큰 검증 + su_id/공급사명 DB 조회(2 read). 보호 엔드포인트 처리량 측정.
|
||
if not self.access_token:
|
||
return
|
||
self.client.get(
|
||
"/v1/auth/me",
|
||
headers={"Authorization": f"Bearer {self.access_token}"},
|
||
name="GET /v1/auth/me",
|
||
)
|
||
|
||
@task(2)
|
||
def login(self):
|
||
# bcrypt + DB read 2 + DB write 가 포함된 무거운 경로.
|
||
self._login()
|
||
|
||
@task(1)
|
||
def refresh(self):
|
||
# refresh 토큰 검증 + su_id DB 존재/활성 확인 후 access 재발급.
|
||
if not self.refresh_token:
|
||
return
|
||
with self.client.post(
|
||
"/v1/auth/refresh_token",
|
||
headers={"Authorization": f"Bearer {self.refresh_token}"},
|
||
name="POST /v1/auth/refresh_token",
|
||
catch_response=True,
|
||
) as resp:
|
||
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
|
||
self.access_token = resp.json().get("access_token")
|
||
resp.success()
|
||
else:
|
||
resp.failure(f"refresh failed: {resp.status_code} {resp.text[:120]}")
|
||
|
||
@task(1)
|
||
def healthz(self):
|
||
# 베이스라인 (앱 오버헤드 측정).
|
||
self.client.get("/healthz", name="GET /healthz")
|
||
|
||
|
||
@events.test_start.add_listener
|
||
def _on_start(environment, **kwargs):
|
||
if not LOAD_SUPPLIER_ID:
|
||
print("[warn] LOAD_SUPPLIER_ID 가 비어있습니다 — /create 가 전부 실패합니다. run_local_locust.sh 로 실행하세요.")
|
||
print("부하 테스트 시작 - self-register 방식. PostgreSQL 커넥션 수 모니터링 권장 (pg_stat_activity)")
|