sinau-c/test-load/locustfile.py

721 lines
23 KiB
Python

"""
Locust Load Tests for Elemes LMS (Sinau-C)
This file contains load test scenarios for the Elemes Learning Management System.
It simulates student and teacher behaviors including:
- Student: login, view lessons, compile code, submit progress
- Teacher: login, view progress reports, export CSV
- Frontend: load HTML pages (home, lesson pages, progress page)
IMPORTANT: This test uses ONLY the teacher token to avoid modifying student data.
Usage:
# API only (Flask backend :5000)
locust -f locustfile.py --host=http://localhost:5000
# Frontend only (SvelteKit :3000)
locust -f locustfile.py --host=http://localhost:3000
# Full stack test (use frontend host, it proxies /api to Flask)
locust -f locustfile.py --host=http://localhost:3000
Or for headless mode:
locust -f locustfile.py --host=http://localhost:3000 --headless -u 100 -r 10 -t 300s
Parameters:
-u: Number of users to simulate
-r: Spawn rate (users per second)
-t: Test duration (e.g., 300s, 5m, 1h)
"""
import random
import time
import logging
from locust import HttpUser, task, between, events
from faker import Faker
fake = Faker()
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class FrontendUser(HttpUser):
"""
Simulates a real user browsing the SvelteKit frontend.
This class tests the actual HTML pages that users see, including:
- Home page (lesson grid)
- Individual lesson pages
- Progress report page
- Static assets (CSS, JS)
NOTE: Set host to http://localhost:3000 for frontend testing
This class will be disabled if host is :5000 (API only)
"""
wait_time = between(2, 5) # Real users browse slower
weight = 5 # Significant portion of traffic
teacher_token = "141214"
lessons = [
"hello_world",
"variables",
"data_types",
"operators",
"conditions",
"for_loops",
"while_loops",
"arrays",
"strings",
"functions",
]
def on_start(self):
"""Check if frontend is available"""
try:
# Quick check if we're testing frontend (port 3000) or API (port 5000)
if self.host and ":5000" in self.host:
logger.warning("FrontendUser disabled: Host is API server (:5000)")
self.weight = 0 # Disable this user class
else:
logger.info("FrontendUser started successfully")
except Exception as e:
logger.error(f"FrontendUser on_start error: {e}")
@task(5)
def load_homepage(self):
"""Load the main homepage (lesson grid)"""
try:
with self.client.get(
"/",
name="/ [Homepage]",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
# Check if page contains expected content
if "Selamat Datang" in response.text or "lesson" in response.text.lower():
response.success()
else:
response.success() # Still success even if content check fails
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"load_homepage error: {e}")
@task(8)
def load_lesson_page(self):
"""Load an individual lesson page"""
try:
lesson = random.choice(self.lessons)
with self.client.get(
f"/lesson/{lesson}",
name="/lesson/[slug]",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"load_lesson_page error: {e}")
@task(3)
def load_progress_page(self):
"""Load the progress report page (teacher view)"""
try:
with self.client.get(
"/progress",
name="/progress",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"load_progress_page error: {e}")
@task(2)
def load_static_assets(self):
"""Load static assets (CSS, JS, manifest)"""
try:
assets = [
"/manifest.json",
"/circuitjs1/circuitjs.html",
]
for asset in random.sample(assets, min(2, len(assets))):
self.client.get(
asset,
name="/static/[asset]",
timeout=10,
)
except Exception as e:
logger.error(f"load_static_assets error: {e}")
class StudentUser(HttpUser):
"""
Simulates a student user interacting with the LMS.
Behavior flow:
1. Login with student token (test accounts)
2. View lesson list (API)
3. View individual lessons (API + Frontend page)
4. Compile code
5. Validate token (read-only)
NOTE: Uses student test tokens (rows 40-72 in tokens_siswa.csv)
These are test accounts that won't affect real student data
For full student simulation with frontend, use FrontendUser class.
This class focuses on API interactions.
"""
wait_time = between(1, 3) # Wait 1-3 seconds between tasks
weight = 8 # 80% of traffic will be students
# Test student tokens from tokens_siswa.csv (rows 40-72)
# These are accounts for testing purposes
student_tokens = [
"CC90965E", # ACHMAD RAFFY IRSYAD RAMADHAN
"F84E3336", # AHMAD GHAUZAN AZRIL IRAWAN
"D78F6DBD", # AINI BINTANG PRAMESWARI
"0D7B2566", # ALVINO MIRZA SAPUTRA
"3ED63295", # ANDREAS DIMAS PRASETIYO
"90526D3E", # ARDEN HABIBIE PRADANA
"298933B4", # ASHIF ZAFRAN ANABIL PUTRA ANWAR
"26B7E4F7", # DANIA AURELY RAHMADANI PUTRI MAWANTI
"D1B9E890", # DANISH AZKA HAFIZ
"CCA98B61", # FADHIL IRSA ARYA MAHITA
"894B6691", # FAREL MARCELINO RAMADHANI
"8EC73A61", # FAVIAN ALLAND YUDHANTO
"88DD18AF", # FITRA KURNIA RAMADHANI
"DFDAE0BD", # GEORGE AXL SISWANDOYO
"516D84E4", # HAIKAL AKBAR MAULANA
"C1BCC269", # JHOFAN RADITYA RAFARDAN
"0B59D8F3", # KEENAN ADHIYAKSA RIDHAA RAMADHAN
"44888E91", # MOCHAMMAD FATHUR ROCHMAN ATTOILAH
"1547C54D", # MUHAMAD FAHMI MAULIDI
"348C765F", # MUHAMMAD AFIF FAITH RAMADHAN
"E3183577", # MUHAMMAD FAIRUZ RAFI IRAWAN
"1B4ACCD0", # MUHAMMAD FAURIL ALFIANSYAH
"BB38203E", # MUHAMMAD KAMIL Hanif RAFASYAH
"B4CD262B", # MUHAMMAD SAIFURRAHMAN AL 'AZIZI
"63BC4A4F", # NADIVA
"567050B0", # PANDU BUDI SATRIO
"531D6090", # RADITYA PASHA
"54933D60", # RAFAEL EZAR ALFATHUR
"FD8D92D0", # RAHMITHA DWI RAHAYU
"9023A552", # RATU AURELIA BEATRICE PRAYOGA
"114B2384", # REVAN RAZIQ AKMAL
"5E878384", # RIZKY NANDI SAPUTRA
"BE0DEBFE", # SATRIYO WICAKSONO
]
# Sample lesson filenames from content/
lessons = [
"hello_world",
"variables",
"data_types",
"operators",
"conditions",
"for_loops",
"while_loops",
"arrays",
"strings",
"functions",
]
# Sample C code for compilation tests
code_samples = [
{
"name": "hello_world",
"code": """#include <stdio.h>
int main() {
printf("Hello, World!\\n");
return 0;
}"""
},
{
"name": "variables",
"code": """#include <stdio.h>
int main() {
int age = 20;
float height = 175.5;
char grade = 'A';
printf("Age: %d\\n", age);
printf("Height: %.1f\\n", height);
printf("Grade: %c\\n", grade);
return 0;
}"""
},
{
"name": "conditions",
"code": """#include <stdio.h>
int main() {
int score = 85;
if (score >= 90) {
printf("Grade: A\\n");
} else if (score >= 80) {
printf("Grade: B\\n");
} else if (score >= 70) {
printf("Grade: C\\n");
} else {
printf("Grade: D\\n");
}
return 0;
}"""
},
{
"name": "for_loop",
"code": """#include <stdio.h>
int main() {
for (int i = 1; i <= 10; i++) {
printf("%d ", i);
}
printf("\\n");
return 0;
}"""
},
{
"name": "array_sum",
"code": """#include <stdio.h>
int main() {
int numbers[5] = {1, 2, 3, 4, 5};
int sum = 0;
for (int i = 0; i < 5; i++) {
sum += numbers[i];
}
printf("Sum: %d\\n", sum);
return 0;
}"""
},
]
def on_start(self):
"""Called when a simulated user starts"""
try:
self.token = random.choice(self.student_tokens)
logger.info(f"StudentUser started with token: {self.token}")
except Exception as e:
logger.error(f"StudentUser on_start error: {e}")
@task(3)
def view_lessons_list(self):
"""View the list of available lessons"""
try:
with self.client.get(
"/api/lessons",
params={"token": self.token},
name="/api/lessons",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
data = response.json()
if "lessons" in data:
response.success()
else:
response.success() # Don't fail, just log
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"view_lessons_list error: {e}")
@task(5)
def view_single_lesson(self):
"""View a single lesson content"""
try:
lesson = random.choice(self.lessons)
with self.client.get(
f"/api/lesson/{lesson}.json",
params={"token": self.token},
name="/api/lesson/[slug].json",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"view_single_lesson error: {e}")
@task(4)
def compile_code(self):
"""Compile and run C code"""
try:
code_sample = random.choice(self.code_samples)
with self.client.post(
"/api/compile",
json={
"code": code_sample["code"],
"language": "c",
},
name="/api/compile",
catch_response=True,
timeout=30, # Compilation can take longer
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"compile_code error: {e}")
@task(1)
def validate_token(self):
"""Validate user token (read-only operation)"""
try:
with self.client.post(
"/api/validate-token",
json={"token": self.token},
name="/api/validate-token",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"validate_token error: {e}")
@task(1)
def logout(self):
"""Logout from the system"""
try:
with self.client.post(
"/api/logout",
name="/api/logout",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"logout error: {e}")
class TeacherUser(HttpUser):
"""
Simulates a teacher user interacting with the LMS.
Behavior flow:
1. Login with teacher token
2. View progress reports
3. Export CSV
4. View individual student progress
5. Track progress (testing with test student tokens)
"""
wait_time = between(2, 5) # Teachers browse slower
weight = 2 # 20% of traffic will be teachers
# Teacher token (first row in tokens_siswa.csv)
teacher_token = "141214" # Anggoro Dwi
# Test student tokens for progress tracking tests
student_tokens = [
"CC90965E", "F84E3336", "D78F6DBD", "0D7B2566", "3ED63295",
"90526D3E", "298933B4", "26B7E4F7", "D1B9E890", "CCA98B61",
]
lessons = [
"hello_world",
"variables",
"data_types",
"operators",
"conditions",
"for_loops",
"while_loops",
"arrays",
"strings",
"functions",
]
def on_start(self):
"""Called when a simulated user starts"""
try:
self.token = self.teacher_token
logger.info(f"TeacherUser started with token: {self.token}")
except Exception as e:
logger.error(f"TeacherUser on_start error: {e}")
@task(4)
def view_progress_report(self):
"""View student progress report"""
try:
with self.client.get(
"/api/progress-report.json",
name="/api/progress-report.json",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"view_progress_report error: {e}")
@task(2)
def export_progress_csv(self):
"""Export progress report as CSV"""
try:
with self.client.get(
"/api/progress-report/export-csv",
name="/api/progress-report/export-csv",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"export_progress_csv error: {e}")
@task(3)
def view_lessons_list(self):
"""View the list of lessons (for grading reference)"""
try:
with self.client.get(
"/api/lessons",
params={"token": self.token},
name="/api/lessons",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"view_lessons_list error: {e}")
@task(2)
def view_single_lesson(self):
"""View a single lesson content"""
try:
lesson = random.choice(self.lessons)
with self.client.get(
f"/api/lesson/{lesson}.json",
params={"token": self.token},
name="/api/lesson/[slug].json",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"view_single_lesson error: {e}")
@task(1)
def validate_token(self):
"""Validate teacher token"""
try:
with self.client.post(
"/api/validate-token",
json={"token": self.token},
name="/api/validate-token",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"validate_token error: {e}")
@task(1)
def track_student_progress(self):
"""Track progress for test student accounts (safe for testing)"""
try:
token = random.choice(self.student_tokens)
lesson = random.choice(self.lessons)
with self.client.post(
"/api/track-progress",
json={
"token": token,
"lesson_name": lesson,
"status": "completed",
},
name="/api/track-progress",
catch_response=True,
timeout=10,
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Status code: {response.status_code}")
except Exception as e:
logger.error(f"track_student_progress error: {e}")
class APIStressTest(HttpUser):
"""
Stress test specific API endpoints.
This user class focuses on hammering specific endpoints to find bottlenecks.
NOTE: Uses test student tokens - safe for production testing
These tokens are designated for testing purposes only
"""
wait_time = between(0.1, 0.5) # Very fast requests
weight = 1 # Lower weight for stress tests
# Test student tokens for stress testing
student_tokens = [
"CC90965E", "F84E3336", "D78F6DBD", "0D7B2566", "3ED63295",
]
lessons = ["hello_world", "variables", "conditions"]
@task(3)
def stress_compile_endpoint(self):
"""Stress test the compile endpoint"""
try:
code = """#include <stdio.h>
int main() { printf("Stress test\\n"); return 0; }"""
self.client.post(
"/api/compile",
json={"code": code, "language": "c"},
name="/api/compile [stress]",
timeout=30,
)
except Exception as e:
logger.error(f"stress_compile_endpoint error: {e}")
@task(2)
def stress_lesson_endpoint(self):
"""Stress test the lesson endpoint"""
try:
token = random.choice(self.student_tokens)
self.client.get(
"/api/lesson/hello_world.json",
params={"token": token},
name="/api/lesson/[slug].json [stress]",
timeout=10,
)
except Exception as e:
logger.error(f"stress_lesson_endpoint error: {e}")
@task(1)
def stress_validate_token(self):
"""Stress test the validate token endpoint"""
try:
token = random.choice(self.student_tokens)
self.client.post(
"/api/validate-token",
json={"token": token},
name="/api/validate-token [stress]",
timeout=10,
)
except Exception as e:
logger.error(f"stress_validate_token error: {e}")
@task(1)
def stress_lessons_list(self):
"""Stress test the lessons list endpoint"""
try:
token = random.choice(self.student_tokens)
self.client.get(
"/api/lessons",
params={"token": token},
name="/api/lessons [stress]",
timeout=10,
)
except Exception as e:
logger.error(f"stress_lessons_list error: {e}")
@task(1)
def stress_track_progress(self):
"""Stress test the progress tracking endpoint"""
try:
token = random.choice(self.student_tokens)
lesson = random.choice(self.lessons)
self.client.post(
"/api/track-progress",
json={
"token": token,
"lesson_name": lesson,
"status": "completed",
},
name="/api/track-progress [stress]",
timeout=10,
)
except Exception as e:
logger.error(f"stress_track_progress error: {e}")
# Event hooks for custom logging
@events.request.add_listener
def on_request(request_type, name, response_time, response_length, response, context, exception, **kwargs):
"""Log slow requests"""
if response_time > 1000: # More than 1 second
logger.warning(f"⚠️ SLOW REQUEST: {name} took {response_time}ms")
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
"""Called when load test starts"""
logger.info("🚀 Starting load test for Elemes LMS")
logger.info(f"📊 Target host: {environment.host}")
logger.info("🔒 Using teacher token only (safe for production testing)")
# Log user class configuration
user_classes = environment.user_classes
total_weight = sum(uc.weight for uc in user_classes if hasattr(uc, 'weight'))
logger.info(f"👥 User classes loaded: {[uc.__name__ for uc in user_classes]}")
logger.info(f"⚖️ Total weight: {total_weight}")
for uc in user_classes:
if hasattr(uc, 'weight'):
logger.info(f" - {uc.__name__}: weight={uc.weight}")
@events.spawning_complete.add_listener
def on_spawning_complete(environment, user_count, **kwargs):
"""Called when spawning is complete"""
logger.info(f"✅ Spawning complete: {user_count} users active")
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
"""Called when load test stops"""
logger.info("✅ Load test completed")
if environment.stats.total.num_requests > 0:
fail_rate = (environment.stats.total.num_failures / environment.stats.total.num_requests) * 100
logger.info(f"📈 Total requests: {environment.stats.total.num_requests}")
logger.info(f"❌ Failures: {environment.stats.total.num_failures} ({fail_rate:.2f}%)")
logger.info(f"⏱️ Average response time: {environment.stats.total.avg_response_time:.2f}ms")