minaChoi/o2o-negosium-original
1
0
Fork 0
Code Pull Requests Activity
8112394aef
o2o-negosium-original/backend/loadtest/negotiation_locustfile.py
민헌 174bcc9f9f test(loadtest): 협상 도메인 부하 시나리오 + run_local_locust 개선 
- negotiation_locustfile: 공급사 브라우징(list/chat init·messages) +
  참여/거부 부하. reject 는 1회성이라 소비 전용 REJECT 풀(큐),
  participate 는 IN_PROGRESS no-op 이라 재사용 READ 풀. 시드 세션마다
  chat 오프닝 행을 심어 agent 호출 회피. catch_response 로 result.success 검증.
- run_local_locust.sh:
  · 단일 파일 자동 선택(불필요한 프롬프트 생략) + 다중 시 입력 검증
  · .venv 존재 확인, glob 수집, psql 에러 노출(2>/dev/null 제거)
  · negotiation 선택 시 세션 풀(CTE+generate_series) 시딩 → 풀 파일 전달
  · LOAD_READ_POOL / LOAD_REJECT_POOL 로 풀 크기 조절

검증: 10 users·20s 헤드리스에서 전 엔드포인트 0% 실패, agent 무호출 확인.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 14:36:48 +09:00

203 lines
9.1 KiB
Python
Raw Blame History

