import json from io import BytesIO from django.core.files.uploadedfile import SimpleUploadedFile from openpyxl import Workbook from api.models import ( EsDepartment, EsExamActivity, EsExamRequirement, EsExamRequirementStudent, EsSetting, EsStudent, EsStudentStudentGroup, EsTag, EsZone, ) from api.tests.helpers import SchedulerTestCase, create_admin_client, post_signed, signed_payload GET_INFO_URL = "/api/admin/import/get-information" IMPORT_TABLE_URL = "/api/admin/import/table" IMPORT_TABLES = [ "es_zone", "es_department", "es_tag", "es_suitability", "es_pos", "es_student_group", "es_invigilator_role", "es_staff", "es_student", "es_location", "es_invigilator", "es_exam_period", "es_session_start", "es_exam_requirement", "es_suitability_invigilator", "es_suitability_location", "es_invigilator_roles", "es_student_student_group", "es_exam_period_unavailability", "es_exam_requirement_student", "es_exam_req_invigilator_s", "es_exam_req_location_s", ] def build_xlsx(sheets): workbook = Workbook() workbook.remove(workbook.active) for name, headers, rows in sheets: worksheet = workbook.create_sheet(name) worksheet.append(headers) for row in rows: worksheet.append(row) buffer = BytesIO() workbook.save(buffer) buffer.seek(0) return SimpleUploadedFile( "import.xlsx", buffer.read(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) class ImportWizardTests(SchedulerTestCase): def setUp(self): super().setUp() self.user, self.client, self.token = create_admin_client(permissions=["import;table"]) EsSetting.objects.update_or_create( param="import_tables", defaults={"value": json.dumps(IMPORT_TABLES)}, ) def post_import(self, uploaded_file): payload = signed_payload({}) payload["import_file"] = uploaded_file return self.client.post(IMPORT_TABLE_URL, payload, format="multipart") def test_get_information_lists_importable_tables(self): response = post_signed(self.client, GET_INFO_URL, {}) self.assertEqual(response.status_code, 200) tables = response.json()["data"]["data"] names = {item["table"] for item in tables} self.assertIn("es_zone", names) self.assertIn("es_exam_requirement", names) self.assertTrue(all("name" in item for item in tables)) sorted_names = [item["name"] for item in tables] self.assertEqual(sorted_names, sorted(sorted_names)) def test_get_information_returns_zone_columns(self): response = post_signed(self.client, GET_INFO_URL, {"destination": "es_zone"}) self.assertEqual(response.status_code, 200) columns = response.json()["data"]["data"] by_id = {item["id"]: item for item in columns} self.assertEqual(by_id["code"]["rules"], ["required", "string", "unique"]) self.assertEqual(by_id["name"]["rules"], ["required", "string"]) self.assertIn("zone_id", by_id) self.assertEqual(by_id["extra_data_1"]["rules"], ["nullable", "string"]) self.assertIn("extra_data_10", by_id) self.assertEqual(by_id["tag"]["rules"], ["nullable", "string"]) def test_get_information_rejects_unknown_destination(self): response = post_signed(self.client, GET_INFO_URL, {"destination": "not_a_table"}) self.assertEqual(response.status_code, 400) def test_import_zone_department_student_and_relation_errors(self): uploaded = build_xlsx([ ("es_zone", ["code", "name", "extra_data_1", "tag"], [ ["Z1", "North", "North campus note", "LAB, MISSINGTAG"], ["", "Skipped", "", ""], ["Z1", "Duplicate", "", ""], ]), ("es_department", ["code", "name", "zone_id", "extra_data_1", "tag"], [ ["D1", "Science", "Z1", "Science extra", "LAB"], ["D2", "Unknown Zone", "ZMISSING", "", ""], ]), ("es_tag", ["code", "name", "department_id"], [ ["LAB", "Lab", "D1"], ]), ("es_student_group", ["code", "name", "department_id"], [ ["G1", "Group One", "D1"], ]), ("es_student", ["code", "name", "department_id", "student_group_id"], [ ["S1", "Alice", "D1", "G1"], ]), ("es_exam_period", ["code", "name", "start_date", "end_date", "start_time", "end_time"], [ ["P1", "Main", "2026-03-02", "2026-03-06", "09:00", "18:00"], ]), ("es_exam_requirement", ["code", "exam_period_id", "name", "planned_size", "writing_time", "reading_time"], [ ["REQ1", "P1", "Chemistry", 10, "02:00", "00:15"], ]), ("es_exam_requirement_student", ["exam_requirement_id", "student_id"], [ ["REQ1", "S1"], ["REQ1", "SMISSING"], ]), ("unknown_sheet", ["code"], [["X1"]]), ]) response = self.post_import(uploaded) self.assertEqual(response.status_code, 200, response.content) body = response.json()["data"] self.assertEqual(body["es_zone"]["inserted"], 1) self.assertGreaterEqual(body["es_zone"]["skipped"], 2) zone = EsZone.objects.get(code="Z1", name="North", extra_data_1="North campus note") tag = EsTag.objects.get(code="LAB") self.assertTrue(zone.tag.filter(tag=tag).exists()) self.assertTrue(any("MISSINGTAG" in str(err.get("error", "")) for err in body["es_zone"]["errors"])) self.assertEqual(body["es_department"]["inserted"], 2) dept_ok = EsDepartment.objects.get(code="D1") self.assertEqual(dept_ok.zone.code, "Z1") self.assertEqual(dept_ok.extra_data_1, "Science extra") self.assertTrue(dept_ok.tag.filter(tag=tag).exists()) dept_overridden = EsDepartment.objects.get(code="D2") self.assertIsNone(dept_overridden.zone_id) self.assertTrue(any("ZMISSING" in str(err.get("error", "")) for err in body["es_department"]["errors"])) self.assertEqual(body["es_student"]["inserted"], 1) student = EsStudent.objects.get(code="S1") self.assertEqual(student.department_id, dept_ok.id) self.assertTrue(EsStudentStudentGroup.objects.filter(student=student, student_group__code="G1").exists()) self.assertEqual(body["es_exam_requirement"]["inserted"], 1) requirement = EsExamRequirement.objects.get(code="REQ1") self.assertTrue(EsExamActivity.objects.filter(exam_requirement=requirement, code="REQ1-01").exists()) self.assertEqual(body["es_exam_requirement_student"]["inserted"], 1) self.assertEqual(body["es_exam_requirement_student"]["skipped"], 1) self.assertTrue( EsExamRequirementStudent.objects.filter(exam_requirement=requirement, student=student).exists() ) self.assertTrue(any(err.get("field") == "student_id" for err in body["es_exam_requirement_student"]["errors"])) self.assertEqual(body["unknown_sheet"]["errors"], "Sheet name not found") clear_uploaded = build_xlsx([ ("es_zone", ["code", "name", "tag"], [ ["Z1", "North", ""], ]), ]) clear_response = self.post_import(clear_uploaded) self.assertEqual(clear_response.status_code, 200, clear_response.content) self.assertFalse(zone.tag.exists()) self.assertTrue(dept_ok.tag.filter(tag=tag).exists()) omit_uploaded = build_xlsx([ ("es_department", ["code", "name"], [ ["D1", "Science"], ]), ]) omit_response = self.post_import(omit_uploaded) self.assertEqual(omit_response.status_code, 200, omit_response.content) self.assertTrue(dept_ok.tag.filter(tag=tag).exists()) def test_import_table_requires_file(self): response = post_signed(self.client, IMPORT_TABLE_URL, {}) self.assertEqual(response.status_code, 400) self.assertIn("No file uploaded", response.json()["error"]) def test_import_table_requires_permission(self): _user, client, _token = create_admin_client(permissions=[]) uploaded = build_xlsx([("es_zone", ["code", "name"], [["Z9", "Zone Nine"]])]) payload = signed_payload({}) payload["import_file"] = uploaded response = client.post(IMPORT_TABLE_URL, payload, format="multipart") self.assertEqual(response.status_code, 403)