chore: add locust-maintainer agent spec and expand locustfile seed data with polls, reactions, and docs slugs

This commit is contained in:
2026-06-13 23:31:28 +00:00
parent 1b89f7c49f
commit 45b2b07038
2 changed files with 594 additions and 9 deletions
+520 -9
View File
@@ -23,16 +23,44 @@ NEWS_SLUGS = []
GIST_SLUGS = []
GIST_UIDS = []
BUG_UIDS = []
BUG_JOB_UIDS = []
NOTIFICATION_UIDS = []
POLLS = []
ADMIN_USER = {}
ADMIN_TARGETS = {}
TOPICS = ["devlog", "showcase", "question", "rant", "fun", "random"]
PROJECT_TYPES = ["game", "game_asset", "software", "mobile_app", "website"]
GIST_LANGUAGES = ["python", "javascript", "go", "rust", "bash", "sql", "json"]
REACTION_EMOJIS = ["\U0001f44d", "\U0001f680", "\U0001f389", "\U0001f525"]
NOTIFICATION_TYPES = ["comment", "reply", "vote", "follow", "message"]
DOCS_SLUGS = [
"index",
"getting-started",
"devii",
"components",
"dashboard",
"claude",
]
UPLOAD_FILE = ("load_test.txt", b"locust upload payload", "text/plain")
UUID_RE = r"[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}"
PASSWORD = "testpass123"
# ── Deliberately NOT load tested (kept out by design) ───────────────
# /openai/v1/*, /devii/clippy/ai/chat - real upstream AI spend / billing.
# /p/{slug}, /projects/.../containers/*, /admin/containers/*
# - drive the host docker daemon.
# /bugs/{number}, /bugs/{number}/comment, /bugs/{number}/status
# - live external Gitea REST calls.
# /admin/ai-quota/reset-guests|reset-all - global destructive quota wipe.
# /admin/services/{name}/run|start|stop|config|clear-logs
# - toggles real background services
# (news grading bills AI; could
# destabilize the load target).
# /admin/trash/.../purge, /admin/media/{uid}/purge of seeded content
# - hard GC of real content (only the
# dedicated disposable media is purged).
# WebSockets (/devii/ws, exec ws) - HttpUser cannot drive them.
def random_username():
return f"lu_{uuid.uuid4().hex[:10]}"
@@ -127,11 +155,15 @@ def seed_data(environment, **kwargs):
try:
opener = logged_in_opener(seed["email"])
body = urllib.parse.urlencode(
{
"content": "x " * 100,
"title": f"Seed post {uuid.uuid4().hex[:6]}",
"topic": random.choice(TOPICS),
}
[
("content", "x " * 100),
("title", f"Seed post {uuid.uuid4().hex[:6]}"),
("topic", random.choice(TOPICS)),
("poll_question", f"Seed poll {uuid.uuid4().hex[:4]}?"),
("poll_options", "alpha"),
("poll_options", "beta"),
("poll_options", "gamma"),
]
).encode()
resp = opener.open(
urllib.request.Request(f"{host}/posts/create", data=body),
@@ -366,8 +398,14 @@ def seed_data(environment, **kwargs):
html = resp.read().decode("utf-8", errors="replace")
matches = re.findall(rf"/comments/delete/{uuid_pat}", html)
COMMENT_UIDS.extend(m for m in matches if m not in COMMENT_UIDS)
poll_uid = re.search(rf'data-poll-uid="{uuid_pat}"', html)
option_uids = re.findall(rf'data-option-uid="{uuid_pat}"', html)
if poll_uid and option_uids:
entry = {"poll_uid": poll_uid.group(1), "options": option_uids}
if entry not in POLLS:
POLLS.append(entry)
except Exception as e:
logger.warning(f"Harvest comment UIDs failed: {e}")
logger.warning(f"Harvest comment/poll UIDs failed: {e}")
try:
opener = logged_in_opener(SEED_USERS[0]["email"])
@@ -394,6 +432,7 @@ class DevPlaceUser(HttpUser):
self.own_post_slugs = []
self.own_gist_slugs = []
self.own_project_slugs = []
self.own_media_uids = []
if random.random() < 0.3 and SEED_USERS:
seed = random.choice(SEED_USERS)
do_login(self.client, seed["email"])
@@ -523,6 +562,47 @@ class DevPlaceUser(HttpUser):
def view_static(self):
self.client.get("/static/css/base.css", name="static")
@task(2)
def view_docs(self):
if random.random() < 0.2:
self.client.get(
random.choice(["/docs", "/docs/download.md", "/docs/download.html"]),
name="docs",
)
return
self.client.get(
f"/docs/{random.choice(DOCS_SLUGS)}.html", name="docs/[slug]"
)
@task(1)
def view_devii_pages(self):
self.client.get(
random.choice(["/devii/", "/devii/session", "/devii/usage"]),
name="devii/[page]",
)
@task(1)
def view_devii_adopt(self):
self.client.get("/devii/adopt", params={"next": "/feed"}, name="devii/adopt")
@task(2)
def view_followers(self):
if not ALL_USERNAMES:
return
username = random.choice(ALL_USERNAMES)
self.client.get(
f"/profile/{username}/{random.choice(['followers', 'following'])}",
name="profile/[username]/follows",
)
@task(2)
def view_notification_counts(self):
self.client.get("/notifications/counts", name="notifications/counts")
@task(2)
def view_saved(self):
self.client.get("/bookmarks/saved", name="bookmarks/saved")
# ── auth flows ──────────────────────────────────────────────
@task(1)
@@ -652,14 +732,28 @@ class DevPlaceUser(HttpUser):
@task(1)
def create_bug(self):
self.client.post(
with self.client.post(
"/bugs/create",
data={
"title": f"Bug {uuid.uuid4().hex[:6]}",
"description": "Load test bug report description.",
},
catch_response=True,
name="bugs/create",
)
) as resp:
if resp.status_code == 503:
resp.success()
return
if resp.status_code != 200:
resp.failure(f"bug create: status={resp.status_code}")
return
resp.success()
try:
uid = resp.json().get("uid")
except ValueError:
uid = None
if uid and uid not in BUG_JOB_UIDS:
BUG_JOB_UIDS.append(uid)
# ── social interaction ──────────────────────────────────────
@@ -716,6 +810,104 @@ class DevPlaceUser(HttpUser):
return
self._vote("bug", random.choice(BUG_UIDS))
def _react(self, target_type, uid):
emoji = random.choice(REACTION_EMOJIS)
with self.client.post(
f"/reactions/{target_type}/{uid}",
data={"emoji": emoji},
headers={"x-requested-with": "fetch"},
catch_response=True,
name=f"reactions/{target_type}",
) as resp:
try:
payload = resp.json()
except ValueError:
resp.failure("reaction response not JSON")
return
if "counts" in payload and "mine" in payload:
resp.success()
else:
resp.failure(f"reaction missing keys: {payload}")
@task(2)
def react_on_post(self):
if not POST_UIDS:
return
self._react("post", random.choice(POST_UIDS))
@task(1)
def react_on_target(self):
pools = []
if GIST_UIDS:
pools.append(("gist", GIST_UIDS))
if PROJECT_UIDS:
pools.append(("project", PROJECT_UIDS))
if COMMENT_UIDS:
pools.append(("comment", COMMENT_UIDS))
if not pools:
return
target_type, uids = random.choice(pools)
self._react(target_type, random.choice(uids))
def _bookmark(self, target_type, uid):
with self.client.post(
f"/bookmarks/{target_type}/{uid}",
headers={"x-requested-with": "fetch"},
catch_response=True,
name=f"bookmarks/{target_type}",
) as resp:
try:
payload = resp.json()
except ValueError:
resp.failure("bookmark response not JSON")
return
if "saved" in payload:
resp.success()
else:
resp.failure(f"bookmark missing keys: {payload}")
@task(2)
def bookmark_post(self):
if not POST_UIDS:
return
self._bookmark("post", random.choice(POST_UIDS))
@task(1)
def bookmark_target(self):
pools = []
if GIST_UIDS:
pools.append(("gist", GIST_UIDS))
if PROJECT_UIDS:
pools.append(("project", PROJECT_UIDS))
if NEWS_UIDS:
pools.append(("news", NEWS_UIDS))
if not pools:
return
target_type, uids = random.choice(pools)
self._bookmark(target_type, random.choice(uids))
@task(2)
def vote_on_poll(self):
if not POLLS:
return
poll = random.choice(POLLS)
with self.client.post(
f"/polls/{poll['poll_uid']}/vote",
data={"option_uid": random.choice(poll["options"])},
headers={"x-requested-with": "fetch"},
catch_response=True,
name="polls/[uid]/vote",
) as resp:
try:
payload = resp.json()
except ValueError:
resp.failure("poll vote response not JSON")
return
if "options" in payload or "total" in payload:
resp.success()
else:
resp.failure(f"poll vote missing keys: {payload}")
@task(1)
def follow_user(self):
targets = [u for u in ALL_USERNAMES if u != self.username]
@@ -893,7 +1085,12 @@ class DevPlaceUser(HttpUser):
except ValueError:
uid = None
if uid:
self.client.delete(f"/uploads/delete/{uid}", name="uploads/delete")
if len(self.own_media_uids) < 5 and random.random() < 0.5:
self.own_media_uids.append(uid)
else:
self.client.delete(
f"/uploads/delete/{uid}", name="uploads/delete"
)
else:
resp.failure(f"upload failed: status={resp.status_code}")
@@ -918,6 +1115,239 @@ class DevPlaceUser(HttpUser):
else:
resp.failure(f"push register: status={resp.status_code}")
@task(1)
def regenerate_api_key(self):
self.client.post(
"/profile/regenerate-api-key", name="profile/regenerate-api-key"
)
@task(1)
def toggle_customization(self):
scope = random.choice(["global", "pagetype"])
self.client.post(
f"/profile/{self.username}/customization/{scope}",
data={"value": random.choice(["true", "false"])},
name=f"profile/customization/{scope}",
)
@task(1)
def set_notification_preference(self):
self.client.post(
f"/profile/{self.username}/notifications",
data={
"notification_type": random.choice(NOTIFICATION_TYPES),
"channel": random.choice(["in_app", "push"]),
"value": random.choice(["true", "false"]),
},
name="profile/notifications",
)
@task(1)
def reset_notification_preferences(self):
self.client.post(
f"/profile/{self.username}/notifications/reset",
name="profile/notifications/reset",
)
@task(1)
def upload_url_attachment(self):
with self.client.post(
"/uploads/upload-url",
data={"url": "https://placehold.co/64x64.png"},
catch_response=True,
name="uploads/upload-url",
) as resp:
if resp.status_code in (201, 400, 422):
resp.success()
else:
resp.failure(f"upload-url: status={resp.status_code}")
@task(1)
def manage_own_media(self):
if not self.own_media_uids:
return
uid = random.choice(self.own_media_uids)
self.client.post(f"/media/{uid}/delete", name="media/delete")
self.client.post(f"/media/{uid}/restore", name="media/restore")
@task(1)
def file_bug_job_status(self):
if not BUG_JOB_UIDS:
return
with self.client.get(
f"/bugs/jobs/{random.choice(BUG_JOB_UIDS)}",
catch_response=True,
name="bugs/jobs/[uid]",
) as resp:
if resp.status_code in (200, 404):
resp.success()
else:
resp.failure(f"bug job status: status={resp.status_code}")
# ── owner-scoped project mutations ──────────────────────────
@task(1)
def edit_project(self):
if not self.own_project_slugs:
return
self.client.post(
f"/projects/edit/{random.choice(self.own_project_slugs)}",
data={
"title": "Edited project",
"description": "Edited description for load test.",
"project_type": random.choice(PROJECT_TYPES),
},
name="projects/edit",
)
@task(1)
def toggle_project_flags(self):
if not self.own_project_slugs:
return
slug = random.choice(self.own_project_slugs)
self.client.post(
f"/projects/{slug}/{random.choice(['private', 'readonly'])}",
name="projects/[slug]/flag",
)
@task(2)
def browse_project_files(self):
if not PROJECT_SLUGS:
return
slug = random.choice(PROJECT_SLUGS)
self.client.get(
f"/projects/{slug}/files", params={"path": ""}, name="projects/[slug]/files"
)
@task(2)
def edit_own_project_files(self):
if not self.own_project_slugs:
return
slug = random.choice(self.own_project_slugs)
path = f"notes/{uuid.uuid4().hex[:6]}.txt"
with self.client.post(
f"/projects/{slug}/files/write",
data={"path": path, "content": "line one\nline two\nline three\n"},
catch_response=True,
name="projects/[slug]/files/write",
) as resp:
if resp.status_code in (200, 302, 303):
resp.success()
else:
resp.failure(f"file write: status={resp.status_code}")
return
self.client.get(
f"/projects/{slug}/files/raw",
params={"path": path},
name="projects/[slug]/files/raw",
)
self.client.get(
f"/projects/{slug}/files/lines",
params={"path": path, "start": 1, "end": 2},
name="projects/[slug]/files/lines",
)
self.client.post(
f"/projects/{slug}/files/append",
data={"path": path, "content": "appended line\n"},
name="projects/[slug]/files/append",
)
self.client.post(
f"/projects/{slug}/files/replace-lines",
data={"path": path, "start": 1, "end": 1, "content": "replaced line"},
name="projects/[slug]/files/replace-lines",
)
self.client.post(
f"/projects/{slug}/files/insert-lines",
data={"path": path, "at": 1, "content": "inserted line"},
name="projects/[slug]/files/insert-lines",
)
self.client.post(
f"/projects/{slug}/files/delete-lines",
data={"path": path, "start": 1, "end": 1},
name="projects/[slug]/files/delete-lines",
)
moved = f"notes/{uuid.uuid4().hex[:6]}.txt"
self.client.post(
f"/projects/{slug}/files/move",
data={"from_path": path, "to_path": moved},
name="projects/[slug]/files/move",
)
self.client.post(
f"/projects/{slug}/files/delete",
data={"path": moved},
name="projects/[slug]/files/delete",
)
@task(1)
def make_project_dir_and_upload(self):
if not self.own_project_slugs:
return
slug = random.choice(self.own_project_slugs)
directory = f"dir_{uuid.uuid4().hex[:6]}"
self.client.post(
f"/projects/{slug}/files/mkdir",
data={"path": directory},
name="projects/[slug]/files/mkdir",
)
name, content, mime = UPLOAD_FILE
self.client.post(
f"/projects/{slug}/files/upload",
data={"path": directory},
files={"file": (name, content, mime)},
name="projects/[slug]/files/upload",
)
@task(1)
def zip_project(self):
if not self.own_project_slugs:
return
slug = random.choice(self.own_project_slugs)
endpoint = random.choice(["zip", "files/zip"])
with self.client.post(
f"/projects/{slug}/{endpoint}",
params={"path": ""},
catch_response=True,
name="projects/[slug]/zip",
) as resp:
if resp.status_code not in (200, 202):
resp.failure(f"zip enqueue: status={resp.status_code}")
return
try:
uid = resp.json().get("uid")
except ValueError:
uid = re.search(rf"/zips/({UUID_RE})", resp.text)
uid = uid.group(1) if uid else None
resp.success()
if uid:
self.client.get(f"/zips/{uid}", name="zips/[uid]")
self.client.get(
f"/zips/{uid}/download",
name="zips/[uid]/download",
catch_response=True,
).success()
@task(1)
def fork_project(self):
if not PROJECT_SLUGS:
return
slug = random.choice(PROJECT_SLUGS)
with self.client.post(
f"/projects/{slug}/fork",
data={"title": f"Fork {uuid.uuid4().hex[:6]}"},
catch_response=True,
name="projects/[slug]/fork",
) as resp:
if resp.status_code not in (200, 202):
resp.failure(f"fork enqueue: status={resp.status_code}")
return
try:
uid = resp.json().get("uid")
except ValueError:
uid = None
resp.success()
if uid:
self.client.get(f"/forks/{uid}", name="forks/[uid]")
# ── static / media ──────────────────────────────────────────
@task(2)
@@ -1036,6 +1466,81 @@ class AdminUser(HttpUser):
def view_admin_services_data(self):
self.client.get("/admin/services/data", name="admin/services/data")
@task(1)
def view_admin_service_detail(self):
self.client.get("/admin/services/audit", name="admin/services/[name]")
self.client.get(
"/admin/services/audit/data", name="admin/services/[name]/data"
)
@task(2)
def view_admin_ai_usage(self):
self.client.get("/admin/ai-usage", name="admin/ai-usage")
self.client.get("/admin/ai-usage/data", name="admin/ai-usage/data")
@task(1)
def view_admin_analytics(self):
self.client.get("/admin/analytics", name="admin/analytics")
@task(2)
def view_admin_audit_log(self):
resp = self.client.get("/admin/audit-log", name="admin/audit-log")
m = re.search(rf"/admin/audit-log/({UUID_RE})", resp.text)
if m:
self.client.get(
f"/admin/audit-log/{m.group(1)}", name="admin/audit-log/[uid]"
)
@task(1)
def view_admin_media(self):
self.client.get("/admin/media", name="admin/media")
@task(1)
def view_admin_notifications(self):
self.client.get("/admin/notifications", name="admin/notifications")
@task(1)
def save_admin_notification_default(self):
self.client.post(
"/admin/notifications",
data={
"notification_type": random.choice(NOTIFICATION_TYPES),
"channel": random.choice(["in_app", "push"]),
"value": random.choice(["true", "false"]),
},
name="admin/notifications/save",
)
@task(1)
def view_admin_user_ai_usage(self):
uid = ADMIN_TARGETS.get("disposable_uid")
if not uid:
return
self.client.get(
f"/admin/users/{uid}/ai-usage", name="admin/users/[uid]/ai-usage"
)
@task(1)
def reset_user_ai_quota(self):
uid = ADMIN_TARGETS.get("disposable_uid")
if not uid:
return
self.client.post(
f"/admin/users/{uid}/reset-ai-quota", name="admin/users/reset-ai-quota"
)
@task(1)
def trash_view_and_restore(self):
resp = self.client.get("/admin/trash", name="admin/trash")
m = re.search(
rf"/admin/trash/([a-z_]+)/({UUID_RE})/restore", resp.text
)
if m:
self.client.post(
f"/admin/trash/{m.group(1)}/{m.group(2)}/restore",
name="admin/trash/restore",
)
@task(1)
def save_settings(self):
self.client.post(
@@ -1130,6 +1635,12 @@ class AnonymousUser(HttpUser):
)
self.client.get(path, name="public/[page]")
@task(1)
def browse_docs(self):
self.client.get(
f"/docs/{random.choice(DOCS_SLUGS)}.html", name="docs/[slug]"
)
@task(2)
def browse_detail(self):
if NEWS_SLUGS: