feat: [ACI-587, ACI-586] implement openedx origin badge progress completion processing

credentials/apps/badges/services/badge_templates.py

@@ -0,0 +1,9 @@
+from django.core.exceptions import ObjectDoesNotExist
+
+from ..models import BadgeTemplate
+
+def get_badge_template_by_id(badge_template_id):
+    try:
+        return BadgeTemplate.objects.get(id=badge_template_id)
+    except ObjectDoesNotExist:
+        return None

credentials/apps/badges/services/user_credentials.py

@@ -0,0 +1,19 @@
+from django.contrib.contenttypes.models import ContentType
+
+from ..models import BadgeTemplate, UserCredential
+
+
+def create_user_credential(username, badge_template):
+    if not isinstance(username, str):
+        raise ValueError("`username` must be a string")
+
+    if not isinstance(badge_template, BadgeTemplate):
+        raise TypeError("`badge_template` must be an instance of BadgeTemplate")
+
+
+    UserCredential.objects.create(
+        credential_content_type=ContentType.objects.get_for_model(
+            badge_template),
+        credential_id=badge_template.id,
+        username=username,
+    )

credentials/apps/badges/signals/handlers.py

@@ -5,8 +5,14 @@
 """
 import logging
 
+from django.dispatch import receiver
+
 from openedx_events.tooling import OpenEdxPublicSignal, load_all_signals
 
+from .signals import BADGE_PROGRESS_COMPLETE
+
+from ..services.badge_templates import get_badge_template_by_id
+from ..services.user_credentials import create_user_credential
 from ..utils import get_badging_event_types
 from ..processing import process
 
@@ -33,4 +39,18 @@ def event_handler(sender, signal, **kwargs):
     logger.debug(f"Received signal {signal}")
 
     # NOTE (performance): all consumed messages from event bus trigger this.
-    process(signal, sender=sender, **kwargs)
+    process(signal, sender=sender, **kwargs)
+
+
+@receiver(BADGE_PROGRESS_COMPLETE)
+def listen_for_completed_badge(sender, username, badge_template_id, **kwargs):  # pylint: disable=unused-argument
+    badge_template = get_badge_template_by_id(badge_template_id)
+
+    if badge_template is None:
+        return
+
+    if badge_template.origin == 'openedx':
+        create_user_credential(username, badge_template)
+        return
+
+    # TODO: add processing for 'credly' origin

credentials/apps/badges/signals/signals.py

@@ -4,3 +4,12 @@
 - BADGE_REQUIREMENTS_COMPLETE - all badge template requirements are finished;
 - BADGE_REQUIREMENTS_NOT_COMPLETE - a reason for earned badge revocation;
 """
+
+from django.dispatch import Signal
+
+# Signal that indicates that user finisher all badge template requirements.
+# providing_args=[
+#         'username',  # String usernam of User
+#         'badge_template_id',  # Integer ID of finished badge template
+#     ]
+BADGE_PROGRESS_COMPLETE = Signal()

credentials/apps/badges/tests/test_services.py

@@ -0,0 +1,59 @@
+from django.test import TestCase
+from django.core.exceptions import ObjectDoesNotExist
+from django.contrib.contenttypes.models import ContentType
+
+from ..models import BadgeTemplate, UserCredential
+from ..services.badge_templates import get_badge_template_by_id
+from ..services.user_credentials import create_user_credential
+
+
+class UserCredentialServiceTestCase(TestCase):
+    def setUp(self):
+        # Create a test badge template
+        self.badge_template = BadgeTemplate.objects.create(
+            origin='openedx', site_id=1)
+
+    def test_create_user_credential(self):
+        # Call create_user_credential with valid arguments
+        create_user_credential('test_user', self.badge_template)
+
+        # Check if user credential is created
+        self.assertTrue(
+            UserCredential.objects.filter(
+                username='test_user',
+                credential_content_type=ContentType.objects.get_for_model(
+                    self.badge_template),
+                credential_id=self.badge_template.id,
+            ).exists()
+        )
+
+    def test_create_user_credential_invalid_username(self):
+        # Call create_user_credential with non-existent username
+        with self.assertRaises(ValueError):
+            # Passing int as username
+            create_user_credential(123, self.badge_template)
+
+    def test_create_user_credential_invalid_template(self):
+        # Call create_user_credential with non-existent badge template
+        with self.assertRaises(TypeError):
+            # Passing None as badge template
+            create_user_credential('test_user', None)
+
+
+class BadgeTemplateServiceTestCase(TestCase):
+    def setUp(self):
+        self.badge_template = BadgeTemplate.objects.create(origin='openedx', site_id=1)
+
+    def test_get_badge_template_by_id(self):
+        # Call get_badge_template_by_id with existing badge template ID
+        badge_template = get_badge_template_by_id(self.badge_template.id)
+
+        # Check if the returned badge template is correct
+        self.assertEqual(badge_template, self.badge_template)
+
+    def test_get_badge_template_by_id_nonexistent(self):
+        # Call get_badge_template_by_id with non-existent ID
+        badge_template = get_badge_template_by_id(999)  # Non-existent ID
+
+        # Check that None is returned
+        self.assertIsNone(badge_template)

credentials/apps/badges/tests/test_signals.py

@@ -0,0 +1,48 @@
+from django.test import TestCase
+from django.core.exceptions import ObjectDoesNotExist
+from django.contrib.contenttypes.models import ContentType
+
+from ..models import BadgeTemplate, UserCredential
+from ..signals.signals import BADGE_PROGRESS_COMPLETE
+
+
+class BadgeSignalReceiverTestCase(TestCase):
+    def setUp(self):
+        # Create a test badge template
+        self.badge_template = BadgeTemplate.objects.create(
+            name='test', site_id=1)
+
+    def test_signal_emission_and_receiver_execution(self):
+        # Emit the signal
+        BADGE_PROGRESS_COMPLETE.send(
+            sender=self,
+            username='test_user',
+            badge_template_id=self.badge_template.id,
+        )
+
+        # UserCredential object
+        user_credential = UserCredential.objects.filter(
+            username='test_user',
+            credential_content_type=ContentType.objects.get_for_model(
+                self.badge_template),
+            credential_id=self.badge_template.id,
+        )
+
+        # Check if user credential is created
+        self.assertTrue(user_credential.exists())
+
+        # Check if user credential status is 'awarded'
+        self.assertTrue(user_credential[0].status == 'awarded')
+
+    def test_behavior_for_nonexistent_badge_templates(self):
+        # Emit the signal with a non-existent badge template ID
+        BADGE_PROGRESS_COMPLETE.send(
+            sender=self,
+            username='test_user',
+            badge_template_id=999,  # Non-existent ID
+        )
+
+        # Check that no user credential is created
+        self.assertFalse(
+            UserCredential.objects.filter(username='test_user').exists()
+        )