Merge branch 'aci.main' into hantkovskyi/aci-742/badge-incompletion-s…

…ignal

credentials/apps/badges/distribution/credly/credly_badges/admin.py

@@ -33,7 +33,7 @@ def sync_organization_badge_templates(self, request, queryset):
         Sync badge templates for selected organizations.
         """
         for organization in queryset:
-            sync_badge_templates_for_organization(organization.uuid)
+            sync_badge_templates_for_organization(organization.uuid, request.site)
 
 
 class CredlyBadgeTemplateAdmin(admin.ModelAdmin):

credentials/apps/badges/distribution/credly/credly_badges/management/commands/sync_organization_badge_templates.py

@@ -2,6 +2,7 @@
 
 from credly_badges.models import CredlyOrganization
 from credly_badges.utils import sync_badge_templates_for_organization
+from django.contrib.sites.models import Site
 from django.core.management.base import BaseCommand
 
 
@@ -12,27 +13,34 @@ class Command(BaseCommand):
     help = "Sync badge templates for a specific organization or all organizations"
 
     def add_arguments(self, parser):
+        parser.add_argument("--site_id", type=int, help="Site ID.")
         parser.add_argument("--organization_id", type=str, help="UUID of the organization.")
 
     def handle(self, *args, **options):
         """
         Sync badge templates for a specific organization or all organizations.
 
         Usage:
-            ./manage.py sync_organization_badge_templates
-            ./manage.py sync_organization_badge_templates --organization_id c117c179-81b1-4f7e-a3a1-e6ae30568c13
+            ./manage.py sync_organization_badge_templates --site_id 1
+            ./manage.py sync_organization_badge_templates --site_id 1 --organization_id c117c179-81b1-4f7e-a3a1-e6ae30568c13
         """
+
         organization_id = options.get("organization_id")
+        site_id = options.get("site_id")
+        try:
+            site = Site.objects.get(id=site_id)
+        except Site.DoesNotExist:
+            logger.info(f"Site with id {site_id} does not exists")
 
         if organization_id:
             logger.info(f"Syncing badge templates for single organization: {organization_id}")
-            sync_badge_templates_for_organization(organization_id)
+            sync_badge_templates_for_organization(organization_id, site)
         else:
             all_organization_ids = CredlyOrganization.get_all_organization_ids()
             logger.info(
                 f"Organization id was not provided. Syncing badge templates for all organizations: {all_organization_ids}"
             )
             for organization_id in all_organization_ids:
-                sync_badge_templates_for_organization(organization_id)
+                sync_badge_templates_for_organization(organization_id, site)
 
         logger.info("Done.")

credentials/apps/badges/distribution/credly/credly_badges/utils.py

@@ -5,7 +5,7 @@
 from .models import CredlyBadgeTemplate, CredlyOrganization
 
 
-def sync_badge_templates_for_organization(organization_id):
+def sync_badge_templates_for_organization(organization_id, site):
     """
     Pull active badge templates for a given Credly Organization.
 
@@ -27,7 +27,7 @@ def sync_badge_templates_for_organization(organization_id):
         CredlyBadgeTemplate.objects.update_or_create(
             uuid=badge_template_data.get("id"),
             defaults={
-                "site": get_current_request().site,
+                "site": site,
                 "organization": organization,
                 "name": badge_template_data.get("name"),
                 "state": badge_template_data.get("state"),

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/signals.py

@@ -7,4 +7,9 @@
 - BADGE_REQUIREMENTS_NOT_COMPLETE - a reason for earned badge revocation;
 """
 
-BADGE_PROGRESS_INCOMPLETE = Signal()
+# Signal that indicates that user didn't finish all badge template requirements.
+# providing_args=[
+#         'username',  # String usernam of User
+#         'badge_template_id',  # Integer ID of finished badge template
+#     ]
+BADGE_PROGRESS_INCOMPLETE = 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()
+        )

credentials/settings/base.py

@@ -553,24 +553,24 @@
 # .. setting_description: Dictionary of event_types mapped to dictionaries of topic to topic-related configuration.
 EVENT_BUS_PRODUCER_CONFIG = {
     # .. setting_name: EVENT_BUS_PRODUCER_CONFIG['org.openedx.learning.badge.awarded.v1']
-    #    ['learning-badge-lifecycle']['enabled']
+    #    ['learning-badges-lifecycle']['enabled']
     # .. toggle_implementation: SettingToggle
     # .. toggle_default: True
     # .. toggle_description: Enables sending org.openedx.learning.badge.awarded.v1 events over the event bus.
     # .. toggle_warning: The default may be changed in a later release.
     # .. toggle_use_cases: opt_in
     "org.openedx.learning.badge.awarded.v1": {
-        "learning-badge-lifecycle": {"event_key_field": "badge.uuid", "enabled": True},
+        "learning-badges-lifecycle": {"event_key_field": "badge.uuid", "enabled": True},
     },
     # .. setting_name: EVENT_BUS_PRODUCER_CONFIG['org.openedx.learning.badge.revoked.v1']
-    #    ['learning-badge-lifecycle']['enabled']
+    #    ['learning-badges-lifecycle']['enabled']
     # .. toggle_implementation: SettingToggle
     # .. toggle_default: True
     # .. toggle_description: Enables sending org.openedx.learning.badge.revoked.v1 events over the event bus.
     # .. toggle_warning: The default may be changed in a later release.
     # .. toggle_use_cases: opt_in
     "org.openedx.learning.badge.revoked.v1": {
-        "learning-badge-lifecycle": {"event_key_field": "badge.uuid", "enabled": True},
+        "learning-badges-lifecycle": {"event_key_field": "badge.uuid", "enabled": True},
     },
 }
 

credentials/tests/test_utils.py

@@ -0,0 +1,44 @@
+import unittest
+
+from credentials.utils import keypath
+
+class TestKeypath(unittest.TestCase):
+    def test_keypath_exists(self):
+        payload = {
+            "course": {
+                "key": "105-3332",
+            }
+        }
+        result = keypath(payload, "course.key")
+        self.assertEqual(result, "105-3332")
+
+    def test_keypath_not_exists(self):
+        payload = {
+            "course": {
+                "id": "105-3332",
+            }
+        }
+        result = keypath(payload, "course.key")
+        self.assertIsNone(result)
+
+    def test_keypath_deep(self):
+        payload = {
+            "course": {
+                "data": {
+                    "identification": {
+                        "id": 25
+                    }
+                }
+            }
+        }
+        result = keypath(payload, "course.data.identification.id")
+        self.assertEqual(result, 25)
+
+    def test_keypath_invalid_path(self):
+        payload = {
+            "course": {
+                "key": "105-3332",
+            }
+        }
+        result = keypath(payload, "course.id")
+        self.assertIsNone(result)

credentials/utils.py

@@ -0,0 +1,14 @@
+def keypath(payload, keys_path):
+    keys = keys_path.split('.')
+    current = payload
+
+    def traverse(current, keys):
+        if not keys:
+            return current
+        key = keys[0]
+        if isinstance(current, dict) and key in current:
+            return traverse(current[key], keys[1:])
+        else:
+            return None
+
+    return traverse(current, keys)