feat: add validations to enhance messaging services

- Add validation to re-queue pulled messages
- Add validation to mark duplicate messages as failed
- Implement integrity checks to prevent redundant processing

Author: leoaulasneo98

submissions/api.py

@@ -50,7 +50,6 @@
 TOP_SUBMISSIONS_CACHE_TIMEOUT = 300
 
 
-# pylint: disable=unused-argument
 def create_external_grader_detail(student_item_dict,
                                   answer,
                                   queue_name: str,
@@ -95,7 +94,7 @@ def create_external_grader_detail(student_item_dict,
             queue_name=queue_name,
             grader_file_name=grader_file_name,
             points_possible=points_possible,
-
+            queue_key=external_grader_additional_data.get('queue_key'),
         )
 
         files_dict = external_grader_additional_data.get('files')

submissions/tests/test_api.py

@@ -869,7 +869,7 @@ def test_get_or_create_student_item_race_condition__item_not_created(self):
                     api._get_or_create_student_item(STUDENT_ITEM)  # pylint: disable=protected-access
 
     def test_create_external_grader_detail(self):
-        external_grader_detail_data = {'queue_name': 'test_queue'}
+        external_grader_detail_data = {'queue_name': 'test_queue', 'queue_key': 'test_queue'}
         external_grader_detail = api.create_external_grader_detail(STUDENT_ITEM, ANSWER_ONE,
                                                                    **external_grader_detail_data)
 
@@ -880,11 +880,11 @@ def test_create_external_grader_detail(self):
         self.assertEqual(external_grader_detail.queue_name, 'test_queue')
 
     def test_create_multiple_external_grader_detail(self):
-        external_grader_detail_data1 = {'queue_name': 'test_queue'}
+        external_grader_detail_data1 = {'queue_name': 'test_queue', 'queue_key': 'test_queue'}
         external_grader_detail1 = api.create_external_grader_detail(STUDENT_ITEM, ANSWER_ONE,
                                                                     **external_grader_detail_data1)
 
-        external_grader_detail_data2 = {'queue_name': 'test_queue'}
+        external_grader_detail_data2 = {'queue_name': 'test_queue', 'queue_key': 'test_queue'}
         external_grader_detail2 = api.create_external_grader_detail(SECOND_STUDENT_ITEM, ANSWER_ONE,
                                                                     **external_grader_detail_data2)
 
@@ -910,10 +910,15 @@ def test_create_external_grader_detail_directly_database_error(self):
                 api.create_external_grader_detail(STUDENT_ITEM, ANSWER_ONE, queue_name="test_queue")
 
     def test_create_multiple_submission_external_grader_details(self):
-        external_grader_detail1 = api.create_external_grader_detail(STUDENT_ITEM, ANSWER_ONE, queue_name="shared_queue")
+        external_grader_detail1 = api.create_external_grader_detail(STUDENT_ITEM,
+                                                                    ANSWER_ONE,
+                                                                    queue_name="shared_queue",
+                                                                    queue_key="test_queue")
 
-        external_grader_detail2 = api.create_external_grader_detail(SECOND_STUDENT_ITEM, ANSWER_TWO,
-                                                                    queue_name="shared_queue")
+        external_grader_detail2 = api.create_external_grader_detail(SECOND_STUDENT_ITEM,
+                                                                    ANSWER_TWO,
+                                                                    queue_name="shared_queue",
+                                                                    queue_key="test_queue")
 
         submission1 = Submission.objects.get(uuid=external_grader_detail1.submission.uuid)
         self.assertEqual(external_grader_detail1.queue_name, 'shared_queue')
@@ -926,7 +931,7 @@ def test_create_multiple_submission_external_grader_details(self):
         self.assertNotEqual(external_grader_detail1.submission.uuid, external_grader_detail2.submission.uuid)
 
     def test_create_external_grader_detail_with_files(self):
-        """Test creating a queue record with file handling."""
+        """Test creating a external grader with file handling."""
 
         test_file = SimpleUploadedFile(
             "test.txt",
@@ -936,6 +941,7 @@ def test_create_external_grader_detail_with_files(self):
 
         event_data = {
             'queue_name': 'test_queue',
+            'queue_key': 'test_queue',
             'files': {'test.txt': test_file}
         }
 
@@ -947,7 +953,7 @@ def test_create_external_grader_detail_with_files(self):
         self.assertEqual(submission_file.original_filename, 'test.txt')
 
     def test_create_external_grader_detail_with_multiple_files(self):
-        """Test creating a queue record with multiple files."""
+        """Test creating a external grader with multiple files."""
         test_files = {
             'test1.txt': SimpleUploadedFile(
                 "test1.txt",
@@ -963,6 +969,7 @@ def test_create_external_grader_detail_with_multiple_files(self):
 
         external_grader_detail = {
             'queue_name': 'test_queue',
+            'queue_key': 'test_queue',
             'files': test_files
         }
         external_grader_detail = api.create_external_grader_detail(STUDENT_ITEM, ANSWER_ONE, **external_grader_detail)
@@ -972,7 +979,10 @@ def test_create_external_grader_detail_with_multiple_files(self):
         self.assertEqual(filenames, {'test1.txt', 'test2.txt'})
 
     def test_create_external_grader_detail_without_files(self):
-        """Test creating a queue record without any files still works."""
-        external_grader_instance = api.create_external_grader_detail(STUDENT_ITEM, ANSWER_ONE, queue_name="test_queue")
+        """Test creating a external grader without any files still works."""
+        external_grader_instance = api.create_external_grader_detail(STUDENT_ITEM,
+                                                                     ANSWER_ONE,
+                                                                     queue_name="test_queue",
+                                                                     queue_key="test_queue")
         self.assertEqual(external_grader_instance.queue_name, 'test_queue')
         self.assertEqual(external_grader_instance.files.count(), 0)

submissions/tests/test_models.py

@@ -324,7 +324,7 @@ def setUp(self):
         )
 
     def test_default_status(self):
-        """Test that new queue records are created with 'pending' status."""
+        """Test that new external graders are created with 'pending' status."""
         self.assertEqual(self.external_grader_detail.status, 'pending')
         self.assertEqual(self.external_grader_detail.num_failures, 0)
 
@@ -788,7 +788,7 @@ def test_unique_uids(self):
         )
 
     def test_related_name_access(self):
-        """Test accessing files through the submission queue record."""
+        """Test accessing files through the submission external grader."""
         files = self.external_grader_detail.files.all()
         self.assertEqual(files.count(), 1)
         self.assertEqual(files.first(), self.submission_file)

submissions/tests/test_viewsets.py

@@ -3,13 +3,15 @@
 """
 import json
 import uuid
+from datetime import timedelta
 from unittest.mock import patch
 
 from django.contrib.auth import get_user_model
 from django.core.files.base import ContentFile
 from django.db import DatabaseError
 from django.test import override_settings
 from django.urls import reverse
+from django.utils import timezone
 from rest_framework import status
 from rest_framework.permissions import AllowAny, IsAuthenticated
 from rest_framework.test import APITestCase
@@ -35,7 +37,7 @@ def setUp(self):
             password='testpass'
         )
         self.submission = SubmissionFactory()
-        self.queue_record = ExternalGraderDetailFactory(
+        self.external_grader_detail = ExternalGraderDetailFactory(
             submission=self.submission,
             pullkey='test_pull_key',
             status='pending',
@@ -145,7 +147,7 @@ def test_get_submission_invalid_transition(self, mock_update_status):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
         new_submission = SubmissionFactory()
-        queue_record = ExternalGraderDetailFactory(
+        external_grader_detail = ExternalGraderDetailFactory(
             submission=new_submission,
             queue_name=queue_name,
             status='pending'
@@ -160,8 +162,8 @@ def test_get_submission_invalid_transition(self, mock_update_status):
             self.viewset.compose_reply(False, "Error processing submission: Invalid transition")
         )
 
-        queue_record.refresh_from_db()
-        self.assertEqual(queue_record.status, 'pending')
+        external_grader_detail.refresh_from_db()
+        self.assertEqual(external_grader_detail.status, 'pending')
 
     def test_put_result_invalid_submission_id(self):
         """Test put_result with non-existent submission ID."""
@@ -253,9 +255,9 @@ def test_put_result_set_score_failure(self):
 
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
-        self.queue_record.refresh_from_db()
-        self.assertEqual(self.queue_record.num_failures, 1)
-        self.assertEqual(self.queue_record.status, 'pending')
+        self.external_grader_detail.refresh_from_db()
+        self.assertEqual(self.external_grader_detail.num_failures, 1)
+        self.assertEqual(self.external_grader_detail.status, 'pending')
 
     def test_put_result_set_score_fail_30_times(self):
         """
@@ -279,10 +281,10 @@ def test_put_result_set_score_fail_30_times(self):
 
             self.assertEqual(response.status_code, status.HTTP_200_OK)
 
-            self.queue_record.refresh_from_db()
-            self.assertEqual(self.queue_record.num_failures, each+1)
+            self.external_grader_detail.refresh_from_db()
+            self.assertEqual(self.external_grader_detail.num_failures, each+1)
 
-        self.assertEqual(self.queue_record.status, 'failed')
+        self.assertEqual(self.external_grader_detail.status, 'failed')
 
     def test_put_result_auto_retire(self):
         """
@@ -292,8 +294,8 @@ def test_put_result_auto_retire(self):
         response = self.client.post(self.url_status)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         initial_failures = 29
-        self.queue_record.num_failures = initial_failures
-        self.queue_record.save()
+        self.external_grader_detail.num_failures = initial_failures
+        self.external_grader_detail.save()
 
         payload = {
             'xqueue_header': json.dumps({
@@ -307,19 +309,19 @@ def test_put_result_auto_retire(self):
             with patch('submissions.api.set_score') as mock_set_score:
                 mock_set_score.side_effect = Exception('Test error')
                 _ = self.client.post(self.url, payload, format='json')
-                self.queue_record.refresh_from_db()
+                self.external_grader_detail.refresh_from_db()
 
-        self.queue_record.refresh_from_db()
+        self.external_grader_detail.refresh_from_db()
 
     @patch('submissions.views.xqueue.log')
     def test_put_result_logging(self, mock_log):
         """
         Test that appropriate logging occurs in various escenarios.
         """
-        self.submission.queue_record.status = 'pulled'
-        self.submission.queue_record.save()
-        self.submission.queue_record.refresh_from_db()
-        self.assertEqual(self.submission.queue_record.status, 'pulled')
+        self.submission.external_grader_detail.status = 'pulled'
+        self.submission.external_grader_detail.save()
+        self.submission.external_grader_detail.refresh_from_db()
+        self.assertEqual(self.submission.external_grader_detail.status, 'pulled')
 
         payload = {
             'xqueue_header': json.dumps({
@@ -344,10 +346,10 @@ def test_put_result_success(self, mock_log):
         """
         Test that appropriate logging occurs in various scenarios.
         """
-        self.submission.queue_record.status = 'pulled'
-        self.submission.queue_record.save()
-        self.submission.queue_record.refresh_from_db()
-        self.assertEqual(self.submission.queue_record.status, 'pulled')
+        self.submission.external_grader_detail.status = 'pulled'
+        self.submission.external_grader_detail.save()
+        self.submission.external_grader_detail.refresh_from_db()
+        self.assertEqual(self.submission.external_grader_detail.status, 'pulled')
 
         payload = {
             'xqueue_header': json.dumps({
@@ -498,7 +500,7 @@ def test_get_submission_with_files(self):
 
         file_content = b'Test file content'
         submission_file = SubmissionFile.objects.create(
-            submission_queue=self.queue_record,
+            submission_queue=self.external_grader_detail,
             file=ContentFile(file_content, name='test.txt'),
             original_filename='test.txt'
         )
@@ -525,7 +527,7 @@ def test_get_submission_with_multiple_files(self):
         created_files = []
         for filename, content in files_data:
             submission_file = SubmissionFile.objects.create(
-                submission_queue=self.queue_record,
+                submission_queue=self.external_grader_detail,
                 file=ContentFile(content, name=filename),
                 original_filename=filename
             )
@@ -548,7 +550,7 @@ def test_get_submission_file_urls_format(self):
 
         test_uuid = uuid.uuid4()
         _ = SubmissionFile.objects.create(
-            submission_queue=self.queue_record,
+            submission_queue=self.external_grader_detail,
             file=ContentFile(b'content', name='test.txt'),
             original_filename='test.txt',
             uid=test_uuid
@@ -561,3 +563,29 @@ def test_get_submission_file_urls_format(self):
 
         expected_url = f'/test_queue/{test_uuid}'
         self.assertEqual(xqueue_files['test.txt'], expected_url)
+
+    def test_get_submission_already_pulled_status(self):
+        """Test get_submission when the status is already 'pulled'."""
+        self.client.login(username='testuser', password='testpass')
+
+        queue_name = 'test_already_pulled'
+        submission = SubmissionFactory()
+        old_time = timezone.now() - timedelta(minutes=10)
+
+        external_grader_detail = ExternalGraderDetailFactory(
+            submission=submission,
+            queue_name=queue_name,
+            status='pulled',
+            status_time=old_time,
+            pullkey='original_key'
+        )
+
+        response = self.client.get(self.get_submission_url, {'queue_name': queue_name})
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        external_grader_detail.refresh_from_db()
+
+        self.assertEqual(external_grader_detail.pullkey, 'original_key')
+        self.assertEqual(external_grader_detail.status, 'pulled')
+        self.assertEqual(external_grader_detail.status_time, old_time)