diff --git a/generic/templates/generic/cid_view.html b/generic/templates/generic/cid_view.html index a7d753e7..3e8cc623 100644 --- a/generic/templates/generic/cid_view.html +++ b/generic/templates/generic/cid_view.html @@ -8,6 +8,13 @@ {% block content %} +
+

Manage Candidates

+ + Import Exam Answers + +
+
{% render_table table %} diff --git a/generic/templates/generic/import_exam_answers.html b/generic/templates/generic/import_exam_answers.html new file mode 100644 index 00000000..bfc845d0 --- /dev/null +++ b/generic/templates/generic/import_exam_answers.html @@ -0,0 +1,32 @@ +{% extends 'generic/base.html' %} +{% load crispy_forms_tags %} + +{% block content %} +
+
+
+
+
+

Import Exam Answers

+
+
+

+ Upload a JSON file containing exam answers manually exported from RTS. +

+
+ {% csrf_token %} +
+ + +
+
+ Cancel + +
+
+
+
+
+
+
+{% endblock %} diff --git a/generic/tests/test_exam_submit_security.py b/generic/tests/test_exam_submit_security.py index 1c3b909d..38d01494 100644 --- a/generic/tests/test_exam_submit_security.py +++ b/generic/tests/test_exam_submit_security.py @@ -170,3 +170,65 @@ def test_candidate_submit_success(anonymous_client, exam_data): data = response.json() assert data["success"] is True assert data["question_count"] == 1 + + +@pytest.fixture +def cid_manager_client(db): + from django.contrib.auth.models import Group + user = User.objects.create_user(username="manager1", password="password", email="manager1@test.com") + g, _ = Group.objects.get_or_create(name="cid_user_manager") + user.groups.add(g) + client = Client() + client.force_login(user) + return client + + +@pytest.mark.django_db +def test_import_exam_answers_unauthorized(anonymous_client): + # Anonymous or non-manager users should be redirected or blocked + url = reverse("generic:import_exam_answers") + response = anonymous_client.get(url) + assert response.status_code == 403 + + +@pytest.mark.django_db +def test_import_exam_answers_get(cid_manager_client): + url = reverse("generic:import_exam_answers") + response = cid_manager_client.get(url) + assert response.status_code == 200 + assert b"Import Exam Answers" in response.content + + +@pytest.mark.django_db +def test_import_exam_answers_post_success(cid_manager_client, exam_data, logged_in_user): + exam, question = exam_data + from io import BytesIO + from anatomy.models import UserAnswer as AnatomyUserAnswer + + import_data = { + "eid": f"anatomy/{exam.pk}", + "cid": f"u-{logged_in_user.pk}", + "start_time": timezone.now().timestamp(), + "answers": [ + { + "qid": question.pk, + "ans": "Imported Test Answer", + "qidn": "1", + } + ] + } + + file_content = json.dumps(import_data).encode("utf-8") + answers_file = BytesIO(file_content) + answers_file.name = "backup.json" + + url = reverse("generic:import_exam_answers") + response = cid_manager_client.post(url, {"answers_file": answers_file}, format="multipart") + + assert response.status_code == 302 # redirects to manage_cids on success + + # Verify answers were saved + answers = AnatomyUserAnswer.objects.filter(exam=exam, user=logged_in_user) + assert answers.count() == 1 + assert answers.first().answer == "Imported Test Answer" + diff --git a/generic/urls.py b/generic/urls.py index 088d36e0..3fa1e2c7 100755 --- a/generic/urls.py +++ b/generic/urls.py @@ -84,6 +84,7 @@ urlpatterns = [ name="candidate_email_details", ), path("cids/manage/", views.CidUserView.as_view(), name="manage_cids"), + path("cids/manage/import-answers/", views.import_exam_answers, name="import_exam_answers"), path("cids/manage/exams", views.CidUserExamView.as_view(), name="manage_cid_exams"), path("cids/group/", views.cid_group_view, name="cid_group_view"), path("cids/group/all", views.cid_group_view_all, name="cid_group_view_all"), diff --git a/generic/views.py b/generic/views.py index 1a680481..25cdc6b5 100644 --- a/generic/views.py +++ b/generic/views.py @@ -4900,6 +4900,216 @@ def cid_group_view(request): {"groups": groups}, ) + +@user_is_cid_user_manager +def import_exam_answers(request): + """ + Imports manual JSON exam answers files exported from RTS when direct API submission failed. + """ + from django.contrib import messages + if request.method == "POST": + json_file = request.FILES.get("answers_file") + if not json_file: + messages.error(request, "No file uploaded") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + try: + data = json.load(json_file) + except Exception as e: + messages.error(request, f"Failed to parse JSON: {e}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + eid_str = data.get("eid") + cid_str = data.get("cid") + start_time_val = data.get("start_time") + answers_data = data.get("answers") + + if not eid_str or not cid_str or answers_data is None: + messages.error(request, "Uploaded JSON is missing required fields (eid, cid, answers)") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + try: + exam_type, exam_id_str = eid_str.split("/") + eid = int(exam_id_str) + except Exception: + messages.error(request, f"Invalid eid format: {eid_str}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + EXAM_IMPORT_MAP = { + "physics": (PhysicsExam, PhysicsUserAnswer, "physics"), + "anatomy": (AnatomyExam, AnatomyUserAnswer, "anatomy"), + "rapid": (RapidsExam, RapidsUserAnswer, "rapids"), + "rapids": (RapidsExam, RapidsUserAnswer, "rapids"), + "short": (ShortsExam, ShortsUserAnswer, "shorts"), + "shorts": (ShortsExam, ShortsUserAnswer, "shorts"), + "long": (LongsExam, LongsUserAnswer, "longs"), + "longs": (LongsExam, LongsUserAnswer, "longs"), + "sba": (SbasExam, SbasUserAnswer, "sbas"), + "sbas": (SbasExam, SbasUserAnswer, "sbas"), + } + + mapping = EXAM_IMPORT_MAP.get(exam_type.lower()) + if not mapping: + messages.error(request, f"Unsupported exam type: {exam_type}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + ExamModel, UserAnswerModel, app_name = mapping + + # Check CID/UID type + uid = False + cid = False + if str(cid_str).startswith("u-"): + try: + uid = int(cid_str[2:]) + except ValueError: + messages.error(request, f"Invalid user CID format: {cid_str}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + else: + try: + cid = int(cid_str) + except ValueError: + messages.error(request, f"Invalid candidate CID format: {cid_str}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + try: + exam = ExamModel.objects.get(pk=eid) + except ExamModel.DoesNotExist: + messages.error(request, f"Exam {eid} of type {exam_type} does not exist") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + # Parse answers (can be string or array) + if isinstance(answers_data, str): + try: + answers_list = json.loads(answers_data) + except Exception as e: + messages.error(request, f"Failed to parse answers string JSON: {e}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + elif isinstance(answers_data, list): + answers_list = answers_data + else: + messages.error(request, "answers field must be a list or a JSON string representation of a list") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + # Import each answer + n = 0 + for answer in answers_list: + try: + qid = int(answer["qid"]) + posted_answer = answer["ans"] + qidn = str(answer.get("qidn", "1")) + except (KeyError, ValueError, TypeError): + continue + + if uid: + existing_answers = UserAnswerModel.objects.filter( + question__id=qid, exam__id=eid, user__id=uid + ) + else: + existing_answers = UserAnswerModel.objects.filter( + question__id=qid, exam__id=eid, cid=cid + ) + + ans = None + if app_name == "rapids": + normal = (posted_answer == "Normal") + if not existing_answers: + if uid: + ans = UserAnswerModel(answer=posted_answer, normal=normal, user=User.objects.get(id=uid)) + else: + ans = UserAnswerModel(answer=posted_answer, normal=normal, cid=cid) + ans.question_id = qid + ans.exam_id = eid + else: + ans = existing_answers[0] + if qidn == "1": + ans.normal = normal + if normal: + ans.answer = "" + elif qidn == "2" and not ans.normal: + ans.answer = posted_answer + elif app_name == "shorts": + if not existing_answers: + if uid: + ans = UserAnswerModel(answer=posted_answer, user=User.objects.get(id=uid)) + else: + ans = UserAnswerModel(answer=posted_answer, cid=cid) + ans.question_id = qid + ans.exam_id = eid + ans.score = None + else: + ans = existing_answers[0] + ans.answer = posted_answer + ans.score = None + elif app_name == "anatomy": + from anatomy.models import Answer as AnatomyAnswer + if not existing_answers: + if uid: + ans = UserAnswerModel(answer=posted_answer, user=User.objects.get(id=uid)) + else: + ans = UserAnswerModel(answer=posted_answer, cid=cid) + ans.question_id = qid + ans.exam_id = eid + ans.score = AnatomyAnswer.MarkOptions.UNMARKED + else: + ans = existing_answers[0] + ans.answer = posted_answer + ans.score = AnatomyAnswer.MarkOptions.UNMARKED + elif app_name == "longs": + if not existing_answers: + if uid: + ans = UserAnswerModel(user=User.objects.get(id=uid)) + else: + ans = UserAnswerModel(cid=cid) + ans.question_id = qid + ans.exam_id = eid + else: + ans = existing_answers[0] + + if qidn == "1": + ans.answer_observations = posted_answer + elif qidn == "2": + ans.answer_interpretation = posted_answer + elif qidn == "3": + ans.answer_principle_diagnosis = posted_answer + elif qidn == "4": + ans.answer_differential_diagnosis = posted_answer + elif qidn == "5": + ans.answer_management = posted_answer + ans.score = ans.ScoreOptions.UNMARKED + + if ans: + time_answered_str = answer.get("time_answered") + if time_answered_str: + from django.utils.dateparse import parse_datetime + parsed_time = parse_datetime(time_answered_str) + if parsed_time: + if timezone.is_naive(parsed_time): + parsed_time = timezone.make_aware(parsed_time) + ans.time_answered = parsed_time + try: + ans.full_clean() + ans.save() + n += 1 + except Exception as e: + messages.warning(request, f"Failed to save answer for question {qid}: {e}") + + # Also get or create CidUserExam + try: + if uid: + t = timezone.now() + if start_time_val: + t = timezone.datetime.fromtimestamp(start_time_val, tz=timezone.utc) + exam.get_or_create_cid_user_exam( + user_user=User.objects.get(id=uid), start_time=t + ) + except Exception: + pass + + messages.success(request, f"Successfully imported {n} answers for CID/UID {cid_str} on exam {eid_str}") + return HttpResponseRedirect(reverse("generic:manage_cids")) + + return render(request, "generic/import_exam_answers.html") + @user_is_cid_user_manager def user_group_add_by_email_user(request, group_id): if request.htmx: diff --git a/todo b/todo index 12cd749b..cc2cfe17 100644 --- a/todo +++ b/todo @@ -1,8 +1,3 @@ +Import exam answers should be accesible as its own page from the admin submenu. it should have a preview mode that shows what is to be imported (and what it will overwrite / update). it should also give some feedback about the imported answers (so we know what has been imported.). -We need to update RTS have consistent logging that can be enabled and disabled via both the gui and url parameters. All current log statements need to be set with an appropriate level. - -RTS currently supports user login in addition to the cid / passcode however this is not particuarly well implemented and is not secure (if someone knows a users user id they can submit results for that user). this needs to be fixed with a more robust implementation. - -In RTS we need to tidy up the resuming of exams that have already been started. - -In RTS we need to improve the fallback that is shown if we fail to manually submit an exam via the finish button. \ No newline at end of file +Add manual instructions to rts about how to fully reset the app if the reset parameter fails. \ No newline at end of file