Merge pull request #23148 from edx/aj/MICROBA-150

MICROBA-150 Creating EnternalIds for users enrolling in MB Programs

openedx/core/djangoapps/catalog/tests/factories.py

@@ -195,6 +195,17 @@ def generate_curricula():
     return curricula
 
 
+class ProgramTypeFactory(DictFactoryBase):
+    name = factory.Faker('word')
+    logo_image = factory.LazyFunction(generate_sized_stdimage)
+
+
+class ProgramTypeAttrsFactory(DictFactoryBase):
+    uuid = factory.Faker('uuid4')
+    slug = factory.Faker('word')
+    coaching_supported = False
+
+
 class ProgramFactory(DictFactoryBase):
     authoring_organizations = factory.LazyFunction(partial(generate_instances, OrganizationFactory, count=1))
     applicable_seat_types = factory.LazyFunction(lambda: [])
@@ -219,6 +230,7 @@ class ProgramFactory(DictFactoryBase):
     subtitle = factory.Faker('sentence')
     title = factory.Faker('catch_phrase')
     type = factory.Faker('word')
+    type_attrs = ProgramTypeAttrsFactory()
     uuid = factory.Faker('uuid4')
     video = VideoFactory()
     weeks_to_complete = fake.random_int(1, 45)
@@ -235,11 +247,6 @@ class CurriculumFactory(DictFactoryBase):
     programs = factory.LazyFunction(lambda: [])
 
 
-class ProgramTypeFactory(DictFactoryBase):
-    name = factory.Faker('word')
-    logo_image = factory.LazyFunction(generate_sized_stdimage)
-
-
 class PathwayFactory(DictFactoryBase):
     id = factory.Sequence(lambda x: x)
     description = factory.Faker('sentence')

openedx/core/djangoapps/external_user_ids/__init__.py

+"""
+edX Platform support for external user IDs.
+
+This package will be used to support generating external User IDs to be shared
+with outside parties.
+"""
+default_app_config = 'openedx.core.djangoapps.external_user_ids.apps.ExternalUserIDConfig'

openedx/core/djangoapps/external_user_ids/apps.py

+"""
+External User ID Application Configuration
+"""
+
+
+from django.apps import AppConfig
+
+
+class ExternalUserIDConfig(AppConfig):
+    """
+    Default configuration for the "openedx.core.djangoapps.credit" Django application.
+    """
+    name = 'openedx.core.djangoapps.external_user_ids'
+
+    def ready(self):
+        from . import signals  # pylint: disable=unused-variable

openedx/core/djangoapps/external_user_ids/migrations/0003_auto_20200224_1836.py

+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.28 on 2020-02-24 18:36
+from __future__ import unicode_literals
+
+from django.conf import settings
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('external_user_ids', '0002_mb_coaching_20200210_1754'),
+    ]
+
+    operations = [
+        migrations.AlterUniqueTogether(
+            name='externalid',
+            unique_together=set([('user', 'external_id_type')]),
+        ),
+    ]

openedx/core/djangoapps/external_user_ids/models.py

@@ -4,12 +4,17 @@ Models for External User Ids that are sent out of Open edX
 
 import uuid as uuid_tools
 
+from logging import getLogger
+
 from django.contrib.auth.models import User
 from django.db import models
 from model_utils.models import TimeStampedModel
 from simple_history.models import HistoricalRecords
 
 
+LOGGER = getLogger(__name__)
+
+
 class ExternalIdType(TimeStampedModel):
     """
     ExternalIdType defines the type (purpose, or expected use) of an external id. A user may have one id that is sent
@@ -17,6 +22,8 @@ class ExternalIdType(TimeStampedModel):
 
     .. no_pii:
     """
+    MICROBACHELORS_COACHING = 'mb_coaching'
+
     name = models.CharField(max_length=32, blank=False, unique=True, db_index=True)
     description = models.TextField()
     history = HistoricalRecords()
