Распределение вариантов
This commit is contained in:
203
tests/smoke_test.py
Normal file
203
tests/smoke_test.py
Normal file
@@ -0,0 +1,203 @@
|
||||
"""Smoke-тест минимального ядра: админ, студенты, варианты лабы, распределение.
|
||||
|
||||
Поднимает оба микросервиса на портах 8100/8101 с чистой БД и прогоняет
|
||||
основной сценарий. Запуск: make test (или python tests/smoke_test.py).
|
||||
"""
|
||||
|
||||
import os
|
||||
import re as _re
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.parse
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
API_PORT = int(os.environ.get("LMS_TEST_API_PORT", "8100"))
|
||||
WEB_PORT = int(os.environ.get("LMS_TEST_WEB_PORT", "8101"))
|
||||
API_PY = os.environ.get("LMS_TEST_API_PY", sys.executable)
|
||||
WEB_PY = os.environ.get("LMS_TEST_WEB_PY", sys.executable)
|
||||
|
||||
PASSED = []
|
||||
|
||||
|
||||
def ok(name):
|
||||
PASSED.append(name)
|
||||
print("PASS:", name)
|
||||
|
||||
|
||||
def free(port):
|
||||
s = socket.socket()
|
||||
try:
|
||||
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
s.bind(("127.0.0.1", port))
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def start_services(data_dir):
|
||||
env = dict(os.environ, LMS_API_PORT=str(API_PORT), FRONTEND_PORT=str(WEB_PORT),
|
||||
LMS_API_URL=f"http://127.0.0.1:{API_PORT}", LMS_DATA_DIR=str(data_dir))
|
||||
procs = [
|
||||
subprocess.Popen(
|
||||
[API_PY, "-m", "uvicorn", "lms_api.main:app", "--app-dir", str(ROOT / "src" / "backend"),
|
||||
"--port", str(API_PORT), "--host", "127.0.0.1"],
|
||||
env=env, cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL),
|
||||
subprocess.Popen(
|
||||
[WEB_PY, "-m", "uvicorn", "lms_web.main:app", "--app-dir", str(ROOT / "src" / "frontend"),
|
||||
"--port", str(WEB_PORT), "--host", "127.0.0.1"],
|
||||
env=env, cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL),
|
||||
]
|
||||
for _ in range(60):
|
||||
try:
|
||||
if httpx.get(f"http://127.0.0.1:{API_PORT}/healthz", timeout=2).status_code == 200:
|
||||
break
|
||||
except httpx.HTTPError:
|
||||
time.sleep(0.5)
|
||||
for _ in range(60):
|
||||
try:
|
||||
if httpx.get(f"http://127.0.0.1:{WEB_PORT}/healthz", timeout=2).status_code == 200:
|
||||
break
|
||||
except httpx.HTTPError:
|
||||
time.sleep(0.5)
|
||||
return procs
|
||||
|
||||
|
||||
def main():
|
||||
for port in (API_PORT, WEB_PORT):
|
||||
assert free(port), f"Порт {port} занят"
|
||||
data_dir = Path(tempfile.mkdtemp(prefix="lms-test-"))
|
||||
procs = start_services(data_dir)
|
||||
try:
|
||||
run(data_dir)
|
||||
finally:
|
||||
for p in procs:
|
||||
p.terminate()
|
||||
for p in procs:
|
||||
try:
|
||||
p.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
p.kill()
|
||||
shutil.rmtree(data_dir, ignore_errors=True)
|
||||
print(f"\nПРОВЕРОК: {len(PASSED)}")
|
||||
|
||||
|
||||
def run(data_dir: Path):
|
||||
db_path = data_dir / "lms.db"
|
||||
|
||||
def dbq(sql, *a):
|
||||
import sqlite3
|
||||
|
||||
db = sqlite3.connect(db_path)
|
||||
db.row_factory = sqlite3.Row
|
||||
try:
|
||||
return [dict(x) for x in db.execute(sql, a).fetchall()]
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
web = f"http://127.0.0.1:{WEB_PORT}"
|
||||
api = f"http://127.0.0.1:{API_PORT}/api/v1"
|
||||
c = httpx.Client(base_url=web, follow_redirects=False, timeout=30)
|
||||
|
||||
# --- админ: первый запуск
|
||||
r = c.get("/")
|
||||
assert r.status_code == 200 and "Первый запуск" in r.text
|
||||
ok("setup page")
|
||||
r = c.post("/setup", data={"login": "teach", "password": "teacher-123", "password2": "teacher-123"})
|
||||
assert r.status_code == 303
|
||||
ok("admin created")
|
||||
r = c.get("/t/labs", follow_redirects=True)
|
||||
assert r.status_code == 200 and "Лабораторные работы" in r.text
|
||||
ok("admin lands on labs list")
|
||||
|
||||
h = httpx.Client(base_url=api, timeout=30)
|
||||
tok = h.post("/auth/login", json={"login": "teach", "password": "teacher-123"}).json()["token"]
|
||||
H = {"Authorization": f"Bearer {tok}"}
|
||||
assert h.get("/groups", headers=H).status_code == 200
|
||||
assert h.get("/groups").status_code == 401
|
||||
ok("api auth enforced")
|
||||
|
||||
# --- студенты из групп
|
||||
for fname in ("ПИ-11БО.xls", "ПИ-12БО.xls"):
|
||||
with open(ROOT / "examples" / fname, "rb") as f:
|
||||
r = c.post("/t/groups/import", files={"file": (fname, f, "application/vnd.ms-excel")})
|
||||
loc = urllib.parse.unquote(r.headers["location"])
|
||||
assert "добавлено 20" in loc, loc
|
||||
gid1 = c.get("/t/groups").text.split("/t/groups/")[1].split('"')[0]
|
||||
ok("two groups imported (20+20 students)")
|
||||
|
||||
r = c.get(f"/t/groups/{gid1}/credentials.csv")
|
||||
rows = [l for l in r.text.splitlines() if l.strip()][1:]
|
||||
pwmap = {row.split(";")[1]: row.split(";")[2] for row in rows}
|
||||
assert len(pwmap) == 20
|
||||
ok("credentials export with generated passwords")
|
||||
|
||||
# --- лабораторная из JSON (только варианты)
|
||||
raw_lab = (ROOT / "examples" / "lab1.json").read_text(encoding="utf-8")
|
||||
r = c.post("/t/labs/new/import", files={"file": ("lab1.json", raw_lab.encode(), "application/json")})
|
||||
assert r.status_code == 200 and "Вариантов" in r.text
|
||||
from html import unescape as _un
|
||||
vj = _un(_re.search(r'name="variants_json" value="([^"]*)"', r.text).group(1))
|
||||
r = c.post("/t/labs/new", data={"variants_json": vj})
|
||||
assert r.status_code == 303, r.text[:200]
|
||||
lid = r.headers["location"].split("/labs/")[1].split("/")[0]
|
||||
assert "вариантов: 30" in urllib.parse.unquote(r.headers["location"])
|
||||
ok("lab created from JSON with 30 variants")
|
||||
|
||||
# --- уровни и распределение
|
||||
s3 = dbq("SELECT s.id_student FROM group_students gs JOIN students s ON s.id_student=gs.student_id "
|
||||
"WHERE gs.group_id=? ORDER BY CAST(s.id_student AS INTEGER) LIMIT 3", int(gid1))
|
||||
for st in s3:
|
||||
c.post(f"/t/groups/{gid1}/levels", data={"student_id": st["id_student"], "level": "advanced"})
|
||||
gids = [r["id"] for r in dbq("SELECT id FROM groups ORDER BY id")]
|
||||
for gid in gids:
|
||||
c.post(f"/t/groups/{gid}/assign", data={"lab_id": lid})
|
||||
a1 = dbq("SELECT sa.level, la.number FROM student_assignment sa "
|
||||
"JOIN group_students gs ON gs.group_id=? AND gs.student_id=sa.student_id "
|
||||
"LEFT JOIN lab_assignments la ON la.student_id=sa.student_id AND la.lab_id=?",
|
||||
int(gid1), int(lid))
|
||||
adv = [x["number"] for x in a1 if x["level"] == "advanced"]
|
||||
main = [x["number"] for x in a1 if x["level"] == "main"]
|
||||
assert sorted(int(x) for x in adv) == [21, 22, 23], adv
|
||||
assert len(main) == 17 and len(set(main)) == 17 and all(1 <= int(x) <= 20 for x in main), main
|
||||
ok("variants distributed: unique within group, pool 1..20 / 21..30")
|
||||
|
||||
man_st = next(x for x in dbq("SELECT student_id, level FROM student_assignment WHERE group_id=?",
|
||||
int(gid1)) if x["level"] == "main")["student_id"]
|
||||
c.post(f"/t/groups/{gid1}/manual", data={"student_id": man_st, "lab_id": lid, "number": "5"})
|
||||
m = dbq("SELECT number, manual FROM lab_assignments WHERE student_id=? AND lab_id=?",
|
||||
man_st, int(lid))[0]
|
||||
assert m["number"] == "5" and m["manual"] == 1
|
||||
ok("manual variant per lab")
|
||||
|
||||
# --- студент: вход, смена пароля, условие в личном кабинете
|
||||
pw = pwmap[man_st]
|
||||
r = c.post("/login", data={"login": man_st, "password": pw})
|
||||
assert r.status_code == 303 and r.headers["location"].startswith("/change-password")
|
||||
c.post("/change-password", data={"old": pw, "new": "newpass-123", "new2": "newpass-123"})
|
||||
r = c.get("/s/", follow_redirects=True)
|
||||
assert r.status_code == 200 and "Мои лабораторные" in r.text
|
||||
r = c.get(f"/s/labs/{lid}", follow_redirects=True)
|
||||
assert "вариант 5" in r.text or "Вариант 5" in r.text
|
||||
assert "Отсортировать массив" in r.text
|
||||
ok("student sees own variant statement after login")
|
||||
|
||||
# --- доступы
|
||||
noauth = httpx.Client(base_url=web, follow_redirects=False, timeout=15)
|
||||
assert noauth.get("/t/labs").status_code == 303
|
||||
c.post("/logout")
|
||||
r = c.post("/login", data={"login": "teach", "password": "teacher-123"})
|
||||
assert r.status_code == 303
|
||||
ok("auth enforced")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user