"""Extended workflow checks. Run after http_smoke.py on the isolated test school.""" import datetime import http.cookiejar import re import sys import urllib.error import urllib.parse import urllib.request import uuid BASE = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8089" TODAY = datetime.date.today().isoformat() checks = 0 def expect(condition, description): global checks if not condition: raise AssertionError(description) checks += 1 print("PASS", description, flush=True) class Client: def __init__(self, role): self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())) self.get("?page=login") self.post("login", {"email": role + "@demo.school", "password": "SchoolDemo!2026"}) expect("Welcome back" in self.last, role + " signs in") def get(self, path): try: with self.opener.open(BASE + "/" + path) as r: self.last = r.read().decode("utf-8-sig") return r.status, self.last except urllib.error.HTTPError as e: self.last = e.read().decode("utf-8-sig") return e.code, self.last def token(self): match = re.search(r'name="csrf" value="([a-f0-9]+)"', self.last) if not match: self.get("?page=account") match = re.search(r'name="csrf" value="([a-f0-9]+)"', self.last) return match.group(1) def post(self, action, data, return_to="?page=dashboard"): data = {"csrf": self.token(), "action": action, "return_to": return_to, **data} request = urllib.request.Request(BASE + "/", urllib.parse.urlencode(data).encode()) with self.opener.open(request) as r: self.last = r.read().decode("utf-8-sig") return self.last def upload_csv(self, content): boundary = "Chikoro" + uuid.uuid4().hex parts = [] for key, val in {"csrf": self.token(), "action": "student.import_preview", "return_to": "?page=students"}.items(): parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{val}\r\n') parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="csv_file"; filename="students.csv"\r\nContent-Type: text/csv\r\n\r\n{content}\r\n--{boundary}--\r\n') request = urllib.request.Request(BASE + "/", "".join(parts).encode(), {"Content-Type": "multipart/form-data; boundary=" + boundary}) with self.opener.open(request) as r: self.last = r.read().decode("utf-8-sig") return self.last def record_id(body, marker, pattern): html = next(r for r in re.findall(r']*>(.*?)', body, re.S) if marker in r) return re.search(pattern, html).group(1) admin = Client("admin") admin.get("?page=students") body = admin.upload_csv("student_number,first_name,last_name,dob,gender,class_name,admitted_on\nINVALID,Invalid,Class,,Female,No such class," + TODAY) expect("Class not configured" in body and "0 valid new rows" in body, "invalid import produces row explanations") uid = uuid.uuid4().hex[:8] csv = ("student_number,first_name,last_name,dob,gender,class_name,admitted_on\n" f"CSV-{uid},Imported,Learner,2016-01-01,Female,Grade 4 Blue,{TODAY}\n" f"MVA-{TODAY[:4]}-001,Anesu,Moyo,2016-01-12,Male,Grade 4 Blue,{TODAY}\n" f"BAD-{uid},Unknown,Class,,Male,No such class,{TODAY}\n") body = admin.upload_csv(csv) expect("1 valid new rows" in body and "1 existing students skipped" in body and "1 invalid rows excluded" in body, "import preview separates valid existing and rejected rows") token = re.search(r'name="import_token" value="([a-f0-9]+)"', body).group(1) body = admin.post("student.import_commit", {"import_token": token}, "?page=students") expect("1 students imported, 1 existing records skipped, 1 invalid rows excluded" in body, "reviewed import commits only valid students") body = admin.upload_csv(csv) expect("0 valid new rows" in body and "2 existing students skipped" in body, "import retry cannot duplicate student numbers") class_name = "Test Primary " + uid admin.get("?page=settings") body = admin.post("class.create", {"name": class_name, "section": "primary", "teacher_id": "0"}, "?page=settings") expect("Class created" in body, "administrator creates a primary class") class_match = re.search(r'' + re.escape(class_name) + r'.*?name="class_id" value="(\d+)"', body, re.S) class_id = class_match.group(1) teacher = Client("teacher") code, body = teacher.get("?page=attendance&class_id=" + class_id) expect(code == 403, "teacher denied a class not assigned to them") teacher.get("?page=account") body = teacher.post("attendance.save", {"class_id": class_id, "attendance_date": TODAY}, "?page=attendance") expect("not assigned to you" in body, "server rejects unassigned teacher mutation") admin.get("?page=students") body = admin.post("student.create", {"first_name": "Pilot", "last_name": "Learner", "student_number": "TEST-" + uid, "dob": "2015-01-01", "gender": "Not specified", "class_id": class_id, "admitted_on": TODAY}, "?page=students") expect("Student enrolled" in body, "student registration creates permanent record") student_id = record_id(body, "TEST-" + uid, r'page=student&id=(\d+)') admin.get("?page=student&id=" + student_id) body = admin.post("guardian.link", {"student_id": student_id, "guardian_id": "1", "relationship": "Guardian", "receive_emails": "1"}, "?page=student&id=" + student_id) expect("Guardian linked" in body, "existing guardian can link another child") parent = Client("parent") code, body = parent.get("?page=student&id=" + student_id) expect(code == 200 and "Pilot Learner" in body, "guardian portal includes newly linked child") admin.get("?page=student&id=" + student_id) body = admin.post("student.move", {"student_id": student_id, "class_id": "2", "status": "active", "effective_on": TODAY, "reason": "Test promotion"}, "?page=student&id=" + student_id) expect("Student movement recorded" in body and "TEST-" + uid in body and "Test promotion" in body, "promotion preserves student number and class history") bursar = Client("bursar") bursar.get("?page=fees") title = "Test levy " + uid body = bursar.post("fees.schedule", {"class_id": "1", "title": title, "amount": "3.25", "currency": "ZAR", "due_on": TODAY}, "?page=fees") expect("Fee schedule created" in body, "class fee schedule created in separate currency") schedule_id = record_id(body, title, r'data-dialog="bill-(\d+)"') body = bursar.post("fees.bill", {"schedule_id": schedule_id}, "?page=fees") expect("new invoice(s) posted" in body, "billing posts eligible invoices") body = bursar.post("fees.bill", {"schedule_id": schedule_id}, "?page=fees") expect("0 new invoice(s) posted" in body, "repeat class billing is idempotent") cashier = Client("cashier") cashier.get("?page=fees") body = cashier.post("payment.create", {"student_id": "1", "amount": "2000.00", "currency": "USD", "method": "cash", "reference": "CREDIT-" + uid, "paid_on": TODAY, "idempotency_key": uuid.uuid4().hex}, "?page=fees") receipt = re.search(r'Payment saved\. Receipt (RC-[A-Z0-9-]+)', body).group(1) payment_id = record_id(body, receipt, r'page=receipt&id=(\d+)') body = cashier.post("refund.create", {"payment_id": payment_id, "amount": "10.00", "reason": "Test return of overpaid funds"}, "?page=fees") expect("Refund requested" in body, "refund request links original receipt") bursar.get("?page=fees") refund_id = record_id(bursar.last, "Test return of overpaid funds", r'name="refund_id" value="(\d+)"') body = bursar.post("refund.approve", {"refund_id": refund_id}, "?page=fees") expect("Refund reviewed" in body, "independent approval records credit refund") body = bursar.post("payment.reverse", {"payment_id": payment_id, "reason": "Must be blocked"}, "?page=fees") expect("approved refund and cannot be reversed" in body, "refunded original receipt cannot be reversed") body = bursar.post("refund.approve", {"refund_id": refund_id}, "?page=fees") expect("already been reviewed" in body, "refund cannot be applied twice") admin.get("?page=notices") body = admin.post("notice.create", {"title": "Escaping " + uid, "body": "", "audience": "everyone", "event_on": ""}, "?page=notices") expect("<script>" in body and "