@@ -36,3 +43,61 @@ class ExternalId(TimeStampedModel):
     external_id_type = models.ForeignKey(ExternalIdType, db_index=True, on_delete=models.CASCADE)
     user = models.ForeignKey(User, db_index=True, on_delete=models.CASCADE)
     history = HistoricalRecords()
+
+    class Meta(object):
+        unique_together = (('user', 'external_id_type'),)
+
+    @classmethod
+    def user_has_external_id(cls, user, type_name):
+        """
+        Checks if a user has an ExternalId of the type_name provided
+        Arguments:
+            user: User to search for
+            type_name (str): Name of the type of ExternalId
+        Returns:
+            (Bool): True if the user already has an external ID, False otherwise.
+        """
+        if not cls.objects.filter(
+            user=user,
+            external_id_type__name=type_name
+        ).exists():
+            LOGGER.info('No external id for user id {user} with type of {type}'.format(
+                user=user.id,
+                type=type_name
+            ))
+            return False
+        return True
+
+    @classmethod
+    def add_new_user_id(cls, user, type_name):
+        """
+        Creates an ExternalId for the User of the type_name provided
+        Arguments:
+            user: User to create the ID for
+            type_name (str): Name of the type of ExternalId
+        Returns:
+            (ExternalId): Returns the external id that was created or retrieved
+            (Bool): True if the External ID was created, False if it already existed
+        """
+        try:
+            type_obj = ExternalIdType.objects.get(name=type_name)
+        except ExternalIdType.DoesNotExist:
+            LOGGER.info(
+                'External ID Creation failed for user {user}, no external id type of {type}'.format(
+                    user=user.id,
+                    type=type_name
+                )
+            )
+            return None
+        external_id, created = cls.objects.get_or_create(
+            user=user,
+            external_id_type=type_obj
+        )
+        if created:
+            LOGGER.info(
+                'External ID Created for user {user}, of type {type}'.format(
+                    user=user.id,
+                    type=type_name
+                )
+            )
+        return external_id, created

openedx/core/djangoapps/external_user_ids/signals.py

+"""
+Signal Handlers for External User Ids to be created and maintainer
+"""
+
+from logging import getLogger
+
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+
+from openedx.core.djangoapps.catalog.utils import get_programs
+from .models import ExternalId, ExternalIdType
+
+LOGGER = getLogger(__name__)
+
+
+@receiver(post_save, sender='student.CourseEnrollment')
+def create_external_id_for_microbachelors_program(
+    sender, instance, created, **kwargs  # pylint: disable=unused-argument
+):
+    """
+    Watches for post_save signal for creates on the CourseEnrollment table.
+    Generate an External ID if the Enrollment is in a MicroBachelors Program
+    """
+    if (
+        created and
+        instance.user and
+        not ExternalId.user_has_external_id(
+            user=instance.user,
+            type_name=ExternalIdType.MICROBACHELORS_COACHING)
+    ):
+        mb_programs = [
+            program for program in get_programs(course=instance.course_id)
+            if program.get('type_attrs', None) and program['type_attrs']['coaching_supported']
+        ]
+        if mb_programs:
+            ExternalId.add_new_user_id(
+                user=instance.user,
+                type_name=ExternalIdType.MICROBACHELORS_COACHING
+            )

openedx/core/djangoapps/external_user_ids/tests/test_signals.py

