Add in StaffGradingWorkflow

openassessment/assessment/api/staff.py

@@ -3,11 +3,14 @@ Public interface for staff grading, used by students/course staff.
 """
 import logging
 from django.db import DatabaseError, IntegrityError, transaction
+from django.utils.timezone import now
 from dogapi import dog_stats_api
 
+from submissions import api as submissions_api
+
 from openassessment.assessment.models import (
     Assessment, AssessmentFeedback, AssessmentPart,
-    InvalidRubricSelection
+    InvalidRubricSelection, StaffWorkflow,
 )
 from openassessment.assessment.serializers import (
     AssessmentFeedbackSerializer, RubricSerializer,
@@ -56,6 +59,72 @@ def assessment_is_finished(submission_uuid, requirements):
     return True
 
 
+def on_start(submission_uuid):
+    """
+    Create a new staff workflow for a student item and submission.
+
+    Creates a unique staff workflow for a student item, associated with a
+    submission.
+
+    Args:
+        submission_uuid (str): The submission associated with this workflow.
+
+    Returns:
+        None
+
+    Raises:
+        StaffAssessmentInternalError: Raised when there is an internal error
+            creating the Workflow.
+
+    """
+    try:
+        submission = submissions_api.get_submission_and_student(submission_uuid)
+        workflow, __ = StaffWorkflow.objects.get_or_create(
+            course_id=submission['student_item']['course_id'],
+            item_id=submission['student_item']['item_id'],
+            submission_uuid=submission_uuid
+        )
+    except DatabaseError:
+        error_message = (
+            u"An internal error occurred while creating a new staff "
+            u"workflow for submission {}"
+            .format(submission_uuid)
+        )
+        logger.exception(error_message)
+        raise StaffAssessmentInternalError(error_message)
+
+
+def on_cancel(submission_uuid):
+    """
+    Cancel the staff workflow for submission.
+
+    Sets the cancelled_at field in staff workflow.
+
+    Args:
+        submission_uuid (str): The submission UUID associated with this workflow.
+
+    Returns:
+        None
+
+    """
+    try:
+        workflow = StaffWorkflow.objects.get(submission_uuid=submission_uuid)
+        workflow.cancelled_at = now()
+        workflow.save(update_fields=['cancelled_at'])
+    except StaffWorkflow.DoesNotExist:
+        # If we can't find a workflow, then we don't have to do anything to
+        # cancel it.
+        pass
+    except DatabaseError:
+        error_message = (
+            u"An internal error occurred while cancelling the staff"
+            u"workflow for submission {}"
+            .format(submission_uuid)
+        )
+        logger.exception(error_message)
+        raise StaffAssessmentInternalError(error_message)
+
+
 def get_score(submission_uuid, requirements):
     """
     Generate a score based on a completed assessment for the given submission.
@@ -160,6 +229,58 @@ def get_assessment_scores_by_criteria(submission_uuid):
         raise StaffAssessmentInternalError(error_message)
 
 
