PHX-38 added the api level functionality

edx_proctoring/api.py

@@ -6,6 +6,14 @@
 In-Proc API (aka Library) for the edx_proctoring subsystem. This is not to be confused with a HTTP REST
 API which is in the views.py file, per edX coding standards
 """
+import pytz
+from datetime import datetime
+from edx_proctoring.exceptions import (
+    ProctoredExamAlreadyExist, ProctoredExamNotFoundException, StudentExamAttemptAlreadyExistException
+)
+from edx_proctoring.models import (
+    ProctoredExam, ProctoredExamStudentAllowance, ProctoredExamStudentAttempt
+)
 
 
 def create_exam(course_id, content_id, exam_name, time_limit_mins,
@@ -16,6 +24,19 @@ def create_exam(course_id, content_id, exam_name, time_limit_mins,
 
     Returns: id (PK)
     """
+    if ProctoredExam.get_exam_by_content_id(course_id, content_id) is not None:
+        raise ProctoredExamAlreadyExist
+
+    proctored_exam = ProctoredExam.objects.create(
+        course_id=course_id,
+        content_id=content_id,
+        external_id=external_id,
+        exam_name=exam_name,
+        time_limit_mins=time_limit_mins,
+        is_proctored=is_proctored,
+        is_active=is_active
+    )
+    return proctored_exam
 
 
 def update_exam(exam_id, exam_name=None, time_limit_mins=None,
@@ -26,6 +47,22 @@ def update_exam(exam_id, exam_name=None, time_limit_mins=None,
 
     Returns: id
     """
+    proctored_exam = ProctoredExam.get_exam_by_id(exam_id)
+    if proctored_exam is None:
+        raise ProctoredExamNotFoundException
+
+    if exam_name is not None:
+        proctored_exam.exam_name = exam_name
+    if time_limit_mins is not None:
+        proctored_exam.time_limit_mins = time_limit_mins
+    if is_proctored is not None:
+        proctored_exam.is_proctored = is_proctored
+    if external_id is not None:
+        proctored_exam.external_id = external_id
+    if is_active is not None:
+        proctored_exam.is_active = is_active
+    proctored_exam.save()
+    return proctored_exam
 
 
 def get_exam_by_id(exam_id):
@@ -34,6 +71,11 @@ def get_exam_by_id(exam_id):
 
     Returns dictionary version of the Django ORM object
     """
+    proctored_exam = ProctoredExam.get_exam_by_id(exam_id)
+    if proctored_exam is None:
+        raise ProctoredExamNotFoundException
+
+    return proctored_exam.__dict__
 
 
 def get_exam_by_content_id(course_id, content_id):
@@ -42,18 +84,27 @@ def get_exam_by_content_id(course_id, content_id):
 
     Returns dictionary version of the Django ORM object
     """
+    proctored_exam = ProctoredExam.get_exam_by_content_id(course_id, content_id)
+    if proctored_exam is None:
+        raise ProctoredExamNotFoundException
+
+    return proctored_exam.__dict__
 
 
 def add_allowance_for_user(exam_id, user_id, key, value):
     """
     Adds (or updates) an allowance for a user within a given exam
     """
+    ProctoredExamStudentAllowance.add_allowance_for_user(exam_id, user_id, key, value)
 
 
 def remove_allowance_for_user(exam_id, user_id, key):
     """
     Deletes an allowance for a user within a given exam.
     """
+    student_allowance = ProctoredExamStudentAllowance.get_allowance_for_user(exam_id, user_id, key)
+    if student_allowance is not None:
+        student_allowance.delete()
 
 
 def start_exam_attempt(exam_id, user_id, external_id):
@@ -63,12 +114,24 @@ def start_exam_attempt(exam_id, user_id, external_id):
 
     Returns: exam_attempt_id (PK)
     """
+    exam_attempt_obj = ProctoredExamStudentAttempt.start_exam_attempt(exam_id, user_id, external_id)
+    if exam_attempt_obj is None:
+        raise StudentExamAttemptAlreadyExistException
+    else:
+        return exam_attempt_obj
 
 
-def stop_exam_attempt(exam_id, user):
+def stop_exam_attempt(exam_id, user_id):
     """
     Marks the exam attempt as completed (sets the completed_at field and updates the record)
     """
+    exam_attempt_obj = ProctoredExamStudentAttempt.get_student_exam_attempt(exam_id, user_id)
+    if exam_attempt_obj is None:
+        raise StudentExamAttemptAlreadyExistException
+    else:
+        exam_attempt_obj.completed_at = datetime.now(pytz.UTC)
+        exam_attempt_obj.save()
+        return exam_attempt_obj
 
 
 def get_active_exams_for_user(user_id, course_id=None):

edx_proctoring/exceptions.py

+"""
+Specialized exceptions for the Notification subsystem
+"""
+
+
+class ProctoredExamAlreadyExist(Exception):
+    """
+    Generic exception when a look up fails. Since we are abstracting away the backends
+    we need to catch any native exceptions and re-throw as a generic exception
+    """
+
+
+class ProctoredExamNotFoundException(Exception):
+    """
+    Generic exception when a look up fails. Since we are abstracting away the backends
+    we need to catch any native exceptions and re-throw as a generic exception
+    """
+
+
+class StudentExamAttemptAlreadyExistException(Exception):
+    """
+    Generic exception when a look up fails. Since we are abstracting away the backends
+    we need to catch any native exceptions and re-throw as a generic exception
+    """

edx_proctoring/models.py

 """
 Data models for the proctoring subsystem
 """
+import pytz
+from datetime import datetime
 from django.db import models
 from django.db.models.signals import post_save, pre_delete
 from django.dispatch import receiver
@@ -37,6 +39,30 @@ class ProctoredExam(TimeStampedModel):
         """ Meta class for this Django model """
         unique_together = (('course_id', 'content_id'),)
 
+    @classmethod
+    def get_exam_by_content_id(cls, course_id, content_id):
+        """
+        Returns the Proctored Exam if found else returns None,
+        Given course_id and content_id
+        """
+        try:
+            proctored_exam = cls.objects.get(course_id=course_id, content_id=content_id)
+        except cls.DoesNotExist:
+            proctored_exam = None
+        return proctored_exam
+
+    @classmethod
+    def get_exam_by_id(cls, exam_id):
+        """
+        Returns the Proctored Exam if found else returns None,
+        Given exam_id (PK)
+        """
+        try:
+            proctored_exam = cls.objects.get(id=exam_id)
+        except cls.DoesNotExist:
+            proctored_exam = None
+        return proctored_exam
+
 
 class ProctoredExamStudentAttempt(TimeStampedModel):
     """
@@ -62,6 +88,34 @@ class ProctoredExamStudentAttempt(TimeStampedModel):
         """ returns boolean if this attempt is considered active """
         return self.started_at and not self.completed_at
 
+    @classmethod
+    def start_exam_attempt(cls, exam_id, user_id, external_id):
+        """
+        create and return an exam attempt entry for a given
+        exam_id. If one already exists, then returns None.
+        """
+        if cls.get_student_exam_attempt(exam_id, user_id) is None:
+            return cls.objects.create(
+                proctored_exam=exam_id,
+                user_id=user_id,
+                external_id=external_id,
+                started_at=datetime.now(pytz.UTC)
+            )
+        else:
+            return None
+
+    @classmethod
+    def get_student_exam_attempt(cls, exam_id, user_id):
+        """
+        Returns the Student Exam Attempt object if found
+        else Returns None.
+        """
+        try:
+            exam_attempt_obj = cls.objects.get(proctored_exam=exam_id, user_id=user_id)
+        except cls.DoesNotExist:
+            exam_attempt_obj = None
+        return exam_attempt_obj
+
 
 class QuerySetWithUpdateOverride(models.query.QuerySet):
     """
@@ -101,6 +155,29 @@ class ProctoredExamStudentAllowance(TimeStampedModel):
         """ Meta class for this Django model """
         unique_together = (('user_id', 'proctored_exam', 'key'),)
 
+    @classmethod
+    def get_allowance_for_user(cls, exam_id, user_id, key):
+        """
+        Returns an allowance for a user within a given exam
+        """
+        try:
+            student_allowance = cls.objects.get(proctored_exam=exam_id, user_id=user_id, key=key)
+        except cls.DoesNotExist:
+            student_allowance = None
+        return student_allowance
+
+    @classmethod
+    def add_allowance_for_user(cls, exam_id, user_id, key, value):
+        """
+        Add or (Update) an allowance for a user within a given exam
+        """
+        try:
+            student_allowance = cls.objects.get(proctored_exam=exam_id, user_id=user_id, key=key)
+            student_allowance.value = value
+            student_allowance.save()
+        except cls.DoesNotExist:
+            cls.objects.create(proctored_exam=exam_id, user_id=user_id, key=key, value=value)
+
 
 class ProctoredExamStudentAllowanceHistory(TimeStampedModel):
     """

edx_proctoring/tests/test_api.py

+"""
+All tests for the models.py
+"""
+from datetime import datetime
+import pytz
+from edx_proctoring.api import create_exam, update_exam, get_exam_by_id, get_exam_by_content_id, add_allowance_for_user, \
+    remove_allowance_for_user, start_exam_attempt, stop_exam_attempt
+from edx_proctoring.exceptions import ProctoredExamAlreadyExist, ProctoredExamNotFoundException, \
+    StudentExamAttemptAlreadyExistException
+from edx_proctoring.models import ProctoredExam, ProctoredExamStudentAllowance, ProctoredExamStudentAllowanceHistory, \
+    ProctoredExamStudentAttempt
+
+from .utils import (
+    LoggedInTestCase
+)
+
+
+class ProctoredExamApiTests(LoggedInTestCase):
+    """
+    All tests for the models.py
+    """
+
+    def setUp(self):
+        """
+        Build out test harnessing
+        """
+        super(ProctoredExamApiTests, self).setUp()
+        self.default_time_limit = 21
+        self.course_id = 'test_course'
+        self.content_id = 'test_content_id'
+        self.exam_name = 'Test Exam'
+        self.user_id = 1
+        self.key = 'Test Key'
+        self.value = 'Test Value'
+        self.external_id = 'test_external_id'
+
+    def _create_proctored_exam(self):
+        return create_exam(
+            course_id=self.course_id,
+            content_id=self.content_id,
+            exam_name=self.exam_name,
+            time_limit_mins=self.default_time_limit
+        )
+
+    def _create_student_exam_attempt_entry(self):
+        proctored_exam = self._create_proctored_exam()
+        return ProctoredExamStudentAttempt.objects.create(
+            proctored_exam=proctored_exam,
+            user_id=self.user_id,
+            external_id=self.external_id,
+            started_at=datetime.now(pytz.UTC)
+        )
+
+    def _add_allowance_for_user(self):
+        proctored_exam = self._create_proctored_exam()
+        return ProctoredExamStudentAllowance.objects.create(
+            proctored_exam=proctored_exam, user_id=self.user_id, key=self.key, value=self.value
+        )
+
+    def test_create_exam(self):
+        """
+        Test to create a proctored exam.
+        """
+        proctored_exam = self._create_proctored_exam()
+        self.assertIsNotNone(proctored_exam)
+
+    def test_create_already_existing_exam_throws_exception(self):
+        """
+        Test to create a proctored exam that has already exist in the
+        database and will throw an exception ProctoredExamAlreadyExist.
+        """
+        ProctoredExam.objects.create(
+            course_id='test_course',
+            content_id='test_content_id',
+            external_id='test_external_id',
+            exam_name='Test Exam',
+            time_limit_mins=21,
+            is_proctored=True,
+            is_active=True
+        )
+        with self.assertRaises(ProctoredExamAlreadyExist):
+            self._create_proctored_exam()
+
+    def test_update_proctored_exam(self):
+        """
+        test update the existing proctored exam
+        """
+        proctored_exam = self._create_proctored_exam()
+        update_proctored_exam = update_exam(
+            proctored_exam.id, exam_name='Updated Exam Name', time_limit_mins=30,
+            is_proctored=True, external_id='external_id', is_active=True
+        )
+
+        # only those fields were updated, whose
+        # values are passed.
+        self.assertEqual(update_proctored_exam.exam_name, 'Updated Exam Name')
+        self.assertEqual(update_proctored_exam.time_limit_mins, 30)
+        self.assertEqual(update_proctored_exam.course_id, 'test_course')
+        self.assertEqual(update_proctored_exam.content_id, 'test_content_id')
+
+    def test_update_non_existing_proctored_exam(self):
+        """
+        test to update the non-existing proctored exam
+        which will throw the exception
+        """
+        with self.assertRaises(ProctoredExamNotFoundException):
+            update_exam(1, exam_name='Updated Exam Name', time_limit_mins=30)
+
+    def test_get_proctored_exam(self):
+        """
+        test to get the exam by the exam_id and
+        then compare their values.
+        """
+        proctored_exam = self._create_proctored_exam()
+        proctored_exam = get_exam_by_id(proctored_exam.id)
+        self.assertEqual(proctored_exam['course_id'], self.course_id)
+        self.assertEqual(proctored_exam['content_id'], self.content_id)
+        self.assertEqual(proctored_exam['exam_name'], self.exam_name)
+
+        proctored_exam = get_exam_by_content_id(self.course_id, self.content_id)
+        self.assertEqual(proctored_exam['course_id'], self.course_id)
+        self.assertEqual(proctored_exam['content_id'], self.content_id)
+        self.assertEqual(proctored_exam['exam_name'], self.exam_name)
+
+    def test_get_invalid_proctored_exam(self):
+        """
+        test to get the exam by the invalid exam_id which will
+        raises exception
+        """
+
+        with self.assertRaises(ProctoredExamNotFoundException):
+            get_exam_by_id(1)
+
+        with self.assertRaises(ProctoredExamNotFoundException):
+            get_exam_by_content_id('teasd', 'tewasda')
+
+    def test_add_allowance_for_user(self):
+        proctored_exam = self._create_proctored_exam()
+        add_allowance_for_user(proctored_exam, self.user_id, self.key, self.value)
+
+        student_allowance = ProctoredExamStudentAllowance.get_allowance_for_user(
+            proctored_exam.id, self.user_id, self.key
+        )
+        self.assertIsNotNone(student_allowance)
+
+    def test_allowance_for_user_already_exists(self):
+        student_allowance = self._add_allowance_for_user()
+        add_allowance_for_user(student_allowance.proctored_exam, self.user_id, self.key, 'new_value')
+
+        student_allowance = ProctoredExamStudentAllowance.get_allowance_for_user(
+            student_allowance.proctored_exam.id, self.user_id, self.key
+        )
+        self.assertIsNotNone(student_allowance)
+        self.assertEqual(student_allowance.value, 'new_value')
+
+    def test_get_allowance_for_user_does_not_exist(self):
+        proctored_exam = self._create_proctored_exam()
+
+        student_allowance = ProctoredExamStudentAllowance.get_allowance_for_user(
+            proctored_exam.id, self.user_id, self.key
+        )
+        self.assertIsNone(student_allowance)
+
+    def test_remove_allowance_for_user(self):
+        student_allowance = self._add_allowance_for_user()
+        self.assertEqual(len(ProctoredExamStudentAllowance.objects.filter()), 1)
+        remove_allowance_for_user(student_allowance.proctored_exam.id, self.user_id, self.key)
+        self.assertEqual(len(ProctoredExamStudentAllowance.objects.filter()), 0)
+
+    def test_student_exam_attempt_entry_already_exists(self):
+        proctored_exam = self._create_proctored_exam()
+        start_exam_attempt(proctored_exam, self.user_id, self.external_id)
+        self.assertIsNotNone(start_exam_attempt)
+
+    def test_create_student_exam_attempt_entry(self):
+        proctored_exam_student_attempt = self._create_student_exam_attempt_entry()
+        with self.assertRaises(StudentExamAttemptAlreadyExistException):
+            start_exam_attempt(proctored_exam_student_attempt.proctored_exam, self.user_id, self.external_id)
+
+    def test_stop_exam_attempt(self):
+        proctored_exam_student_attempt = self._create_student_exam_attempt_entry()
+        self.assertIsNone(proctored_exam_student_attempt.completed_at)
+        proctored_exam_student_attempt = stop_exam_attempt(proctored_exam_student_attempt.proctored_exam, self.user_id)
+        self.assertIsNotNone(proctored_exam_student_attempt.completed_at)
+
+    def test_stop_invalid_exam_attempt_raises_exception(self):
+        proctored_exam = self._create_proctored_exam()
+        with self.assertRaises(StudentExamAttemptAlreadyExistException):
+            stop_exam_attempt(proctored_exam, self.user_id)