feat: created mock_apps for ci testing

.gitignore

@@ -63,3 +63,7 @@ docs/channel_integrations.*.rst
 # Private requirements
 requirements/private.in
 requirements/private.txt
+
+# Code editor settings
+.idea/
+.vscode/

CHANGELOG.rst

@@ -16,10 +16,13 @@ Unreleased
 
 *
 
-0.1.0 – 2024-10-30
+0.1.0 – 2025-01-16
 **********************************************
 
 Added
 =====
 
+* Created ``mock_apps`` for testing purposes.
+* Updated requirements in ``base.in`` and run ``make requirements``.
+* Migrated ``integrated_channel`` app from edx-enterprise.
 * First release on PyPI.

mock_apps/consent/errors.py

@@ -0,0 +1,7 @@
+"""
+Errors thrown by the APIs in the Consent application.
+"""
+
+
+class InvalidProxyConsent(Exception):
+    """A proxy consent object with the given details could not be created."""

mock_apps/consent/mixins.py

@@ -0,0 +1,117 @@
+"""
+Mixins for edX Enterprise's Consent application.
+"""
+import logging
+
+from django.contrib import auth
+
+from enterprise.models import EnterpriseCourseEnrollment
+
+LOGGER = logging.getLogger(__name__)
+User = auth.get_user_model()
+
+
+class ConsentModelMixin:
+    """
+    A mixin for Data Sharing Consent classes that require common, reusable functionality.
+    """
+
+    def __str__(self):
+        """
+        Return a human-readable string representation.
+        """
+        return "<{class_name} for user {username} of Enterprise {enterprise_name}>".format(
+            class_name=self.__class__.__name__,
+            username=self.username,
+            enterprise_name=self.enterprise_customer.name,
+        )
+
+    def __repr__(self):
+        """
+        Return a uniquely identifying string representation.
+        """
+        return self.__str__()
+
+    def consent_required(self):
+        """
+        Return a boolean value indicating whether a consent action must be taken.
+        """
+        children = getattr(self, '_child_consents', [])
+        if children:
+            return any(child.consent_required() for child in children)
+
+        if self.granted:
+            return False
+
+        required = bool(
+            (self.enterprise_customer.enforces_data_sharing_consent('at_enrollment')) and
+            (self.enterprise_customer.catalog_contains_course(self.course_id))
+        )
+
+        if not required and self.enterprise_customer.enforces_data_sharing_consent('at_enrollment'):
+            LOGGER.info(
+                '[ENTERPRISE DSC] Consent not required becuase catalog does not contain course. '
+                'Course: [%s], Username: [%s], Enterprise: [%s], Exists: [%s]',
+                self.course_id,
+                self.username,
+                self.enterprise_customer.uuid,
+                self.exists,
+            )
+
+        return required
+
+    @property
+    def enterprise_enrollment_exists(self):
+        """
+        Determine whether there exists an EnterpriseCourseEnrollment related to this consent record.
+        """
+        if self.course_id:
+            try:
+                user_id = User.objects.get(username=self.username).pk
+            except User.DoesNotExist:
+                return False
+            return EnterpriseCourseEnrollment.objects.filter(
+                course_id=self.course_id,
+                enterprise_customer_user__user_id=user_id,
+                enterprise_customer_user__enterprise_customer=self.enterprise_customer,
+            ).exists()
+        return False
+
+    @property
+    def exists(self):
+        """
+        Determine whether a record related to the consent scenario exists.
+
+        First, check the instance's own `_exists` attribute; this is set to True
+        on database-backed instances that have a primary key, and may be manually
+        set to true on ProxyDataSharingConsent objects that have database-backed
+        children. If unsuccessful, check to see if an EnterpriseCourseEnrollment
+        related to this consent record exists; we treat that as though this record
+        exists for the purposes of serializable API responses.
+
+        We want to check for EnterpriseCourseEnrollment records because there are
+        cases where one will be created, but not the other. In particular, proxy
+        enrollments create an ECE but not any consent record. The LMS uses the
+        API's 'exists' key to determine if consent action should be taken for course
+        enrollments that have prior existence but for which consent has not been
+        granted. Thus, 'exists' is used as a proxy for the question "has any workflow
+        been entered which may involve a necessity for the learner to grant consent?"
+        """
+        return self._exists or self.enterprise_enrollment_exists
+
+    def serialize(self):
+        """
+        Return a dictionary that provides the core details of the consent record.
+        """
+        details = {
+            'username': self.username,
+            'enterprise_customer_uuid': self.enterprise_customer.uuid,
+            'exists': self.exists,
+            'consent_provided': self.granted,
+            'consent_required': self.consent_required(),
+        }
+        if self.course_id:
+            details['course_id'] = self.course_id
+        if getattr(self, 'program_uuid', None):
+            details['program_uuid'] = self.program_uuid
+        return details