+def get_submission_to_assess(course_id, item_id, scorer_id):
+    """Get a submission for staff evaluation.
+
+    Retrieves a submission for assessment for the given staff member.
+
+    Args:
+        course_id (str): The course that we would like to fetch submissions from.
+        item_id (str): The student_item (problem) that we would like to retrieve submissions for.
+        scorer_id (str): The user id of the staff member scoring this submission
+
+    Returns:
+        dict: A student submission for assessment. This contains a 'student_item',
+            'attempt_number', 'submitted_at', 'created_at', and 'answer' field to be
+            used for assessment.
+
+    Raises:
+        StaffAssessmentInternalError: Raised when there is an internal error
+            retrieving staff workflow information.
+
+    Examples:
+        >>> get_submission_to_assess("a_course_id", "an_item_id", "a_scorer_id")
+        {
+            'student_item': 2,
+            'attempt_number': 1,
+            'submitted_at': datetime.datetime(2014, 1, 29, 23, 14, 52, 649284, tzinfo=<UTC>),
+            'created_at': datetime.datetime(2014, 1, 29, 17, 14, 52, 668850, tzinfo=<UTC>),
+            'answer': u'The answer is 42.'
+        }
+
+    """
+    student_submission_uuid = StaffWorkflow.get_submission_for_review(course_id, item_id, scorer_id)
+    if student_submission_uuid:
+        try:
+            submission_data = submissions_api.get_submission(student_submission_uuid)
+            return submission_data
+        except submissions_api.SubmissionNotFoundError:
+            error_message = (
+                u"Could not find a submission with the uuid {}"
+            ).format(student_submission_uuid)
+            logger.exception(error_message)
+            raise StaffAssessmentInternalError(error_message)
+    else:
+        logger.info(
+            u"No submission found for staff to assess ({}, {})"
+            .format(
+                course_id,
+                item_id,
+            )
+        )
+        return None
+
+
 def create_assessment(
     submission_uuid,
     scorer_id,
@@ -215,6 +336,11 @@ def create_assessment(
         >>> create_assessment("Tim", options_selected, criterion_feedback, feedback, rubric_dict)
     """
     try:
+        try:
+            scorer_workflow = StaffWorkflow.objects.get(submission_uuid=submission_uuid)
+        except StaffWorkflow.DoesNotExist:
+            scorer_workflow = None
+
         assessment = _complete_assessment(
             submission_uuid,
             scorer_id,
@@ -222,7 +348,8 @@ def create_assessment(
             criterion_feedback,
             overall_feedback,
             rubric_dict,
-            scored_at
+            scored_at,
+            scorer_workflow
         )
         return full_assessment_dict(assessment)
 
@@ -250,7 +377,8 @@ def _complete_assessment(
         criterion_feedback,
         overall_feedback,
         rubric_dict,
-        scored_at
+        scored_at,
+        scorer_workflow
 ):
     """
     Internal function for atomic assessment creation. Creates a staff assessment
@@ -295,4 +423,7 @@ def _complete_assessment(
     # match the rubric.
     AssessmentPart.create_from_option_names(assessment, options_selected, feedback=criterion_feedback)
 
+    # Close the active assessment
+    if scorer_workflow is not None:
+        scorer_workflow.close_active_assessment(assessment, scorer_id)
     return assessment

openassessment/assessment/migrations/0002_staffworkflow.py

+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('assessment', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='StaffWorkflow',
+            fields=[
+                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
+                ('scorer_id', models.CharField(max_length=40, db_index=True)),
+                ('course_id', models.CharField(max_length=40, db_index=True)),
+                ('item_id', models.CharField(max_length=128, db_index=True)),
+                ('submission_uuid', models.CharField(unique=True, max_length=128, db_index=True)),
+                ('created_at', models.DateTimeField(default=django.utils.timezone.now, db_index=True)),
+                ('grading_completed_at', models.DateTimeField(null=True, db_index=True)),
+                ('grading_started_at', models.DateTimeField(null=True, db_index=True)),
+                ('cancelled_at', models.DateTimeField(null=True, db_index=True)),
+                ('assessment', models.CharField(max_length=128, null=True, db_index=True)),
+            ],
+            options={
+                'ordering': ['created_at', 'id'],
+            },
+        ),
+    ]

openassessment/assessment/models/__init__.py

@@ -8,3 +8,4 @@ from .peer import *
 from .training import *
 from .student_training import *
 from .ai import *
+from .staff import *

openassessment/assessment/models/staff.py

+"""
+Models for managing staff assessments.
+"""
+from datetime import timedelta
+
+from django.db import models, DatabaseError
+from django.utils.timezone import now
+
+from openassessment.assessment.models.base import Assessment
+from openassessment.assessment.errors import StaffAssessmentInternalError
+
+
+class StaffWorkflow(models.Model):
+    """
+    Internal Model for tracking Staff Assessment Workflow
+
+    This model can be used to determine the following information required
+    throughout the Staff Assessment Workflow:
+
+    1) Get next submission that requires assessment.
+    2) Does a submission have a staff assessment?
+    3) Does this staff member already have a submission open for assessment?
+    4) Close open assessments when completed.
+
+    """
+    # Amount of time before a lease on a submission expires
+    TIME_LIMIT = timedelta(hours=8)
+
+    scorer_id = models.CharField(max_length=40, db_index=True)
+    course_id = models.CharField(max_length=40, db_index=True)
+    item_id = models.CharField(max_length=128, db_index=True)
+    submission_uuid = models.CharField(max_length=128, db_index=True, unique=True)
+    created_at = models.DateTimeField(default=now, db_index=True)
+    grading_completed_at = models.DateTimeField(null=True, db_index=True)
+    grading_started_at = models.DateTimeField(null=True, db_index=True)
+    cancelled_at = models.DateTimeField(null=True, db_index=True)
+    assessment = models.CharField(max_length=128, db_index=True, null=True)
+
+    class Meta:
+        ordering = ["created_at", "id"]
+        app_label = "assessment"
+
+    @property
+    def is_cancelled(self):
+        """
+        Check if the workflow is cancelled.
+
+        Returns:
+            True/False
+        """
+        return bool(self.cancelled_at)
+
+    @classmethod
+    def get_submission_for_review(cls, course_id, item_id, scorer_id):
+        """
+        Find a submission for staff assessment. This function will find the next
+        submission that requires assessment, excluding any submission that has been
+        completely graded, or is actively being reviewed by other staff members.
+
+        Args:
+            submission_uuid (str): The submission UUID from the student
+                requesting a submission for assessment. This is used to explicitly
+                avoid giving the student their own submission, and determines the
+                associated Peer Workflow.
+            item_id (str): The student_item that we would like to retrieve submissions for.
+            scorer_id (str): The user id of the staff member scoring this submission
+
+        Returns:
+            submission_uuid (str): The submission_uuid for the submission to review.
+
+        Raises:
+            StaffAssessmentInternalError: Raised when there is an error retrieving
+                the workflows for this request.
+
+        """
+        timeout = (now() - cls.TIME_LIMIT).strftime("%Y-%m-%d %H:%M:%S")
+        try:
+            # Search for existing submissions that the scorer has worked on.
+            staff_workflows = StaffWorkflow.objects.filter(
+                course_id=course_id,
+                item_id=item_id,
+                scorer_id=scorer_id,
+                grading_completed_at=None,
+                cancelled_at=None,
+            )
+            # If no existing submissions exist, then get any other
+            # available workflows.
+            if not staff_workflows:
+                staff_workflows = StaffWorkflow.objects.filter(
+                    models.Q(scorer_id='') | models.Q(grading_started_at__lte=timeout),
+                    course_id=course_id,
+                    item_id=item_id,
+                    grading_completed_at=None,
+                    cancelled_at=None,
+                )
+            if not staff_workflows:
+                return None
+
+            workflow = staff_workflows[0]
+            workflow.scorer_id = scorer_id
+            workflow.grading_started_at = now()
+            workflow.save()
+            return workflow.submission_uuid
+        except DatabaseError:
+            error_message = (
+                u"An internal error occurred while retrieving a submission for staff grading"
+            )
+            logger.exception(error_message)
+            raise StaffAssessmentInternalError(error_message)
+
+    def close_active_assessment(self, assessment, scorer_id):
+        """
+        Assign assessment to workflow, and mark the grading as complete.
+        """
+        self.assessment = assessment.id
+        self.scorer_id = scorer_id
+        self.grading_completed_at = now()
+        self.save()

openassessment/assessment/test/test_staff.py

 # coding=utf-8
 import copy
 import mock
+from datetime import timedelta
 
 from django.db import DatabaseError
 from django.test.utils import override_settings
+from django.utils.timezone import now
 from ddt import ddt, data, file_data, unpack
 from nose.tools import raises
 
@@ -18,7 +20,7 @@ from openassessment.test_utils import CacheResetTest
 from openassessment.assessment.api import staff as staff_api, ai as ai_api, peer as peer_api
 from openassessment.assessment.api.self import create_assessment as self_assess
 from openassessment.assessment.api.peer import create_assessment as peer_assess
-from openassessment.assessment.models import Assessment, PeerWorkflow
+from openassessment.assessment.models import Assessment, PeerWorkflow, StaffWorkflow
 from openassessment.assessment.errors import StaffAssessmentRequestError, StaffAssessmentInternalError
 from openassessment.workflow import api as workflow_api
 from submissions import api as sub_api
@@ -106,6 +108,10 @@ class TestStaffAssessment(CacheResetTest):
         # Verify that we're still waiting on a staff assessment
         self._verify_done_state(tim_sub["uuid"], self.STEP_REQUIREMENTS_WITH_STAFF, expect_done=False)
 
+        # Verify that a StaffWorkflow step has been created and is not complete
+        workflow = StaffWorkflow.objects.get(submission_uuid=tim_sub['uuid'])
+        self.assertIsNone(workflow.grading_completed_at)
+
         # Staff assess
         staff_assessment = staff_api.create_assessment(
             tim_sub["uuid"],
@@ -117,6 +123,9 @@ class TestStaffAssessment(CacheResetTest):
         # Verify assesment made, score updated, and no longer waiting
         self.assertEqual(staff_assessment["points_earned"], OPTIONS_SELECTED_DICT[key]["expected_points"])
         self._verify_done_state(tim_sub["uuid"], self.STEP_REQUIREMENTS_WITH_STAFF)
+        # Verify that a StaffWorkflow step has been marked as complete
+        workflow.refresh_from_db()
+        self.assertIsNotNone(workflow.grading_completed_at)
 
     @data(*ASSESSMENT_SCORES_DDT)
     def test_create_assessment_score_overrides(self, key):
@@ -382,6 +391,62 @@ class TestStaffAssessment(CacheResetTest):
             u"An error occurred while creating an assessment by the scorer with this ID: {}".format("Dumbledore")
         )
 
+    def test_fetch_next_submission(self):
+        bob_sub, bob = self._create_student_and_submission("bob", "bob's answer")
+        tim_sub, tim = self._create_student_and_submission("Tim", "Tim's answer")
+        submission = staff_api.get_submission_to_assess(tim['course_id'], tim['item_id'], tim['student_id'])
+        self.assertIsNotNone(submission)
+        self.assertEqual(bob_sub, submission)
+
+    def test_fetch_same_submission(self):
+        bob_sub, bob = self._create_student_and_submission("bob", "bob's answer")
+        tim_sub, tim = self._create_student_and_submission("Tim", "Tim's answer")
+        tim_to_grade = staff_api.get_submission_to_assess(tim['course_id'], tim['item_id'], tim['student_id'])
+        self.assertEqual(bob_sub, tim_to_grade)
+        # Ensure that Bob doesn't pick up the submission that Tim is grading.
+        bob_to_grade = staff_api.get_submission_to_assess(tim['course_id'], tim['item_id'], bob['student_id'])
+        tim_to_grade = staff_api.get_submission_to_assess(tim['course_id'], tim['item_id'], tim['student_id'])
+        self.assertEqual(bob_sub, tim_to_grade)
+        self.assertEqual(tim_sub, bob_to_grade)
+
+    def test_fetch_submission_delayed(self):
+        bob_sub, bob = self._create_student_and_submission("bob", "bob's answer")
+        # Fetch the submission for Tim to grade
+        tim_to_grade = staff_api.get_submission_to_assess(bob['course_id'], bob['item_id'], "Tim")
+        self.assertEqual(bob_sub, tim_to_grade)
+
+        bob_to_grade = staff_api.get_submission_to_assess(bob['course_id'], bob['item_id'], bob['student_id'])
+        self.assertIsNone(bob_to_grade)
+
+        # Change the grading_started_at timestamp so that the 'lock' on the
+        # problem is released.
+        workflow = StaffWorkflow.objects.get(scorer_id="Tim")
+        timestamp = (now() - (workflow.TIME_LIMIT + timedelta(hours=1))).strftime("%Y-%m-%d %H:%M:%S")
+        workflow.grading_started_at = timestamp
+        workflow.save()
+
+        bob_to_grade = staff_api.get_submission_to_assess(bob['course_id'], bob['item_id'], bob['student_id'])
+        self.assertEqual(tim_to_grade, bob_to_grade)
+
+    def test_next_submission_error(self):
+        tim_sub, tim = self._create_student_and_submission("Tim", "Tim's answer")
+        with mock.patch('openassessment.assessment.api.staff.submissions_api.get_submission') as patched_get_submission:
+            patched_get_submission.side_effect = sub_api.SubmissionNotFoundError('Failed')
+            with self.assertRaises(staff_api.StaffAssessmentInternalError):
+                submission = staff_api.get_submission_to_assess(tim['course_id'], tim['item_id'], tim['student_id'])
+
+    def test_no_available_submissions(self):
+        tim_sub, tim = self._create_student_and_submission("Tim", "Tim's answer")
+        # Use a non-existent course and non-existent item.
+        submission = staff_api.get_submission_to_assess('test_course_id', 'test_item_id', tim['student_id'])
+        self.assertIsNone(submission)
+
+    def test_cancel_staff_workflow(self):
+        tim_sub, tim = self._create_student_and_submission("Tim", "Tim's answer")
+        workflow_api.cancel_workflow(tim_sub['uuid'], "Test Cancel", "Bob", {})
+        workflow = StaffWorkflow.objects.get(submission_uuid=tim_sub['uuid'])
+        self.assertIsNotNone(workflow.cancelled_at)
+
     @staticmethod
     def _create_student_and_submission(student, answer, date=None, problem_steps=None):
         """