+"""
+Signal Tests for External User Ids that are sent out of Open edX
+"""
+
+from opaque_keys.edx.keys import CourseKey
+from django.conf import settings
+from django.core.cache import cache
+from edx_django_utils.cache import RequestCache
+
+from openedx.core.djangoapps.catalog.tests.factories import (
+    CourseFactory,
+    ProgramFactory,
+)
+from student.tests.factories import TEST_PASSWORD, UserFactory
+from openedx.core.djangoapps.catalog.cache import PROGRAM_CACHE_KEY_TPL, COURSE_PROGRAMS_CACHE_KEY_TPL
+from student.models import CourseEnrollment
+from course_modes.models import CourseMode
+from openedx.core.djangolib.testing.utils import skip_unless_lms
+
+from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+
+# external_ids is not in CMS' INSTALLED_APPS so these imports will error during test collection
+if settings.ROOT_URLCONF == 'lms.urls':
+    from openedx.core.djangoapps.external_user_ids.models import ExternalId, ExternalIdType
+
+
+@skip_unless_lms
+class MicrobachelorsExternalIDTest(ModuleStoreTestCase, CacheIsolationTestCase):
+    """
+    Test cases for Signals for External User Ids
+    """
+    ENABLED_CACHES = ['default']
+
+    @classmethod
+    def setUpClass(cls):
+        super(MicrobachelorsExternalIDTest, cls).setUpClass()
+
+        cls.course_list = []
+        cls.user = UserFactory.create()
+        cls.course_keys = [
+            CourseKey.from_string('course-v1:edX+DemoX+Test_Course'),
+            CourseKey.from_string('course-v1:edX+DemoX+Another_Test_Course'),
+        ]
+        ExternalIdType.objects.create(
+            name=ExternalIdType.MICROBACHELORS_COACHING,
+            description='test'
+        )
+
+    def setUp(self):
+        super(MicrobachelorsExternalIDTest, self).setUp()
+        RequestCache.clear_all_namespaces()
+        self.program = self._create_cached_program()
+        self.client.login(username=self.user.username, password=TEST_PASSWORD)
+
+    def _create_cached_program(self):
+        """ helper method to create a cached program """
+        program = ProgramFactory.create()
+
+        for course_key in self.course_keys:
+            program['courses'].append(CourseFactory(id=course_key))
+
+        program['type'] = 'MicroBachelors'
+        program['type_attrs']['coaching_supported'] = True
+
+        for course in program['courses']:
+            course_run = course['course_runs'][0]['key']
+            cache.set(
+                COURSE_PROGRAMS_CACHE_KEY_TPL.format(course_run_id=course_run),
+                [program['uuid']],
+                None
+            )
+        cache.set(
+            PROGRAM_CACHE_KEY_TPL.format(uuid=program['uuid']),
+            program,
+            None
+        )
+
+        return program
+
+    def test_enroll_mb_create_external_id(self):
+        course_run_key = self.program['courses'][0]['course_runs'][0]['key']
+
+        # Enroll user
+        enrollment = CourseEnrollment.objects.create(
+            course_id=course_run_key,
+            user=self.user,
+            mode=CourseMode.VERIFIED,
+        )
+        enrollment.save()
+        external_id = ExternalId.objects.get(
+            user=self.user
+        )
+        assert external_id is not None
+        assert external_id.external_id_type.name == ExternalIdType.MICROBACHELORS_COACHING
+
+    def test_second_enroll_mb_no_new_external_id(self):
+        course_run_key1 = self.program['courses'][0]['course_runs'][0]['key']
+        course_run_key2 = self.program['courses'][1]['course_runs'][0]['key']
+
+        # Enroll user
+        CourseEnrollment.objects.create(
+            course_id=course_run_key1,
+            user=self.user,
+            mode=CourseMode.VERIFIED,
+        )
+        external_id = ExternalId.objects.get(
+            user=self.user
+        )
+        assert external_id is not None
+        assert external_id.external_id_type.name == ExternalIdType.MICROBACHELORS_COACHING
+        original_external_user_uuid = external_id.external_user_id
+
+        CourseEnrollment.objects.create(
+            course_id=course_run_key2,
+            user=self.user,
+            mode=CourseMode.VERIFIED,
+        )
+        enrollments = CourseEnrollment.objects.filter(user=self.user)
+
+        assert len(enrollments) == 2
+
+        external_ids = ExternalId.objects.filter(
+            user=self.user
+        )
+
+        assert len(external_ids) == 1
+        assert external_ids[0].external_id_type.name == ExternalIdType.MICROBACHELORS_COACHING
+        assert original_external_user_uuid == external_ids[0].external_user_id