penracourses/generic/tests/test_exam_submit_security.py

import json
import pytest
from django.urls import reverse
from django.contrib.auth.models import User
from django.test import Client
from django.utils import timezone
from anatomy.models import Exam as AnatomyExam, AnatomyQuestion, QuestionType, Structure
from generic.models import CidUser, CidUserGroup

@pytest.fixture
def logged_in_user(db):
    user = User.objects.create_user(username="user1", password="password", email="user1@test.com")
    return user

@pytest.fixture
def logged_in_client(logged_in_user):
    client = Client()
    client.force_login(logged_in_user)
    return client

@pytest.fixture
def anonymous_client(db):
    return Client()

@pytest.fixture
def exam_data(db):
    exam = AnatomyExam.objects.create(
        name="Anatomy Exam Test",
        exam_mode=True,
        active=True,
        open_access=False,
        exam_open_access=True,  # allows open access so we don't have to populate cid/user lists for candidate test
    )
    qtype = QuestionType.objects.create(id=1, text="Test Type")
    structure = Structure.objects.create(name="Test Structure")
    question = AnatomyQuestion.objects.create(
        question_type=qtype,
        structure=structure,
        image="test.jpg",
        description="Test Question description",
    )
    question.exams.set([exam])
    return exam, question

@pytest.mark.django_db
def test_authenticated_user_submit_success(logged_in_client, logged_in_user, exam_data):
    exam, question = exam_data
    payload = {
        "eid": f"anatomy/{exam.pk}",
        "cid": f"u-{logged_in_user.pk}",
        "start_time": timezone.now().timestamp(),
        "answers": json.dumps([
            {
                "eid": f"anatomy/{exam.pk}",
                "cid": f"u-{logged_in_user.pk}",
                "qid": question.pk,
                "ans": "Normal",
                "qidn": "1",
            }
        ])
    }
    response = logged_in_client.post(reverse("global_exam_answers_submit"), payload)
    assert response.status_code == 200
    data = response.json()
    assert data["success"] is True
    assert data["question_count"] == 1

@pytest.mark.django_db
def test_authenticated_user_submit_mismatch_other_user(logged_in_client, logged_in_user, exam_data):
    exam, question = exam_data
    payload = {
        "eid": f"anatomy/{exam.pk}",
        "cid": f"u-{logged_in_user.pk + 999}", # root cid doesn't match authenticated user
        "start_time": timezone.now().timestamp(),
        "answers": json.dumps([
            {
                "eid": f"anatomy/{exam.pk}",
                "cid": f"u-{logged_in_user.pk}",
                "qid": question.pk,
                "ans": "Normal",
                "qidn": "1",
            }
        ])
    }
    response = logged_in_client.post(reverse("global_exam_answers_submit"), payload)
    assert response.status_code == 200
    data = response.json()
    assert data["success"] is False
    assert "user / cid mismatch" in data["error"]

@pytest.mark.django_db
def test_authenticated_user_submit_mismatch_answer_level(logged_in_client, logged_in_user, exam_data):
    exam, question = exam_data
    payload = {
        "eid": f"anatomy/{exam.pk}",
        "cid": f"u-{logged_in_user.pk}",
        "start_time": timezone.now().timestamp(),
        "answers": json.dumps([
            {
                "eid": f"anatomy/{exam.pk}",
                "cid": f"u-{logged_in_user.pk + 999}", # answer level cid doesn't match authenticated user
                "qid": question.pk,
                "ans": "Normal",
                "qidn": "1",
            }
        ])
    }
    response = logged_in_client.post(reverse("global_exam_answers_submit"), payload)
    assert response.status_code == 200
    data = response.json()
    assert data["success"] is False
    assert "user / uid mismatch" in data["error"]

@pytest.mark.django_db
def test_anonymous_user_submit_user_prefix_fails(anonymous_client, exam_data):
    exam, question = exam_data
    payload = {
        "eid": f"anatomy/{exam.pk}",
        "cid": "u-1", # prefix u- is forbidden for anonymous users
        "start_time": timezone.now().timestamp(),
        "answers": json.dumps([
            {
                "eid": f"anatomy/{exam.pk}",
                "cid": "u-1",
                "qid": question.pk,
                "ans": "Normal",
                "qidn": "1",
            }
        ])
    }
    response = anonymous_client.post(reverse("global_exam_answers_submit"), payload)
    assert response.status_code == 200
    data = response.json()
    assert data["success"] is False
    assert "authentication required" in data["error"]

