"""End-to-end smoke test for the **professional interactive course plans** flow. Run with the live backend running on http://127.0.0.1:8069. We exercise every layer added by the "Professional Interactive Course Plans" change: 1. Login as the entity-A admin. 2. Generate a fresh course plan (LLM call; we accept the deterministic skeleton fallback when no key is configured). 3. Upload the Headway elementary workbook PDF as a RAG source and wait for the SourceIndexer to flip it to ``indexed``. 4. Run ``extract-workbooks`` — the ExerciseExtractor should produce ``interactive_workbook`` materials whose ``extracted_from`` field points back at the uploaded PDF. 5. Run ``materials/v2`` for week 1 — the v2 generator must inject RAG passages and persist a non-empty ``grounded_on`` array on every created material. 6. As an admin (entity-scope fallback in ``_resolve_attempt_target``), submit a deliberately-correct attempt against the first interactive workbook material and assert the server-side scoring persists the answers + percent. 7. Reload the attempt via ``GET .../attempts/me`` and verify it survived the round-trip. Expected provider failures: - If ``OPENAI_API_KEY`` / embedding key isn't configured the indexer flips to ``failed`` — the script reports SKIP for the RAG paths and still validates the API contracts (non-RAG generation, attempt persistence). That keeps the smoke test useful in CI without secrets. """ from __future__ import annotations import json import os import sys import time import urllib.error import urllib.parse import urllib.request import uuid API = os.environ.get("ENCOACH_API", "http://127.0.0.1:8069") LOGIN = os.environ.get("ENCOACH_LOGIN", "admin@encoach.test") PASSWORD = os.environ.get("ENCOACH_PASSWORD", "admin123") PDF_PATH = os.environ.get( "HEADWAY_PDF", "docs/toaz.info-headway-elementary-workbook-5th-edition-pr_a611cb12a73c05e5609ef05996a16ea3 (1).pdf", ) INDEXING_TIMEOUT_S = int(os.environ.get("INDEXING_TIMEOUT_S", "120")) # --------------------------------------------------------------------------- # HTTP helpers (stdlib, identical pattern to smoke_assignment_workflow.py) # --------------------------------------------------------------------------- def _req(method, path, token=None, body=None, raw=None, content_type=None): url = f"{API}{path}" data = None headers = {} if token: headers["Authorization"] = f"Bearer {token}" if body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" elif raw is not None: data = raw if content_type: headers["Content-Type"] = content_type req = urllib.request.Request(url, data=data, method=method, headers=headers) try: with urllib.request.urlopen(req, timeout=120) as r: payload = r.read().decode() or "{}" return r.status, json.loads(payload) if payload.strip() else {} except urllib.error.HTTPError as e: body_raw = e.read().decode() if e.fp else "" try: parsed = json.loads(body_raw) except Exception: parsed = {"raw": body_raw} return e.code, parsed def _multipart(path, token, fields, file_field, file_path, file_name=None, file_mime="application/pdf"): """Tiny stdlib multipart/form-data uploader for the ``/sources`` endpoint. We avoid the ``requests`` dependency on purpose — the existing smoke tests stick to ``urllib`` and we want this script to be runnable in a bare CI container. """ boundary = f"----encoachSmoke{uuid.uuid4().hex}" cr = b"\r\n" body = [] for name, value in fields.items(): body.append(f"--{boundary}".encode()) body.append( f'Content-Disposition: form-data; name="{name}"'.encode() ) body.append(b"") body.append(str(value).encode()) with open(file_path, "rb") as fh: file_bytes = fh.read() body.append(f"--{boundary}".encode()) body.append( ( f'Content-Disposition: form-data; name="{file_field}"; ' f'filename="{file_name or os.path.basename(file_path)}"' ).encode() ) body.append(f"Content-Type: {file_mime}".encode()) body.append(b"") body.append(file_bytes) body.append(f"--{boundary}--".encode()) body.append(b"") payload = cr.join(body) headers = { "Authorization": f"Bearer {token}", "Content-Type": f"multipart/form-data; boundary={boundary}", "Content-Length": str(len(payload)), } req = urllib.request.Request( f"{API}{path}", data=payload, method="POST", headers=headers, ) try: with urllib.request.urlopen(req, timeout=180) as r: raw = r.read().decode() or "{}" return r.status, json.loads(raw) if raw.strip() else {} except urllib.error.HTTPError as e: raw = e.read().decode() if e.fp else "" try: parsed = json.loads(raw) except Exception: parsed = {"raw": raw} return e.code, parsed # --------------------------------------------------------------------------- # Pretty printing # --------------------------------------------------------------------------- def hr(title): print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") def ok(msg): print(f" PASS {msg}") def warn(msg): print(f" WARN {msg}") def skip(msg): print(f" SKIP {msg}") def fail(msg): print(f" FAIL {msg}") sys.exit(1) def must(label, code, data, ok_codes=(200, 201)): if code not in ok_codes: fail(f"{label}: expected {ok_codes}, got {code} -> {data}") print(f" OK {label} ({code})") # --------------------------------------------------------------------------- # 1. Login # --------------------------------------------------------------------------- def login(): code, data = _req("POST", "/api/login", body={"login": LOGIN, "password": PASSWORD}) if code != 200: fail(f"login: {code} -> {data}") return data["access_token"] # --------------------------------------------------------------------------- # 2. Generate a fresh plan # --------------------------------------------------------------------------- def create_plan(token): title = f"RAG smoke {uuid.uuid4().hex[:6]}" body = { "title": title, "cefr_level": "A2", "total_weeks": 4, "contact_hours_per_week": 4, "skills_division": "Reading 25% / Writing 25% / Listening 25% / Speaking 25%", "learner_profile": "Adult learners preparing for everyday English use.", "notes": "Use the uploaded Headway elementary workbook as the primary book.", } code, data = _req("POST", "/api/ai/course-plan", token=token, body=body) if code not in (200, 201): fail(f"create plan: {code} -> {data}") plan = data.get("data") or {} pid = plan.get("id") if not pid: fail(f"create plan: missing id -> {data}") ok(f"created plan #{pid} ({plan.get('name')})") return plan # --------------------------------------------------------------------------- # 3. Upload + wait for indexing # --------------------------------------------------------------------------- def upload_pdf(token, plan_id, path): if not os.path.exists(path): skip(f"PDF missing at {path} — skipping upload (RAG paths will be skipped)") return None code, data = _multipart( f"/api/ai/course-plan/{plan_id}/sources", token, fields={"kind": "file", "name": os.path.basename(path), "auto_index": "1"}, file_field="file", file_path=path, ) if code not in (200, 201): fail(f"upload PDF: {code} -> {data}") src = data.get("data") or {} sid = src.get("id") if not sid: fail(f"upload PDF: missing id -> {data}") ok(f"uploaded source #{sid} (status={src.get('status')})") return src def wait_for_index(token, plan_id, source_id, timeout=INDEXING_TIMEOUT_S): deadline = time.time() + timeout last_status = None while time.time() < deadline: code, data = _req( "GET", f"/api/ai/course-plan/{plan_id}/sources", token=token, ) if code != 200: fail(f"list sources: {code} -> {data}") items = data.get("items") or [] src = next((s for s in items if s["id"] == source_id), None) if not src: fail(f"source {source_id} disappeared") if src["status"] != last_status: print(f" source #{source_id} status = {src['status']}" f" chunks={src.get('chunks_count', 0)}") last_status = src["status"] if src["status"] == "indexed" and src.get("chunks_count", 0) > 0: ok(f"indexed in pgvector ({src['chunks_count']} chunks," f" {src.get('extracted_chars', 0)} chars)") return src if src["status"] == "failed": warn(f"indexing failed: {src.get('error', '?')}") return src time.sleep(2.0) warn(f"indexing did not complete in {timeout}s; last status = {last_status}") return None # --------------------------------------------------------------------------- # 4. Extract workbooks # --------------------------------------------------------------------------- def extract_workbooks(token, plan_id): code, data = _req( "POST", f"/api/ai/course-plan/{plan_id}/extract-workbooks", token=token, body={"max_batches": 4}, ) if code != 200: fail(f"extract-workbooks: {code} -> {data}") stats = (data.get("data") or {}) print( " extractor stats:" f" materials_created={stats.get('materials_created')}" f" exercises_total={stats.get('exercises_total')}" f" batches_run={stats.get('batches_run')}" f" reason={stats.get('reason')}" ) return stats # --------------------------------------------------------------------------- # 5. Generate week 1 v2 (RAG) # --------------------------------------------------------------------------- def generate_week_v2(token, plan_id, week_number=1): code, data = _req( "POST", f"/api/ai/course-plan/{plan_id}/weeks/{week_number}/materials/v2", token=token, ) if code == 409: warn("v2 generator returned 409 (no indexed sources) — RAG path skipped") return None if code != 200: fail(f"generate v2 week {week_number}: {code} -> {data}") materials = data.get("items") or [] rag = data.get("rag") or {} print( f" week {week_number} produced {len(materials)} materials," f" RAG sources_used={rag.get('sources_used')}" ) return materials def fallback_generate_v1(token, plan_id, week_number=1): """Used only when RAG isn't available (no sources / index failed).""" code, data = _req( "POST", f"/api/ai/course-plan/{plan_id}/weeks/{week_number}/materials", token=token, ) if code != 200: fail(f"generate v1 week {week_number}: {code} -> {data}") return data.get("items") or [] # --------------------------------------------------------------------------- # 6. Find a workbook material + assign plan to ourselves so we can submit # --------------------------------------------------------------------------- def get_full_plan(token, plan_id): code, data = _req("GET", f"/api/ai/course-plan/{plan_id}", token=token) if code != 200: fail(f"get plan: {code} -> {data}") return data.get("data") or {} def pick_workbook_material(plan): """Return the first material that exposes ``exercises`` we can answer. Prefers ``interactive_workbook`` types, but the v2 generator can also embed an ``interactive_workbook`` block inside ``grammar_lesson`` / ``vocabulary_list`` bodies so we look there too. """ for m in plan.get("materials") or []: if m.get("material_type") == "interactive_workbook": body = m.get("body") or {} if (body.get("exercises") or []): return m, body["exercises"] body = m.get("body") or {} wb = body.get("interactive_workbook") or {} ex = wb.get("exercises") or [] if ex: return m, ex return None, [] def build_correct_answers(exercises): """Construct the answer key the server should grade as 100%.""" out = {} for ex in exercises: eid = ex.get("id") if not eid: continue kind = ex.get("type") if kind == "gap_fill" and ex.get("answer") is not None: out[eid] = ex["answer"] elif kind == "multiple_choice" and ex.get("answer") is not None: out[eid] = ex["answer"] elif kind == "match_pairs" and ex.get("answer"): out[eid] = ex["answer"] elif kind == "reorder_words" and ex.get("answer") is not None: out[eid] = ex["answer"] elif kind == "transformation" and ex.get("answer") is not None: out[eid] = ex["answer"] elif kind == "short_answer": ans = ex.get("answer") or ( ex.get("accepted") or [None] )[0] if ans: out[eid] = ans.replace("*", "anything") return out # --------------------------------------------------------------------------- # 7. Submit attempt + reload # --------------------------------------------------------------------------- def submit_attempt(token, plan_id, material_id, answers, finalize=False): code, data = _req( "POST", f"/api/student/course-plans/{plan_id}/materials/{material_id}/attempts", token=token, body={"answers": answers, "finalize": finalize}, ) if code != 200: fail(f"save attempt: {code} -> {data}") return data.get("data") or {} def get_my_attempt(token, plan_id, material_id): code, data = _req( "GET", f"/api/student/course-plans/{plan_id}/materials/{material_id}/attempts/me", token=token, ) if code != 200: fail(f"get my attempt: {code} -> {data}") return data.get("data") # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): hr("0. Login") token = login() ok(f"logged in as {LOGIN}") hr("1. Create a fresh plan") plan = create_plan(token) plan_id = plan["id"] hr("2. Upload Headway PDF + wait for pgvector indexing") src = upload_pdf(token, plan_id, PDF_PATH) rag_ready = False if src: indexed = wait_for_index(token, plan_id, src["id"]) rag_ready = bool( indexed and indexed.get("status") == "indexed" and indexed.get("chunks_count", 0) > 0 ) hr("3. Extract workbook exercises from indexed sources") if rag_ready: stats = extract_workbooks(token, plan_id) if (stats.get("materials_created") or 0) > 0: ok( f"extractor created {stats['materials_created']} workbook(s)" f" with {stats.get('exercises_total', 0)} exercise(s)" ) elif stats.get("reason"): warn(f"extractor returned reason='{stats['reason']}'") else: warn("extractor produced 0 workbooks (LLM may be unavailable)") else: skip("no indexed source — extract-workbooks skipped") hr("4. Generate week 1 with the RAG-grounded v2 endpoint") materials = None if rag_ready: materials = generate_week_v2(token, plan_id, 1) if materials: grounded_count = sum( 1 for m in materials if m.get("grounded_on") ) if grounded_count > 0: ok( f"{grounded_count}/{len(materials)} materials carry" f" grounded_on provenance" ) else: warn( "no material has grounded_on — pipeline may have" " fallen back to deterministic skeleton" ) if not materials: skip("falling back to v1 generator so the rest of the smoke can run") fallback_generate_v1(token, plan_id, 1) hr("5. Pick an interactive workbook + submit a correct attempt") plan_full = get_full_plan(token, plan_id) material, exercises = pick_workbook_material(plan_full) if not material or not exercises: skip( "no interactive workbook material on this plan — attempt" " persistence path skipped (extractor + v2 likely returned" " nothing because no LLM/RAG keys were configured)" ) hr("DONE — partial smoke (no LLM/RAG path exercised)") return print( f" using material #{material['id']} ({material.get('material_type')})" f" with {len(exercises)} exercise(s)" ) answers = build_correct_answers(exercises) if not answers: skip("could not build any answers from exercise schema") hr("DONE — partial smoke") return attempt = submit_attempt(token, plan_id, material["id"], answers) print( f" server scored {attempt.get('correct_count')}/" f"{attempt.get('total_count')} = {attempt.get('percent')}%" ) if (attempt.get("total_count") or 0) <= 0: fail("attempt persisted with total_count == 0 — scoring did not run") ok( f"attempt #{attempt.get('id')} persisted" f" (attempt_number={attempt.get('attempt_number')})" ) hr("6. Reload the attempt and verify the answers + score round-trip") reloaded = get_my_attempt(token, plan_id, material["id"]) if not reloaded: fail("attempt vanished after reload") if reloaded.get("id") != attempt.get("id"): fail( f"reload returned a different attempt" f" ({reloaded.get('id')} vs {attempt.get('id')})" ) if reloaded.get("answers") != answers: fail( "answers diverged between save and reload" f"\n saved: {answers}" f"\n reloaded: {reloaded.get('answers')}" ) ok("attempt + answers persisted across reload") hr("7. Finalise the attempt (locks scoring)") finalised = submit_attempt( token, plan_id, material["id"], answers, finalize=True, ) if not finalised.get("is_final"): fail("finalize=true did not flip is_final") ok(f"attempt finalised at {finalised.get('submitted_at')}") hr("DONE — Professional interactive course-plan smoke passed") print( "Tip: open the plan in /admin/course-plans/, click 'Student" " view' to see the same workbook the student sees, then exit" " to view the teacher dashboard." ) if __name__ == "__main__": main()