204 lines
8.2 KiB
Python
204 lines
8.2 KiB
Python
"""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()
|