@pytest.mark.django_db
def test_candidate_submit_success(anonymous_client, exam_data):
    exam, question = exam_data
    cid_user = CidUser.objects.create(
        cid=8888,
        name="Candidate 1",
        email="candidate1@test.com",
        passcode="candidatepass",
    )
    # Associate CID user with the exam
    exam.exam_open_access = False
    group = CidUserGroup.objects.create(name="Test Group")
    group.ciduser_set.add(cid_user)
    exam.cid_user_groups.add(group)
    exam.valid_cid_users.add(cid_user)

    payload = {
        "eid": f"anatomy/{exam.pk}",
        "cid": "8888",
        "start_time": timezone.now().timestamp(),
        "answers": json.dumps([
            {
                "eid": f"anatomy/{exam.pk}",
                "cid": "8888",
                "passcode": "candidatepass",
                "qid": question.pk,
                "ans": "Normal",
                "qidn": "1",
            }
        ])
    }
    response = anonymous_client.post(reverse("global_exam_answers_submit"), payload)
    assert response.status_code == 200
    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")
    
    # 1. First stage: Upload file -> renders preview
    response = cid_manager_client.post(url, {"answers_file": answers_file}, format="multipart")
    assert response.status_code == 200
    assert b"Preview Import" in response.content
    assert b"Imported Test Answer" in response.content
    
    # Check session has pending data
    assert cid_manager_client.session["pending_import_data"] == import_data
    
    # 2. Second stage: Confirm import
    response2 = cid_manager_client.post(url, {"confirm": "1"})
    assert response2.status_code == 200
    assert b"Import Successful" in response2.content
    assert b"1" in response2.content # created_count count
    
    # Verify answers were saved to DB
    answers = AnatomyUserAnswer.objects.filter(exam=exam, user=logged_in_user)
    assert answers.count() == 1
    assert answers.first().answer == "Imported Test Answer"


@pytest.mark.django_db
def test_answer_history_and_compacting(cid_manager_client, exam_data, logged_in_user):
    exam, question = exam_data
    from anatomy.models import UserAnswer as AnatomyUserAnswer
    import reversion
    
    # 1. Create an answer within a revision context
    with reversion.create_revision():
        ans = AnatomyUserAnswer.objects.create(
            exam=exam,
            question=question,
            user=logged_in_user,
            answer="Version 1",
        )
        reversion.set_comment("Initial answer")
        
    # 2. Modify the answer within a revision context
    with reversion.create_revision():
        ans.answer = "Version 2"
        ans.save()
        reversion.set_comment("Correction")
        
    url = reverse("generic:view_answer_history", kwargs={
        "app_name": "anatomy",
        "model_name": "UserAnswer",
        "pk": ans.pk,
    })
    
    # Check GET revision list
    response = cid_manager_client.get(url)
    assert response.status_code == 200
    assert b"Version 1" in response.content
    assert b"Version 2" in response.content
    assert b"Initial answer" in response.content
    
    # Test that compact_single is forbidden
    response_compact_single = cid_manager_client.post(url, {"action": "compact_single"})
    assert response_compact_single.status_code == 403 # forbidden/denied
    
    # Test Compact Bulk action is forbidden for non-admin/non-staff
    response_compact_bulk_fail = cid_manager_client.post(url, {"action": "compact_bulk"})
    assert response_compact_bulk_fail.status_code == 403
    
    # Elevate the user to superuser/admin to test successful bulk compaction
    from django.contrib.auth import get_user_model
    User = get_user_model()
    manager_user = User.objects.get(username="manager1")
    manager_user.is_superuser = True
    manager_user.save()
    
    # Test Compact Bulk action (exam-wide compaction) succeeds for admin
    response_compact = cid_manager_client.post(url, {"action": "compact_bulk"})
    assert response_compact.status_code == 302 # redirect
    
    # Verify only 1 version remains for our answer object
    from reversion.models import Version
    versions = Version.objects.get_for_object(ans)
    assert versions.count() == 1
    assert versions.first().field_dict["answer"] == "Version 2"