#!/usr/bin/env python """The world the end-to-end tests wake up in. Small on purpose. Every row here exists because a test names it, and a fixture that drifts past what the tests actually use becomes a second application to maintain — one whose bugs look like product bugs. Idempotent: running it twice leaves the same world, so a developer can re-seed between runs without tearing the stack down. """ import os import sys from datetime import datetime sys.path.insert(0, "/app") from sqlalchemy import text # noqa: E402 from app.database import Base, SessionLocal, engine # noqa: E402 from app.models.article import Article # noqa: E402 from app.models.email_verification import EmailVerification # noqa: E402 from app.models.exam import Exam, QuestionExamLink # noqa: E402 from app.models.flashcard import Flashcard, FlashcardDeck # noqa: E402 from app.models.question import Question # noqa: E402 from app.models.question_category import QuestionCategory # noqa: E402 from app.models.quiz import Quiz # noqa: E402 from app.models.quiz_question_link import QuizQuestionLink # noqa: E402 from app.models.user import User # noqa: E402 from app.utils.auth import get_password_hash # noqa: E402 #: The two accounts the suite signs in as. `.example.com` rather than `.test`, #: because EmailStr refuses reserved TLDs and sign-in would 422 before it ever #: reached the password check. #: Fixed by default and in the #: repository, because the tests have to know them and this database is #: created and destroyed by the run. Override either from the environment when #: you want a stack you can poke at by hand without the passwords being #: something anybody reading the repo already knows. EDUCATOR = (os.environ.get("E2E_EDUCATOR_EMAIL", "educator@e2e.example.com"), os.environ.get("E2E_EDUCATOR_PASSWORD", "e2e-educator-password")) LEARNER = (os.environ.get("E2E_LEARNER_EMAIL", "learner@e2e.example.com"), os.environ.get("E2E_LEARNER_PASSWORD", "e2e-learner-password")) STEMS = [ ("A 3-year-old has a barking cough, stridor at rest and no drooling. " "The most appropriate next step is", ["Dexamethasone", "Intubation", "Nebulised saline", "Ceftriaxone", "Racemic adrenaline only"], "Dexamethasone", "A single dose of dexamethasone shortens croup at every severity, and a " "child with stridor at rest has moderate croup."), ("A 6-month-old with bronchiolitis has intermittent apnoea and feeds at " "half volume. The next step is", ["Admit for observation", "Discharge with advice", "Oral antibiotics", "Chest radiograph", "Salbutamol"], "Admit for observation", "Apnoea and poor feeding are the two admission criteria that matter most " "in the first six months."), ("A neonate at 36 hours has a total bilirubin above the phototherapy line. " "The next step is", ["Start phototherapy", "Repeat in 24 hours", "Exchange transfusion", "Stop breastfeeding", "Intravenous immunoglobulin"], "Start phototherapy", "The threshold line is the decision. Repeating the level to see whether " "it climbs is how a treatable jaundice becomes kernicterus."), ] def verified(db, user): row = db.query(EmailVerification).filter(EmailVerification.user_id == user.id).first() if row is None: db.add(EmailVerification(user_id=user.id, token=f"seed-{user.id}", expires_at=datetime(2099, 1, 1), verified_at=datetime.utcnow())) elif row.verified_at is None: row.verified_at = datetime.utcnow() def account(db, email, password, role): user = db.query(User).filter(User.email == email).first() if user is None: user = User(email=email, name=email.split("@")[0].title(), hashed_password=get_password_hash(password), role=role) db.add(user) db.flush() user.role = role user.hashed_password = get_password_hash(password) verified(db, user) return user #: A drawing written the way the real ones are: a viewBox, and no width or #: height. That is the shape that renders as nothing in Safari inside a #: shrink-to-fit box, so the figure test is only worth having if the fixture #: has the same defect the real files have. FIGURE = """ Stridor at rest Stridor on exertion """ def figure(db, educator): """One drawing, in the library and in an article that renders it.""" from app.models.media import MediaAsset path = "figures/e2e-stridor.svg" target = os.path.join("/app/uploads", path) os.makedirs(os.path.dirname(target), exist_ok=True) with open(target, "w", encoding="utf-8") as handle: handle.write(FIGURE) asset = db.query(MediaAsset).filter(MediaAsset.path == path).first() if asset is None: db.add(MediaAsset( path=path, title="Stridor at rest and on exertion", caption="Which of the two decides whether croup is treated now.", alt_text="Two panels comparing stridor at rest with stridor on exertion.", source="PedsHub", user_id=educator.id, )) article = db.query(Article).filter(Article.slug == "croup").first() if article and "e2e-stridor" not in (article.content or ""): article.content = ( f"![Two panels comparing stridor at rest with stridor on exertion.]" f"(/uploads/{path})\n\n" + (article.content or "")) def objective(db, users): """The exam every seeded account is already studying for. Set here rather than clicked in the tests, because "what are you studying for?" is a modal over the first page a signed-in person sees — and a modal that four parallel workers race each other to dismiss, on one shared account, produces failures that say nothing about the product. """ exam = db.query(Exam).filter(Exam.slug == "e2e-boards").first() if exam is None: exam = Exam(slug="e2e-boards", name="E2E Boards", family="Boards", sort_order=1, is_active=1) db.add(exam) db.flush() for (question_id,) in db.query(Question.id).all(): if not db.query(QuestionExamLink).filter_by( question_id=question_id, exam_id=exam.id).first(): db.add(QuestionExamLink(question_id=question_id, exam_id=exam.id)) for user in users: user.active_exam_id = exam.id return exam #: One learner per parallel worker. They share a database, and a session #: remembers where it was left — so two workers sitting the same session on the #: same account interfere in ways that look like product bugs and are not. WORKERS = 4 def worker_learners(db): made = [] for index in range(WORKERS): email = LEARNER[0].replace("@", f"+w{index}@") made.append(account(db, email, LEARNER[1], role="user")) return made def main() -> int: # The stack is empty on first boot, so the schema comes from the models and # is then stamped: exactly what a fresh deploy does. Base.metadata.create_all(bind=engine) db = SessionLocal() try: educator = account(db, *EDUCATOR, role="moderator") account(db, *LEARNER, role="user") learners = worker_learners(db) db.commit() if db.query(QuestionCategory).count() == 0: db.add_all([ QuestionCategory(id=1, name="Respiratory", user_id=educator.id), QuestionCategory(id=2, name="Croup", parent_id=1, user_id=educator.id), QuestionCategory(id=3, name="Neonatology", user_id=educator.id), ]) db.flush() if db.query(Question).count() == 0: for index, (stem, options, answer, explanation) in enumerate(STEMS, start=1): db.add(Question( id=index, question_text=stem, question_type="mcq", options=options, correct_answer=answer, explanation=explanation, question_category_id=2 if index < 3 else 3, difficulty=["easy", "medium", "hard"][index - 1], user_id=educator.id, )) db.flush() if db.query(Quiz).count() == 0: quiz = Quiz(id=1, title="E2E study session", user_id=educator.id, mode="learning", questions_count=len(STEMS), is_published=1, is_shared=1) db.add(quiz) db.flush() for position, question_id in enumerate(range(1, len(STEMS) + 1)): db.add(QuizQuestionLink(quiz_id=quiz.id, question_id=question_id, position=position)) if db.query(Article).count() == 0: db.add(Article( id=1, title="Croup", slug="croup", status="published", user_id=educator.id, category_id=2, summary="Barking cough, stridor, and one dose of dexamethasone.", content="Croup is viral laryngotracheobronchitis. See [[2|bronchiolitis]].", sections=[{"id": "workup", "title": "Workup", "content": "Clinical. A radiograph is for the child who is not croup."}], )) db.add(Article( id=2, title="Bronchiolitis", slug="bronchiolitis", status="published", user_id=educator.id, category_id=1, summary="Supportive care, and the two reasons to admit.", content="Bronchiolitis is a first-winter illness of small airways.", sections=[], )) if db.query(FlashcardDeck).count() == 0: deck = FlashcardDeck(id=1, title="Cards: Croup", user_id=educator.id, card_count=2, is_shared=1) db.add(deck) db.flush() db.add_all([ Flashcard(deck_id=deck.id, front="Dose of dexamethasone in croup?", back="0.15–0.6 mg/kg once, oral."), Flashcard(deck_id=deck.id, front="Stridor at rest means?", back="At least moderate croup — treat, do not watch."), ]) db.flush() figure(db, educator) objective(db, [educator, db.query(User).filter(User.email == LEARNER[0]).one(), *learners]) db.commit() # Stamped rather than migrated: the schema came from the models a moment # ago, so replaying every migration over it would fail on the first # CREATE TABLE. This is what a fresh deploy of the real app does too. with engine.begin() as conn: conn.execute(text("CREATE TABLE IF NOT EXISTS alembic_version " "(version_num VARCHAR(32) NOT NULL)")) head = os.environ.get("ALEMBIC_HEAD", "p5f6a7b8c9d0") if not conn.execute(text("SELECT 1 FROM alembic_version")).first(): conn.execute(text("INSERT INTO alembic_version VALUES (:v)"), {"v": head}) counts = { "users": db.query(User).count(), "questions": db.query(Question).count(), "quizzes": db.query(Quiz).count(), "articles": db.query(Article).count(), "cards": db.query(Flashcard).count(), } # Printed, so `docker compose -f docker-compose.test.yml logs seed` # says how to sign in without anybody reading the source. print("seeded", counts, flush=True) print(f" educator: {EDUCATOR[0]} / {EDUCATOR[1]}", flush=True) print(f" learner: {LEARNER[0]} / {LEARNER[1]}", flush=True) return 0 finally: db.close() if __name__ == "__main__": raise SystemExit(main())