79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
import itertools
|
|
import os
|
|
import time
|
|
from typing import Dict
|
|
|
|
from locust import HttpUser, between, task
|
|
|
|
|
|
# Load-test target (localhost only by default).
|
|
TARGET_HOST = os.getenv("TARGET_HOST", "http://103.47.83.27")
|
|
REGISTER_PATH = os.getenv("REGISTER_PATH", "/api/user/register")
|
|
REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "10"))
|
|
WAIT_MIN = float(os.getenv("WAIT_MIN", "0"))
|
|
WAIT_MAX = float(os.getenv("WAIT_MAX", "0.2"))
|
|
|
|
# Register payload settings.
|
|
USERNAME_PREFIX = os.getenv("USERNAME_PREFIX", "load_user_")
|
|
DEFAULT_PASSWORD = os.getenv("REGISTER_PASSWORD", "123456")
|
|
EMAIL_DOMAIN = os.getenv("EMAIL_DOMAIN", "loadtest.local")
|
|
|
|
# Field names (adapt if your backend uses different keys).
|
|
USERNAME_FIELD = os.getenv("USERNAME_FIELD", "username")
|
|
PASSWORD_FIELD = os.getenv("PASSWORD_FIELD", "password")
|
|
CONFIRM_PASSWORD_FIELD = os.getenv("CONFIRM_PASSWORD_FIELD", "confirmPassword")
|
|
EMAIL_FIELD = os.getenv("EMAIL_FIELD", "email")
|
|
NICKNAME_FIELD = os.getenv("NICKNAME_FIELD", "nickname")
|
|
|
|
INCLUDE_CONFIRM_PASSWORD = os.getenv("INCLUDE_CONFIRM_PASSWORD", "1").lower() in {
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
}
|
|
INCLUDE_EMAIL = os.getenv("INCLUDE_EMAIL", "0").lower() in {"1", "true", "yes"}
|
|
INCLUDE_NICKNAME = os.getenv("INCLUDE_NICKNAME", "0").lower() in {"1", "true", "yes"}
|
|
|
|
SUCCESS_CODES = {
|
|
int(item.strip())
|
|
for item in os.getenv("SUCCESS_CODES", "200,201").split(",")
|
|
if item.strip()
|
|
}
|
|
USER_SEQ = itertools.count(int(time.time() * 1000))
|
|
|
|
|
|
def build_register_payload() -> Dict[str, str]:
|
|
seq = next(USER_SEQ)
|
|
username = f"{USERNAME_PREFIX}{seq}"
|
|
payload: Dict[str, str] = {
|
|
USERNAME_FIELD: username,
|
|
PASSWORD_FIELD: DEFAULT_PASSWORD,
|
|
}
|
|
|
|
if INCLUDE_CONFIRM_PASSWORD:
|
|
payload[CONFIRM_PASSWORD_FIELD] = DEFAULT_PASSWORD
|
|
if INCLUDE_EMAIL:
|
|
payload[EMAIL_FIELD] = f"{username}@{EMAIL_DOMAIN}"
|
|
if INCLUDE_NICKNAME:
|
|
payload[NICKNAME_FIELD] = f"u{seq}"
|
|
return payload
|
|
|
|
|
|
class RegisterLoadUser(HttpUser):
|
|
host = TARGET_HOST
|
|
wait_time = between(WAIT_MIN, WAIT_MAX)
|
|
|
|
@task
|
|
def register(self) -> None:
|
|
payload = build_register_payload()
|
|
with self.client.post(
|
|
REGISTER_PATH,
|
|
json=payload,
|
|
name=f"POST {REGISTER_PATH}",
|
|
timeout=REQUEST_TIMEOUT,
|
|
catch_response=True,
|
|
) as response:
|
|
if response.status_code in SUCCESS_CODES:
|
|
response.success()
|
|
return
|
|
response.failure(f"register failed ({response.status_code})")
|