462 lines
16 KiB
Python
462 lines
16 KiB
Python
"""
|
|
Locust E2E Test Suite for Elemes LMS.
|
|
|
|
Auto-generated test scenarios from content/ folder via content_parser.py.
|
|
Reads test_data.json to know which lessons exist and their types.
|
|
|
|
Usage (from elemes/load-test/):
|
|
1. python content_parser.py # generate test_data.json
|
|
2. locust -f locustfile.py # open web UI at http://localhost:8089
|
|
→ Set host to your Elemes URL in the web UI
|
|
|
|
Environment variables:
|
|
API_PREFIX — Prefix for API paths (default: '/api')
|
|
Use '/api' when targeting SvelteKit frontend (:3000)
|
|
Use '' when targeting Flask backend directly (:5000)
|
|
VELXIO_HOST — Velxio backend URL for Arduino compilation
|
|
(default: http://localhost:8001)
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import random
|
|
import time
|
|
|
|
import requests
|
|
from requests.exceptions import JSONDecodeError
|
|
from locust import HttpUser, task, between, events
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Load test data
|
|
# ---------------------------------------------------------------------------
|
|
|
|
TEST_DATA_FILE = os.path.join(os.path.dirname(__file__), 'test_data.json')
|
|
|
|
def load_test_data() -> dict:
|
|
"""Load test_data.json generated by content_parser.py."""
|
|
if not os.path.exists(TEST_DATA_FILE):
|
|
raise FileNotFoundError(
|
|
f"{TEST_DATA_FILE} not found. "
|
|
f"Run 'python content_parser.py' first."
|
|
)
|
|
with open(TEST_DATA_FILE, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
|
|
try:
|
|
TEST_DATA = load_test_data()
|
|
except FileNotFoundError as e:
|
|
logging.error(str(e))
|
|
TEST_DATA = {'token': '', 'teacher_token': '', 'lessons': []}
|
|
|
|
TOKENS = TEST_DATA.get('tokens', [])
|
|
TEACHER_TOKEN = TEST_DATA.get('teacher_token', '')
|
|
ALL_LESSONS = TEST_DATA.get('lessons', [])
|
|
|
|
# API prefix: '/api' for SvelteKit (default), '' for direct Flask
|
|
API = os.environ.get('API_PREFIX', '/api')
|
|
|
|
# Velxio backend URL is determined dynamically via SvelteKit / Tailscale reverse proxy path /velxio/
|
|
|
|
# Pre-filter lessons by type for task assignment
|
|
C_LESSONS = [l for l in ALL_LESSONS if l['type'] == 'c' and l.get('solution_code')]
|
|
PYTHON_LESSONS = [l for l in ALL_LESSONS if l['type'] == 'python' and l.get('solution_python')]
|
|
HYBRID_LESSONS = [l for l in ALL_LESSONS if l['type'] == 'hybrid']
|
|
ARDUINO_LESSONS = [l for l in ALL_LESSONS if l['type'] == 'arduino']
|
|
COMPILABLE_LESSONS = C_LESSONS + PYTHON_LESSONS + [
|
|
l for l in HYBRID_LESSONS if l.get('solution_code') or l.get('solution_python')
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def check_key_text(code: str, key_text: str) -> bool:
|
|
"""Client-side key_text check (mirrors exercise.ts logic)."""
|
|
if not key_text.strip():
|
|
return True
|
|
keys = [k.strip() for k in key_text.split('\n') if k.strip()]
|
|
return all(k in code for k in keys)
|
|
|
|
|
|
def safe_json(resp) -> dict:
|
|
"""Safely decode JSON from a response, handling HTML/404 errors."""
|
|
try:
|
|
return resp.json()
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Locust User
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ElemesStudent(HttpUser):
|
|
"""
|
|
Simulates a student interacting with the Elemes LMS.
|
|
|
|
Each user logs in once (on_start), then performs weighted tasks
|
|
that mirror real student behavior.
|
|
"""
|
|
|
|
wait_time = between(1, 5)
|
|
|
|
def on_start(self):
|
|
"""Login once when user spawns."""
|
|
self.token = random.choice(TOKENS) if TOKENS else ''
|
|
with self.client.post(
|
|
f'{API}/login',
|
|
json={'token': self.token},
|
|
name='/login',
|
|
catch_response=True
|
|
) as resp:
|
|
data = safe_json(resp)
|
|
if not data.get('success'):
|
|
resp.failure(f"Login failed: {data.get('message', f'HTTP {resp.status_code}')}")
|
|
|
|
# ── Task 1: Browse Lessons (weight=3) ──────────────────────────────
|
|
|
|
@task(3)
|
|
def browse_lessons(self):
|
|
"""Fetch lesson list — simulates landing on home page."""
|
|
with self.client.get(f'{API}/lessons', name='/lessons', catch_response=True) as resp:
|
|
data = safe_json(resp)
|
|
lessons = data.get('lessons', [])
|
|
if len(lessons) == 0:
|
|
resp.failure("No lessons returned")
|
|
elif len(lessons) != len(ALL_LESSONS):
|
|
resp.failure(
|
|
f"Expected {len(ALL_LESSONS)} lessons, got {len(lessons)}"
|
|
)
|
|
|
|
# ── Task 2: View Lesson Detail (weight=5) ──────────────────────────
|
|
|
|
@task(5)
|
|
def view_lesson_detail(self):
|
|
"""Fetch a random lesson's detail JSON."""
|
|
if not ALL_LESSONS:
|
|
return
|
|
|
|
lesson = random.choice(ALL_LESSONS)
|
|
slug = lesson['slug']
|
|
|
|
with self.client.get(
|
|
f'{API}/lesson/{slug}.json',
|
|
name='/lesson/[slug].json',
|
|
catch_response=True
|
|
) as resp:
|
|
data = safe_json(resp)
|
|
|
|
if 'error' in data or not data:
|
|
resp.failure(f"Lesson {slug} not found")
|
|
return
|
|
|
|
# Validate fields based on type
|
|
if lesson['type'] == 'arduino':
|
|
if not data.get('initial_code_arduino'):
|
|
resp.failure(f"{slug}: missing initial_code_arduino")
|
|
if not data.get('velxio_circuit'):
|
|
resp.failure(f"{slug}: missing velxio_circuit")
|
|
# Validate velxio_circuit is valid JSON
|
|
try:
|
|
json.loads(data['velxio_circuit'])
|
|
except (json.JSONDecodeError, TypeError):
|
|
resp.failure(f"{slug}: velxio_circuit is not valid JSON")
|
|
elif lesson['type'] in ('c', 'hybrid'):
|
|
if not data.get('initial_code'):
|
|
resp.failure(f"{slug}: missing initial_code")
|
|
elif lesson['type'] == 'python':
|
|
if not data.get('initial_python') and not data.get('initial_code'):
|
|
resp.failure(f"{slug}: missing initial code")
|
|
|
|
# ── Task 3: Compile C Lesson (weight=4) ────────────────────────────
|
|
|
|
@task(4)
|
|
def compile_c_lesson(self):
|
|
"""Compile a C lesson's solution and validate output."""
|
|
if not C_LESSONS:
|
|
return
|
|
|
|
lesson = random.choice(C_LESSONS)
|
|
code = lesson.get('solution_code', lesson.get('initial_code_c', ''))
|
|
expected = lesson.get('expected_output', '')
|
|
|
|
if not code:
|
|
return
|
|
|
|
with self.client.post(
|
|
f'{API}/compile',
|
|
json={'code': code, 'language': 'c'},
|
|
name='/compile [C]',
|
|
catch_response=True
|
|
) as resp:
|
|
data = safe_json(resp)
|
|
|
|
if not data.get('success'):
|
|
resp.failure(f"Compile failed: {data.get('error', 'unknown')}")
|
|
return
|
|
|
|
output = data.get('output', '')
|
|
|
|
# Validate expected output
|
|
if expected and expected.strip() not in output:
|
|
resp.failure(
|
|
f"{lesson['slug']}: expected '{expected.strip()[:50]}' "
|
|
f"not in output '{output[:50]}'"
|
|
)
|
|
return
|
|
|
|
# Validate key_text
|
|
key_text = lesson.get('key_text', '')
|
|
if key_text:
|
|
# Untuk hybrid, gabungkan dengan code python agar semua key_text terpenuhi
|
|
full_code = code + "\n" + lesson.get('solution_python', '')
|
|
if not check_key_text(full_code, key_text):
|
|
resp.failure(f"{lesson['slug']}: key_text check failed")
|
|
|
|
# ── Task 4: Compile Python Lesson (weight=3) ───────────────────────
|
|
|
|
@task(3)
|
|
def compile_python_lesson(self):
|
|
"""Compile a Python lesson's solution and validate output."""
|
|
if not PYTHON_LESSONS:
|
|
return
|
|
|
|
lesson = random.choice(PYTHON_LESSONS)
|
|
code = lesson.get('solution_python', lesson.get('initial_python', ''))
|
|
expected = lesson.get('expected_output_python', lesson.get('expected_output', ''))
|
|
|
|
if not code:
|
|
return
|
|
|
|
with self.client.post(
|
|
f'{API}/compile',
|
|
json={'code': code, 'language': 'python'},
|
|
name='/compile [Python]',
|
|
catch_response=True
|
|
) as resp:
|
|
data = safe_json(resp)
|
|
|
|
if not data.get('success'):
|
|
resp.failure(f"Compile failed: {data.get('error', 'unknown')}")
|
|
return
|
|
|
|
output = data.get('output', '')
|
|
if expected and expected.strip() not in output:
|
|
resp.failure(
|
|
f"{lesson['slug']}: expected '{expected.strip()[:50]}' "
|
|
f"not in output '{output[:50]}'"
|
|
)
|
|
|
|
# ── Task 5: Verify Arduino Lesson Structure (weight=2) ─────────────
|
|
|
|
@task(2)
|
|
def verify_arduino_lesson(self):
|
|
"""
|
|
Verify an Arduino lesson's data structure.
|
|
Validates JSON integrity of velxio_circuit and expected_wiring.
|
|
"""
|
|
if not ARDUINO_LESSONS:
|
|
return
|
|
|
|
lesson = random.choice(ARDUINO_LESSONS)
|
|
slug = lesson['slug']
|
|
|
|
with self.client.get(
|
|
f'{API}/lesson/{slug}.json',
|
|
name='/lesson/[arduino].json',
|
|
catch_response=True
|
|
) as resp:
|
|
data = safe_json(resp)
|
|
|
|
errors = []
|
|
|
|
# Check required fields
|
|
if not data.get('initial_code_arduino'):
|
|
errors.append('missing initial_code_arduino')
|
|
|
|
vc = data.get('velxio_circuit', '')
|
|
if vc:
|
|
try:
|
|
circuit = json.loads(vc)
|
|
if 'board' not in circuit:
|
|
errors.append('velxio_circuit missing board field')
|
|
except (json.JSONDecodeError, TypeError):
|
|
errors.append('velxio_circuit invalid JSON')
|
|
else:
|
|
errors.append('missing velxio_circuit')
|
|
|
|
# Check expected_wiring JSON validity if present
|
|
ew = data.get('expected_wiring', '')
|
|
if ew:
|
|
try:
|
|
wiring = json.loads(ew)
|
|
if 'wires' not in wiring:
|
|
errors.append('expected_wiring missing wires array')
|
|
except (json.JSONDecodeError, TypeError):
|
|
errors.append('expected_wiring invalid JSON')
|
|
|
|
|
|
if errors:
|
|
resp.failure(f"{slug}: {'; '.join(errors)}")
|
|
|
|
# ── Task 8: Compile Arduino via Velxio (weight=3) ──────────────────
|
|
|
|
@task(3)
|
|
def compile_arduino_lesson(self):
|
|
"""
|
|
Compile an Arduino lesson's code via Velxio backend.
|
|
Hits Velxio's POST /velxio/api/compile endpoint.
|
|
Validates: compile success + hex_content returned.
|
|
Note: serial output can only be validated in-browser (avr8js sim).
|
|
"""
|
|
if not ARDUINO_LESSONS:
|
|
return
|
|
|
|
lesson = random.choice(ARDUINO_LESSONS)
|
|
code = lesson.get('initial_code_arduino', '')
|
|
if not code:
|
|
return
|
|
|
|
with self.client.post(
|
|
'/velxio/api/compile',
|
|
json={'code': code, 'board_fqbn': 'arduino:avr:uno'},
|
|
name='/velxio/api/compile [Arduino]',
|
|
timeout=30,
|
|
catch_response=True
|
|
) as resp:
|
|
data = safe_json(resp)
|
|
|
|
if not data.get('success'):
|
|
# Report as Locust failure for stats tracking
|
|
resp.failure(
|
|
f"{lesson['slug']}: compile failed — "
|
|
f"{data.get('error', data.get('stderr', 'unknown'))}"
|
|
)
|
|
return
|
|
|
|
hex_content = data.get('hex_content', '')
|
|
if not hex_content:
|
|
resp.failure(f"{lesson['slug']}: no hex_content")
|
|
return
|
|
|
|
# ── Task 6: Complete Lesson Flow (weight=2) ────────────────────────
|
|
|
|
@task(2)
|
|
def complete_lesson_flow(self):
|
|
"""
|
|
Full flow: fetch lesson → compile (if possible) → track progress.
|
|
Simulates a student completing a lesson.
|
|
"""
|
|
if not ALL_LESSONS:
|
|
return
|
|
|
|
lesson = random.choice(ALL_LESSONS)
|
|
slug = lesson['slug']
|
|
|
|
# 1. Fetch lesson detail
|
|
resp = self.client.get(
|
|
f'{API}/lesson/{slug}.json',
|
|
name='/lesson/[slug].json (flow)'
|
|
)
|
|
if resp.status_code != 200:
|
|
return
|
|
|
|
# 2. Compile if it's a C or Python lesson with solution
|
|
if lesson['type'] in ('c', 'hybrid') and lesson.get('solution_code'):
|
|
self.client.post(
|
|
f'{API}/compile',
|
|
json={'code': lesson['solution_code'], 'language': 'c'},
|
|
name='/compile (flow)'
|
|
)
|
|
time.sleep(0.5) # Simulate student reading output
|
|
|
|
elif lesson['type'] == 'python' and lesson.get('solution_python'):
|
|
self.client.post(
|
|
f'{API}/compile',
|
|
json={'code': lesson['solution_python'], 'language': 'python'},
|
|
name='/compile (flow)'
|
|
)
|
|
time.sleep(0.5)
|
|
|
|
elif lesson['type'] == 'arduino' and lesson.get('initial_code_arduino'):
|
|
self.client.post(
|
|
'/velxio/api/compile',
|
|
json={'code': lesson['initial_code_arduino'],
|
|
'board_fqbn': 'arduino:avr:uno'},
|
|
name='/velxio/api/compile (flow)',
|
|
timeout=30
|
|
)
|
|
time.sleep(0.5)
|
|
|
|
# 3. Track progress
|
|
with self.client.post(
|
|
f'{API}/track-progress',
|
|
json={
|
|
'token': self.token,
|
|
'lesson_name': slug,
|
|
'status': 'completed'
|
|
},
|
|
name='/track-progress',
|
|
catch_response=True
|
|
) as resp:
|
|
data = safe_json(resp)
|
|
if not data.get('success'):
|
|
resp.failure(f"Track progress failed: {data.get('message', f'HTTP {resp.status_code}')}")
|
|
|
|
# ── Task 7: Progress Report (weight=1) ─────────────────────────────
|
|
|
|
@task(1)
|
|
def view_progress_report(self):
|
|
"""Fetch progress report (teacher perspective)."""
|
|
if not TEACHER_TOKEN:
|
|
return
|
|
|
|
# Login as teacher
|
|
self.client.post(
|
|
f'{API}/login',
|
|
json={'token': TEACHER_TOKEN},
|
|
name='/login [teacher]'
|
|
)
|
|
|
|
with self.client.get(
|
|
f'{API}/progress-report.json',
|
|
name='/progress-report.json',
|
|
catch_response=True
|
|
) as resp:
|
|
data = safe_json(resp)
|
|
students = data.get('students', [])
|
|
lessons = data.get('lessons', [])
|
|
|
|
if not lessons:
|
|
resp.failure("No lessons in progress report")
|
|
|
|
# Re-login as student for subsequent tasks
|
|
self.client.post(
|
|
f'{API}/login',
|
|
json={'token': self.token},
|
|
name='/login [re-auth]'
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Event hooks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@events.test_start.add_listener
|
|
def on_test_start(environment, **kwargs):
|
|
"""Log test configuration at start."""
|
|
stats = TEST_DATA.get('stats', {})
|
|
logging.info("=" * 60)
|
|
logging.info(" Elemes LMS — Locust E2E Test")
|
|
logging.info("=" * 60)
|
|
logging.info(f" Lessons: {stats.get('total', 0)} total")
|
|
logging.info(f" C: {stats.get('c', 0)}, Python: {stats.get('python', 0)}, "
|
|
f"Hybrid: {stats.get('hybrid', 0)}, Arduino: {stats.get('arduino', 0)}")
|
|
logging.info(f" Compilable: {stats.get('compilable', 0)}")
|
|
logging.info(f" Tokens: {len(TOKENS)} loaded")
|
|
logging.info(f" API prefix: '{API}'")
|
|
logging.info("=" * 60)
|