from api.models import EsInvigilatorSetting, EsSessionInvigilator, EsSetting from api.services.invigilator_setting import ( DEFAULT_INVIGILATOR_SETTING, SETTING_PARAM, compute_invigilator_counts, load_invigilator_setting, save_invigilator_setting, ) from api.services.invigilator_setting_v2 import ( DEFAULT_INVIGILATOR_SETTING as DEFAULT_INVIGILATOR_SETTING_V2, compute_rule_quantity, load_invigilator_setting as load_invigilator_setting_v2, save_invigilator_setting as save_invigilator_setting_v2, ) from api.tests.helpers import SchedulerTestCase, build_exam_world, post_signed RULES_URL = "/api/admin/invigilator-setting/rules" UPDATE_URL = "/api/admin/invigilator-setting/update" RULES_V2_URL = "/api/admin/invigilator-setting/rules-v2" UPDATE_V2_URL = "/api/admin/invigilator-setting/update-v2" def _rule(*, role_ids, is_floating=0, is_inclusive=0, mode="fixed", quantity=1, students_count=None, setting_id=None, priority=0): payload = { "role_ids": role_ids, "is_floating": is_floating, "is_inclusive": is_inclusive, "quantity_mode": EsInvigilatorSetting.QUANTITY_MODE_TO_CODE[mode], "quantity": quantity, "students_count": students_count, "priority": priority, } if setting_id is not None: payload["id"] = setting_id return payload class InvigilatorSettingTests(SchedulerTestCase): def test_compute_counts_include_and_exclude(self): setting = { "fixed_invigilators": [{"quantity": 1, "role_ids": [1]}], "students_count": 20, "invigilators_count": 1, "include_fixed_invigilators": False, } excluded = compute_invigilator_counts(45, setting) self.assertEqual(excluded["ratio_required"], 3) self.assertEqual(excluded["fixed_count"], 1) self.assertEqual(excluded["auto_count"], 3) self.assertEqual(excluded["total_required"], 4) setting["include_fixed_invigilators"] = True included = compute_invigilator_counts(45, setting) self.assertEqual(included["ratio_required"], 3) self.assertEqual(included["auto_count"], 2) self.assertEqual(included["total_required"], 3) def test_rules_returns_default_setting(self): response = post_signed(self.client, RULES_URL, {}) self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["info"]["students_count"], 0) self.assertEqual(payload["info"]["invigilators_count"], 0) self.assertFalse(payload["info"]["include_fixed_invigilators"]) self.assertEqual(payload["info"]["fixed_invigilators"], [{"quantity": 0, "role_ids": []}]) self.assertEqual(payload["info"]["total_assigned_invigilators"], {"total": 0, "roles": {}}) self.assertIn("role", payload["rules"]) def test_update_persists_and_round_trips(self): world = build_exam_world() role = world.add_invigilator_role(name="Chief Invigilator") payload = { "fixed_invigilators": [{"quantity": 1, "role_ids": [role.id]}], "students_count": 20, "invigilators_count": 1, "include_fixed_invigilators": False, } update = post_signed(self.client, UPDATE_URL, payload) self.assertEqual(update.status_code, 200) self.assertEqual(update.json()["data"]["students_count"], 20) self.assertEqual(update.json()["data"]["fixed_invigilators"][0]["role_ids"], [role.id]) self.assertNotIn("total_assigned_invigilators", update.json()["data"]) rules = post_signed(self.client, RULES_URL, {}) self.assertEqual(rules.status_code, 200) info = rules.json()["data"]["info"] self.assertEqual(info["students_count"], 20) self.assertEqual(info["invigilators_count"], 1) self.assertFalse(info["include_fixed_invigilators"]) self.assertEqual(info["fixed_invigilators"], [{"quantity": 1, "role_ids": [role.id]}]) self.assertTrue(any(item["id"] == role.id for item in rules.json()["data"]["rules"]["role"])) def test_rules_returns_current_assigned_invigilators_by_role(self): world = build_exam_world() session = world.add_session() chief = world.add_invigilator_role(name="Chief") assistant = world.add_invigilator_role(name="Assistant") first = world.add_invigilator() second = world.add_invigilator() third = world.add_invigilator() world.attach_invigilator_role(first, chief) world.attach_invigilator_role(second, assistant) world.attach_invigilator_role(third, assistant) EsSessionInvigilator.objects.create(session=session, invigilator=first) EsSessionInvigilator.objects.create(session=session, invigilator=second) EsSessionInvigilator.objects.create(session=session, invigilator=third) response = post_signed(self.client, RULES_URL, {}) self.assertEqual(response.status_code, 200) self.assertEqual( response.json()["data"]["info"]["total_assigned_invigilators"], { "total": 3, "roles": { str(chief.id): 1, str(assistant.id): 2, }, }, ) def test_invalid_role_id_returns_400(self): response = post_signed( self.client, UPDATE_URL, { "fixed_invigilators": [{"quantity": 1, "role_ids": [999999999]}], "students_count": 20, "invigilators_count": 1, "include_fixed_invigilators": False, }, ) self.assertEqual(response.status_code, 400) self.assertEqual(load_invigilator_setting()["students_count"], DEFAULT_INVIGILATOR_SETTING["students_count"]) def test_negative_quantity_returns_400(self): response = post_signed( self.client, UPDATE_URL, { "fixed_invigilators": [{"quantity": -1, "role_ids": []}], "students_count": 20, "invigilators_count": 1, "include_fixed_invigilators": False, }, ) self.assertEqual(response.status_code, 400) def test_quantity_requires_roles(self): response = post_signed( self.client, UPDATE_URL, { "fixed_invigilators": [{"quantity": 2, "role_ids": []}], "students_count": 20, "invigilators_count": 1, "include_fixed_invigilators": True, }, ) self.assertEqual(response.status_code, 400) def test_load_falls_back_when_setting_missing(self): EsSetting.objects.filter(param=SETTING_PARAM).delete() self.assertEqual(load_invigilator_setting()["fixed_invigilators"], DEFAULT_INVIGILATOR_SETTING["fixed_invigilators"]) saved = save_invigilator_setting({ "fixed_invigilators": [], "students_count": 10, "invigilators_count": 1, "include_fixed_invigilators": True, }) self.assertEqual(saved["fixed_invigilators"], []) self.assertEqual(saved["students_count"], 10) class InvigilatorSettingV2Tests(SchedulerTestCase): def test_compute_rule_quantity_example_math(self): session_senior = _rule(role_ids=[2], mode="ratio", quantity=1, students_count=30) session_assistant = _rule(role_ids=[3], mode="ratio", quantity=2, students_count=30) floating_chief_fixed = _rule(role_ids=[1], is_floating=1, mode="fixed", quantity=1) floating_chief_ratio = _rule(role_ids=[1], is_floating=1, mode="ratio", quantity=1, students_count=200) self.assertEqual(compute_rule_quantity(session_senior, 30), 1) self.assertEqual(compute_rule_quantity(session_assistant, 30), 2) self.assertEqual(compute_rule_quantity(floating_chief_fixed, 150), 1) self.assertEqual(compute_rule_quantity(floating_chief_ratio, 1000), 5) self.assertEqual(compute_rule_quantity(session_senior, 100), 4) self.assertEqual(compute_rule_quantity(session_assistant, 100), 7) def test_rules_returns_default_setting(self): response = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["info"]["configuration"], []) self.assertEqual(payload["rules"]["allow_remove"], ["configuration"]) self.assertEqual(payload["info"]["total_assigned_invigilators"], {"total": 0, "roles": {}}) self.assertEqual(payload["info"]["total_estimated_invigilators"], {"total": 0, "roles": {}, "total_sessions": 0}) self.assertIn("role", payload["rules"]) self.assertEqual(payload["rules"]["role"][0], {"id": EsInvigilatorSetting.ANY_ROLE_ID, "name": "Any"}) self.assertEqual( payload["rules"]["is_floating"], [ {"id": EsInvigilatorSetting.IS_FLOATING_TO_CODE["false"], "name": "No"}, {"id": EsInvigilatorSetting.IS_FLOATING_TO_CODE["true"], "name": "Yes"}, ], ) self.assertEqual( payload["rules"]["is_inclusive"], [ {"id": EsInvigilatorSetting.IS_INCLUSIVE_TO_CODE["false"], "name": "No"}, {"id": EsInvigilatorSetting.IS_INCLUSIVE_TO_CODE["true"], "name": "Yes"}, ], ) self.assertEqual( payload["rules"]["quantity_mode"], [ {"id": EsInvigilatorSetting.QUANTITY_MODE_TO_CODE["fixed"], "name": "Fixed"}, {"id": EsInvigilatorSetting.QUANTITY_MODE_TO_CODE["ratio"], "name": "Ratio"}, ], ) def test_update_persists_and_round_trips(self): world = build_exam_world() role = world.add_invigilator_role(name="Chief Invigilator") payload = { "configuration": [ _rule(role_ids=[role.id], is_floating=1, mode="fixed", quantity=1), _rule(role_ids=[role.id], mode="ratio", quantity=1, students_count=30), ], } update = post_signed(self.client, UPDATE_V2_URL, payload) self.assertEqual(update.status_code, 200) saved = update.json()["data"]["configuration"] self.assertEqual(len(saved), 2) self.assertEqual(saved[0]["is_floating"], 1) self.assertEqual(saved[0]["is_inclusive"], 0) self.assertEqual(saved[0]["priority"], 0) self.assertEqual(saved[0]["quantity_mode"], EsInvigilatorSetting.QUANTITY_MODE_TO_CODE["fixed"]) self.assertEqual(saved[0]["quantity"], 1) self.assertIsNone(saved[0]["students_count"]) self.assertEqual(saved[0]["role_ids"], [role.id]) self.assertEqual(saved[1]["quantity_mode"], EsInvigilatorSetting.QUANTITY_MODE_TO_CODE["ratio"]) self.assertEqual(saved[1]["students_count"], 30) self.assertNotIn("total_assigned_invigilators", update.json()["data"]) self.assertEqual(update.json()["data"]["total_estimated_invigilators"], {"total": 0, "roles": {}, "total_sessions": 0}) rules = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(rules.status_code, 200) info = rules.json()["data"]["info"] self.assertEqual(len(info["configuration"]), 2) self.assertEqual(info["configuration"][0]["id"], saved[0]["id"]) self.assertEqual(info["configuration"][1]["students_count"], 30) self.assertEqual(info["configuration"][1]["is_floating"], 0) self.assertTrue(any(item["id"] == role.id for item in rules.json()["data"]["rules"]["role"])) def test_update_replaces_missing_ids(self): world = build_exam_world() chief = world.add_invigilator_role(name="Chief") senior = world.add_invigilator_role(name="Senior") created = post_signed( self.client, UPDATE_V2_URL, { "configuration": [ _rule(role_ids=[chief.id], is_floating=1, quantity=1), _rule(role_ids=[senior.id], mode="ratio", quantity=1, students_count=30), ], }, ) self.assertEqual(created.status_code, 200) first_id = created.json()["data"]["configuration"][0]["id"] second_id = created.json()["data"]["configuration"][1]["id"] updated = post_signed( self.client, UPDATE_V2_URL, { "configuration": [ _rule(role_ids=[chief.id], is_floating=1, quantity=2, setting_id=first_id), ], }, ) self.assertEqual(updated.status_code, 200) saved = updated.json()["data"]["configuration"] self.assertEqual(len(saved), 1) self.assertEqual(saved[0]["id"], first_id) self.assertEqual(saved[0]["quantity"], 2) self.assertFalse(EsInvigilatorSetting.objects.filter(id=second_id).exists()) def test_rules_returns_current_assigned_invigilators_by_role(self): world = build_exam_world() session = world.add_session() chief = world.add_invigilator_role(name="Chief") assistant = world.add_invigilator_role(name="Assistant") first = world.add_invigilator() second = world.add_invigilator() third = world.add_invigilator() world.attach_invigilator_role(first, chief) world.attach_invigilator_role(second, assistant) world.attach_invigilator_role(third, assistant) EsSessionInvigilator.objects.create(session=session, invigilator=first) EsSessionInvigilator.objects.create(session=session, invigilator=second) EsSessionInvigilator.objects.create(session=session, invigilator=third) response = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(response.status_code, 200) self.assertEqual( response.json()["data"]["info"]["total_assigned_invigilators"], { "total": 3, "roles": { str(chief.id): 1, str(assistant.id): 2, }, }, ) def test_rules_estimates_overlapping_floaters_once(self): world = build_exam_world() for _ in range(5): world.add_session(students_enrolled=30) chief = world.add_invigilator_role(name="Chief") senior = world.add_invigilator_role(name="Senior") assistant = world.add_invigilator_role(name="Assistant") expected = { "total": 16, "roles": { str(chief.id): 1, str(senior.id): 5, str(assistant.id): 10, }, "total_sessions": 5, } update = post_signed( self.client, UPDATE_V2_URL, { "configuration": [ _rule(role_ids=[chief.id], is_floating=1, quantity=1), _rule(role_ids=[senior.id], mode="ratio", quantity=1, students_count=30), _rule(role_ids=[assistant.id], mode="ratio", quantity=2, students_count=30), ], }, ) self.assertEqual(update.status_code, 200) self.assertEqual(update.json()["data"]["total_estimated_invigilators"], expected) response = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["info"]["total_estimated_invigilators"], expected) def test_rules_estimates_inclusive_credit(self): world = build_exam_world() world.add_session(students_enrolled=50) world.add_session(start_time=world.datetime_for(1), students_enrolled=100) senior = world.add_invigilator_role(name="Senior") extra = world.add_invigilator_role(name="Invigilator") save_invigilator_setting_v2({ "configuration": [ _rule(role_ids=[senior.id], is_inclusive=1, quantity=1), _rule(role_ids=[extra.id], mode="ratio", quantity=1, students_count=50), ], }) response = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(response.status_code, 200) self.assertEqual( response.json()["data"]["info"]["total_estimated_invigilators"], { "total": 3, "roles": { str(senior.id): 2, str(extra.id): 1, }, "total_sessions": 2, }, ) def test_rules_estimates_empty_role_ids_as_any(self): world = build_exam_world() world.add_session(students_enrolled=20) save_invigilator_setting_v2({ "configuration": [_rule(role_ids=[], quantity=2)], }) response = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(response.status_code, 200) self.assertEqual( response.json()["data"]["info"]["total_estimated_invigilators"], {"total": 2, "roles": {"0": 2}, "total_sessions": 1}, ) def test_rules_estimates_floating_once_per_cohort(self): world = build_exam_world() world.add_session(students_enrolled=20) world.add_session(start_time=world.datetime_for(1), students_enrolled=20) chief = world.add_invigilator_role(name="Chief") save_invigilator_setting_v2({ "configuration": [_rule(role_ids=[chief.id], is_floating=1, quantity=1)], }) response = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(response.status_code, 200) self.assertEqual( response.json()["data"]["info"]["total_estimated_invigilators"], { "total": 2, "roles": { str(chief.id): 2, }, "total_sessions": 2, }, ) def test_invalid_role_id_returns_400(self): response = post_signed( self.client, UPDATE_V2_URL, {"configuration": [_rule(role_ids=[999999999], quantity=1)]}, ) self.assertEqual(response.status_code, 400) self.assertEqual(load_invigilator_setting_v2(), DEFAULT_INVIGILATOR_SETTING_V2) def test_invalid_is_floating_returns_400(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, { "configuration": [{ "role_ids": [role.id], "is_floating": 9, "quantity_mode": 1, "quantity": 1, "students_count": None, }], }, ) self.assertEqual(response.status_code, 400) def test_is_floating_defaults_to_false(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, { "configuration": [{ "role_ids": [role.id], "quantity_mode": 1, "quantity": 1, }], }, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["configuration"][0]["is_floating"], 0) def test_invalid_is_inclusive_returns_400(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, { "configuration": [{ "role_ids": [role.id], "is_inclusive": 9, "quantity_mode": 1, "quantity": 1, "students_count": None, }], }, ) self.assertEqual(response.status_code, 400) def test_is_inclusive_defaults_to_false(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, { "configuration": [{ "role_ids": [role.id], "quantity_mode": 1, "quantity": 1, }], }, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["configuration"][0]["is_inclusive"], 0) def test_update_persists_is_inclusive(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, {"configuration": [_rule(role_ids=[role.id], is_inclusive=1, quantity=1)]}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["configuration"][0]["is_inclusive"], 1) rules = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(rules.status_code, 200) self.assertEqual(rules.json()["data"]["info"]["configuration"][0]["is_inclusive"], 1) def test_priority_defaults_to_zero(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, { "configuration": [{ "role_ids": [role.id], "quantity_mode": 1, "quantity": 1, }], }, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["configuration"][0]["priority"], 0) def test_update_persists_priority(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, {"configuration": [_rule(role_ids=[role.id], quantity=1, priority=10)]}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["data"]["configuration"][0]["priority"], 10) rules = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(rules.status_code, 200) self.assertEqual(rules.json()["data"]["info"]["configuration"][0]["priority"], 10) def test_rules_returns_higher_priority_first(self): world = build_exam_world() low = world.add_invigilator_role(name="Low") high = world.add_invigilator_role(name="High") update = post_signed( self.client, UPDATE_V2_URL, { "configuration": [ _rule(role_ids=[low.id], quantity=1, priority=0), _rule(role_ids=[high.id], quantity=1, priority=10), ], }, ) self.assertEqual(update.status_code, 200) saved = update.json()["data"]["configuration"] self.assertEqual(saved[0]["role_ids"], [high.id]) self.assertEqual(saved[0]["priority"], 10) self.assertEqual(saved[1]["role_ids"], [low.id]) rules = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(rules.status_code, 200) configuration = rules.json()["data"]["info"]["configuration"] self.assertEqual(configuration[0]["id"], saved[0]["id"]) self.assertEqual(configuration[0]["priority"], 10) self.assertEqual(configuration[1]["id"], saved[1]["id"]) def test_rules_returns_same_priority_by_id_asc(self): world = build_exam_world() first_role = world.add_invigilator_role(name="First") second_role = world.add_invigilator_role(name="Second") created = post_signed( self.client, UPDATE_V2_URL, { "configuration": [ _rule(role_ids=[first_role.id], quantity=1), _rule(role_ids=[second_role.id], quantity=1), ], }, ) self.assertEqual(created.status_code, 200) first_id = created.json()["data"]["configuration"][0]["id"] second_id = created.json()["data"]["configuration"][1]["id"] reversed_update = post_signed( self.client, UPDATE_V2_URL, { "configuration": [ _rule(role_ids=[second_role.id], quantity=1, setting_id=second_id), _rule(role_ids=[first_role.id], quantity=1, setting_id=first_id), ], }, ) self.assertEqual(reversed_update.status_code, 200) self.assertEqual( [row["id"] for row in reversed_update.json()["data"]["configuration"]], [first_id, second_id], ) rules = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(rules.status_code, 200) self.assertEqual( [row["id"] for row in rules.json()["data"]["info"]["configuration"]], [first_id, second_id], ) def test_rules_estimates_higher_priority_first(self): world = build_exam_world() world.add_session(students_enrolled=50) senior = world.add_invigilator_role(name="Senior") extra = world.add_invigilator_role(name="Invigilator") save_invigilator_setting_v2({ "configuration": [ _rule(role_ids=[senior.id], is_inclusive=1, quantity=1, priority=0), _rule(role_ids=[extra.id], mode="ratio", quantity=1, students_count=50, priority=10), ], }) response = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(response.status_code, 200) self.assertEqual( response.json()["data"]["info"]["total_estimated_invigilators"], { "total": 2, "roles": { str(senior.id): 1, str(extra.id): 1, }, "total_sessions": 1, }, ) def test_negative_quantity_returns_400(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, {"configuration": [_rule(role_ids=[role.id], quantity=-1)]}, ) self.assertEqual(response.status_code, 400) def test_empty_role_ids_means_any(self): response = post_signed( self.client, UPDATE_V2_URL, {"configuration": [_rule(role_ids=[], quantity=2)]}, ) self.assertEqual(response.status_code, 200) saved = response.json()["data"]["configuration"][0] self.assertEqual(saved["role_ids"], [EsInvigilatorSetting.ANY_ROLE_ID]) self.assertEqual(saved["quantity"], 2) rules = post_signed(self.client, RULES_V2_URL, {}) self.assertEqual(rules.status_code, 200) self.assertEqual(rules.json()["data"]["info"]["configuration"][0]["role_ids"], [EsInvigilatorSetting.ANY_ROLE_ID]) as_zero = post_signed( self.client, UPDATE_V2_URL, {"configuration": [_rule(role_ids=[EsInvigilatorSetting.ANY_ROLE_ID], quantity=2)]}, ) self.assertEqual(as_zero.status_code, 200) self.assertEqual(as_zero.json()["data"]["configuration"][0]["role_ids"], [EsInvigilatorSetting.ANY_ROLE_ID]) def test_ratio_requires_students_count(self): world = build_exam_world() role = world.add_invigilator_role() response = post_signed( self.client, UPDATE_V2_URL, {"configuration": [_rule(role_ids=[role.id], mode="ratio", quantity=1, students_count=None)]}, ) self.assertEqual(response.status_code, 400) def test_load_falls_back_when_empty(self): EsInvigilatorSetting.objects.all().delete() self.assertEqual(load_invigilator_setting_v2(), DEFAULT_INVIGILATOR_SETTING_V2) saved = save_invigilator_setting_v2({"configuration": []}) self.assertEqual(saved, DEFAULT_INVIGILATOR_SETTING_V2)