"""협상(negotiation) 도메인 부하 테스트 — 공급사 브라우징 + 참여/거부 시나리오.
각 가상 유저가 on_start 에서 self-register + 로그인(auth locustfile 과 동일)한 뒤,
로그인 유저가 속한 부하 공급사의 협상 세션을 대상으로:
- 읽기: 목록 조회(list) / 채팅 진입(chat_init) / 대화 히스토리(chat_messages)
- 변경: 참여(participate) / 거부(reject)
를 가중치대로 반복한다.
세션은 supplier backend 에 생성 API 가 없어(negodata 가 생성) DB 로 직접 시딩해야 한다.
run_local_locust.sh 가 부하 공급사에 세션 풀을 시딩하고 그 id 들을 풀 파일로 넘긴다(LOAD_NEGO_POOL_FILE):
- READ 풀 : IN_PROGRESS + open 견적 + chat 오프닝 행. 재사용(비소비) — list/init/messages/participate 대상.
- REJECT 풀: 위와 동일하나 reject 로 한 번씩만 소비(REJECTED 로 전이). 큐에서 pop, 유저 간 중복 없음.
풀 파일 형식(run_local_locust.sh 가 생성):
READ:<uuid>,<uuid>,...
REJECT:<uuid>
REJECT:<uuid>
...
주의:
- participate 는 IN_PROGRESS 세션에 대해 성공 no-op(무변경)이라 지속 반복해도 안전하다
(CREATED→IN_PROGRESS 쓰기 전이 자체는 pytest 가 검증). 여기선 엔드포인트 정상상태 비용을 측정.
- reject 는 1회성이라 REJECT 풀이 소진되면 해당 태스크는 건너뛴다(소진 사실을 로그로 남긴다 — silent cap 방지).
- chat_messages 는 빈 IN_PROGRESS 세션에서 agent 오프닝을 seed 하므로, 시드 세션마다 chat 행을 심어 agent 호출을 피한다.
직접 실행 시:
LOAD_SUPPLIER_ID=<공급사 uuid> LOAD_NEGO_POOL_FILE=<풀 파일 경로> \
locust -f loadtest/negotiation_locustfile.py --host http://localhost:9300
"""
import os
import queue
import random
from locust import HttpUser, between, events, task
LOAD_SUPPLIER_ID = os.environ.get("LOAD_SUPPLIER_ID", "")
LOAD_NEGO_POOL_FILE = os.environ.get("LOAD_NEGO_POOL_FILE", "")
LOAD_PW = "loadpw1234"
# 풀 파일에서 로드되는 세션 id 저장소(프로세스 공유 — locust 는 유저를 단일 프로세스의 greenlet 으로 실행).
READ_IDS: list[str] = [] # 비소비(재사용): list/init/messages/participate 대상
REJECT_QUEUE: "queue.Queue[str]" = queue.Queue() # 소비: reject 가 하나씩 pop
def _load_pool(path: str) -> None:
"""풀 파일을 읽어 READ_IDS 리스트와 REJECT_QUEUE 큐를 채운다."""
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if line.startswith("READ:"):
READ_IDS.extend(x for x in line[len("READ:"):].split(",") if x)
elif line.startswith("REJECT:"):
sid = line[len("REJECT:"):].strip()
if sid:
REJECT_QUEUE.put(sid)
class NegotiationUser(HttpUser):
wait_time = between(0.5, 2.0)
def on_start(self):
# 유저마다 고유 계정을 생성(self-register)하고 로그인해 토큰을 확보한다(auth 부하와 동일 패턴).
self.login_id = f"load_user_{random.randint(0, 1_000_000_000)}"
self.access_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]}")
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"):
self.access_token = resp.json().get("access_token")
resp.success()
else:
resp.failure(f"login failed: {resp.status_code} {resp.text[:120]}")
def _auth_header(self):
return {"Authorization": f"Bearer {self.access_token}"}
@task(6)
def list_sessions(self):
# 공급사 세션 목록 — 여러 스키마를 앱 레벨에서 조인하는 무거운 조회. 랜덤 필터/정렬/페이지.
if not self.access_token:
return
params = {"order": random.choice(["asc", "desc"]), "page": random.randint(1, 3), "page_size": 20}
if random.random() < 0.5:
params["status"] = random.choice([1, 2, 3])
if random.random() < 0.3:
params["qt_type"] = random.choice([1, 2])
with self.client.get(
"/v1/negotiation/sessions",
params=params,
headers=self._auth_header(),
name="GET /v1/negotiation/sessions",
catch_response=True,
) as resp:
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
resp.success()
else:
resp.failure(f"list failed: {resp.status_code} {resp.text[:120]}")
@task(4)
def chat_init(self):
# 채팅 진입(상품·견적 메타). 읽기 풀에서 임의 세션 선택.
if not self.access_token or not READ_IDS:
return
sid = random.choice(READ_IDS)
with self.client.get(
f"/v1/negotiation/sessions/{sid}/chat/init",
headers=self._auth_header(),
name="GET /v1/negotiation/sessions/[id]/chat/init",
catch_response=True,
) as resp:
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
resp.success()
else:
resp.failure(f"init failed: {resp.status_code} {resp.text[:120]}")
@task(3)
def chat_messages(self):
# 대화 히스토리(재진입 복원). 시드된 오프닝 행이 있어 agent 를 호출하지 않는다.
if not self.access_token or not READ_IDS:
return
sid = random.choice(READ_IDS)
with self.client.get(
f"/v1/negotiation/sessions/{sid}/chat/messages",
headers=self._auth_header(),
name="GET /v1/negotiation/sessions/[id]/chat/messages",
catch_response=True,
) as resp:
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
resp.success()
else:
resp.failure(f"messages failed: {resp.status_code} {resp.text[:120]}")
@task(2)
def participate(self):
# 참여 — 읽기 풀은 이미 IN_PROGRESS 라 성공 no-op(무변경). 엔드포인트 정상상태 비용(인증+조회+검증) 측정.
if not self.access_token or not READ_IDS:
return
sid = random.choice(READ_IDS)
with self.client.post(
f"/v1/negotiation/sessions/{sid}/participate",
headers=self._auth_header(),
name="POST /v1/negotiation/sessions/[id]/participate",
catch_response=True,
) as resp:
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
resp.success()
else:
resp.failure(f"participate failed: {resp.status_code} {resp.text[:120]}")
@task(1)
def reject(self):
# 거부 — REJECT 풀에서 하나 pop 해 소비(IN_PROGRESS→REJECTED 쓰기 전이). 풀 소진 시 건너뛴다.
if not self.access_token:
return
try:
sid = REJECT_QUEUE.get_nowait()
except queue.Empty:
return # 풀 소진 — test_stop 에서 소진 사실을 알린다
with self.client.post(
f"/v1/negotiation/sessions/{sid}/reject",
json={"reject_reason": "부하테스트 단종"},
headers=self._auth_header(),
name="POST /v1/negotiation/sessions/[id]/reject",
catch_response=True,
) as resp:
if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
resp.success()
else:
resp.failure(f"reject failed: {resp.status_code} {resp.text[:120]}")
@events.test_start.add_listener
def _on_start(environment, **kwargs):
if not LOAD_SUPPLIER_ID:
print("[warn] LOAD_SUPPLIER_ID 가 비어있습니다 — self-register 가 전부 실패합니다. run_local_locust.sh 로 실행하세요.")
if LOAD_NEGO_POOL_FILE and os.path.exists(LOAD_NEGO_POOL_FILE):
_load_pool(LOAD_NEGO_POOL_FILE)
print(f"협상 부하 시작 — READ 풀 {len(READ_IDS)}건 / REJECT 풀 {REJECT_QUEUE.qsize()}건")
if not READ_IDS:
print("[warn] READ 풀이 비었습니다 — init/messages/participate 가 동작하지 않습니다. run_local_locust.sh 로 실행하세요.")
@events.test_stop.add_listener
def _on_stop(environment, **kwargs):
if REJECT_QUEUE.empty():
print("[info] REJECT 풀 소진 — reject 태스크는 이후 건너뛰었습니다. 더 오래/세게 돌리려면 LOAD_REJECT_POOL 을 키우세요.")
else:
print(f"[info] REJECT 풀 잔여 {REJECT_QUEUE.qsize()}건.")
View Git Blame Copy Permalink