mock_apps/consent/models.py

@@ -0,0 +1,248 @@
+from simple_history.models import HistoricalRecords
+
+from django.core.exceptions import ImproperlyConfigured
+from django.contrib import auth
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+from model_utils.models import TimeStampedModel
+
+from consent.errors import InvalidProxyConsent
+from consent.mixins import ConsentModelMixin
+from enterprise.api_client.discovery import get_course_catalog_api_service_client
+from enterprise.logging import getEnterpriseLogger
+from enterprise.models import EnterpriseCustomer
+
+LOGGER = getEnterpriseLogger(__name__)
+User = auth.get_user_model()
+
+
+class DataSharingConsentQuerySet(models.query.QuerySet):
+    """
+    Customized QuerySets for the ``DataSharingConsent`` model.
+
+    When searching for any ``DataSharingConsent`` object, if it doesn't exist, return a single
+    ``ProxyDataSharingConsent`` object which behaves just like a ``DataSharingConsent`` object
+    except is not saved in the database until committed.
+    """
+
+    def proxied_get(self, *args, **kwargs):
+        """
+        Perform the query and returns a single object matching the given keyword arguments.
+
+        This customizes the queryset to return an instance of ``ProxyDataSharingConsent`` when
+        the searched-for ``DataSharingConsent`` instance does not exist.
+        """
+        original_kwargs = kwargs.copy()
+        if 'course_id' in kwargs:
+            try:
+                # Try to get the record for the course OR course run, depending on what we got in kwargs,
+                # course_id or course_run_id
+                return self.get(*args, **kwargs)
+            except DataSharingConsent.DoesNotExist:
+                # If here, either the record for course OR course run doesn't exist.
+                # Try one more time by modifying the query parameters to look for just a course record this time.
+                site = None
+                if 'enterprise_customer' in kwargs:
+                    site = kwargs['enterprise_customer'].site
+
+                try:
+                    course_id = get_course_catalog_api_service_client(site=site).get_course_id(
+                        course_identifier=kwargs['course_id']
+                    )
+                    kwargs['course_id'] = course_id
+                except ImproperlyConfigured:
+                    LOGGER.warning('CourseCatalogApiServiceClient is improperly configured.')
+
+        try:
+            # Try to get the record of course
+            return self.get(*args, **kwargs)
+        except DataSharingConsent.DoesNotExist:
+            # If here, the record doesn't exist for course AND course run, so return a proxy record instead.
+            return ProxyDataSharingConsent(**original_kwargs)
+
+
+class DataSharingConsentManager(models.Manager.from_queryset(DataSharingConsentQuerySet)):
+    """
+    Model manager for :class:`.DataSharingConsent` model.
+
+    Uses a QuerySet that returns a ``ProxyDataSharingConsent`` object when the searched-for
+    ``DataSharingConsent`` object does not exist. Otherwise behaves the same as a normal manager.
+    """
+
+
+class ProxyDataSharingConsent(ConsentModelMixin):
+    """
+    A proxy-model of the ``DataSharingConsent`` model; it's not a real model, but roughly behaves like one.
+
+    Upon commit, a real ``DataSharingConsent`` object which mirrors the ``ProxyDataSharingConsent`` object's
+    pseudo-model-fields is created, returned, and saved in the database. The remnant, in-heap
+    ``ProxyDataSharingConsent`` object may be deleted afterwards, but if not, its ``exists`` fields remains ``True``
+    to indicate that the object has been committed.
+
+    NOTE: This class will be utilized when we implement program level consent by having an abstraction over these
+          consent objects per course.
+    """
+
+    objects = DataSharingConsentManager()
+
+    def __init__(
+            self,
+            enterprise_customer=None,
+            username='',
+            course_id='',
+            program_uuid='',
+            granted=False,
+            exists=False,
+            child_consents=None,
+            **kwargs
+    ):
+        """
+        Initialize a proxy version of ``DataSharingConsent`` which behaves similarly but does not exist in the DB.
+        """
+        ec_keys = {}
+        for key, value in kwargs.items():
+            if str(key).startswith('enterprise_customer__'):
+                enterprise_customer_detail = key[len('enterprise_customer__'):]
+                ec_keys[enterprise_customer_detail] = value
+
+        if ec_keys:
+            enterprise_customer = EnterpriseCustomer.objects.get(**ec_keys)
+
+        self.enterprise_customer = enterprise_customer
+        self.username = username
+        self.course_id = course_id
+        self.program_uuid = program_uuid
+        self.granted = granted
+        self._exists = exists
+        self._child_consents = child_consents or []
+
+    @classmethod
+    def from_children(cls, program_uuid, *children):
+        """
+        Build a ProxyDataSharingConsent using the details of the received consent records.
+        """
+        if not children or any(child is None for child in children):
+            return None
+        granted = all(child.granted for child in children)
+        exists = any(child.exists for child in children)
+        usernames = {child.username for child in children}
+        enterprises = {child.enterprise_customer for child in children}
+        if not len(usernames) == len(enterprises) == 1:
+            raise InvalidProxyConsent(
+                'Children used to create a bulk proxy consent object must '
+                'share a single common username and EnterpriseCustomer.'
+            )
+        username = children[0].username
+        enterprise_customer = children[0].enterprise_customer
+        return cls(
+            enterprise_customer=enterprise_customer,
+            username=username,
+            program_uuid=program_uuid,
+            exists=exists,
+            granted=granted,
+            child_consents=children
+        )
+
+    def commit(self):
+        """
+        Commit a real ``DataSharingConsent`` object to the database, mirroring current field settings.
+
+        :return: A ``DataSharingConsent`` object if validation is successful, otherwise ``None``.
+        """
+        if self._child_consents:
+            consents = []
+
+            for consent in self._child_consents:
+                consent.granted = self.granted
+                consents.append(consent.save() or consent)
+
+            return ProxyDataSharingConsent.from_children(self.program_uuid, *consents)
+
+        consent, _ = DataSharingConsent.objects.update_or_create(
+            enterprise_customer=self.enterprise_customer,
+            username=self.username,
+            course_id=self.course_id,
+            defaults={
+                'granted': self.granted
+            }
+        )
+        self._exists = consent.exists
+        return consent
+
+    def save(self, *args, **kwargs):
+        """
+        Synonym function for ``commit``.
+        """
+        return self.commit()
+
+
+class Consent(TimeStampedModel):
+    """
+    An abstract base model for representing any type of consent.
+    """
+
+    class Meta:
+        """
+        Meta class for the ``Consent`` model.
+        """
+
+        abstract = True
+        app_label = 'consent'
+        indexes = [
+            models.Index(fields=['username'])
+        ]
+
+    enterprise_customer = models.ForeignKey(
+        EnterpriseCustomer,
+        blank=False,
+        null=False,
+        related_name='enterprise_customer_consent',
+        on_delete=models.deletion.CASCADE
+    )
+    username = models.CharField(
+        max_length=255,
+        blank=False,
+        null=False,
+        help_text=_("Name of the user whose consent state is stored.")
+    )
+    granted = models.BooleanField(null=True, help_text=_("Whether consent is granted."))
+
+    @property
+    def _exists(self):
+        """
+        Return whether the instance exists or not.
+        """
+        return bool(self.pk)
+
+
+class DataSharingConsent(ConsentModelMixin, Consent):
+    """
+    An abstract representation of Data Sharing Consent granted to an Enterprise for a course by a User.
+
+    The model is used to store a persistent, historical consent state for users granting, not granting, or revoking
+    data sharing consent to an Enterprise for a course.
+
+    .. pii: The username field inherited from Consent contains PII.
+    .. pii_types: username
+    .. pii_retirement: consumer_api
+    """
+
+    class Meta(Consent.Meta):
+        """
+        Meta class for the ``DataSharingConsent`` model.
+        """
+
+        abstract = False
+        verbose_name = _("Data Sharing Consent Record")
+        verbose_name_plural = _("Data Sharing Consent Records")
+        unique_together = (("enterprise_customer", "username", "course_id"),)
+
+    objects = DataSharingConsentManager()
+
+    course_id = models.CharField(
+        max_length=255,
+        blank=False,
+        help_text=_("Course key for which data sharing consent is granted.")
+    )
+    history = HistoricalRecords()