feat: initialize agreements app

Adds an app to openedx for the Agreements feature.
Includes an IntegritySignature model with a basic
API, as well as a waffle flag to support rollout.

lms/envs/common.py

@@ -3086,6 +3086,9 @@ INSTALLED_APPS = [
     # in the LMS process at the moment, so anything that has Django admin access
     # permissions needs to be listed as an LMS app or the script will fail.
     'user_tasks',
+
+    # Agreements
+    'openedx.core.djangoapps.agreements'
 ]
 
 ######################### CSRF #########################################

openedx/core/djangoapps/agreements/admin.py

+"""
+Django admin page for the Agreements app
+"""
+
+from django.contrib import admin
+
+from openedx.core.djangoapps.agreements.models import IntegritySignature
+
+
+class IntegritySignatureAdmin(admin.ModelAdmin):
+    """
+    Admin for the IntegritySignature Model
+    """
+    list_display = ('user', 'course_key',)
+    readonly_fields = ('user', 'course_key',)
+    search_fields = ('user__username', 'course_key',)
+
+    class Meta:
+        model = IntegritySignature
+
+
+admin.site.register(IntegritySignature, IntegritySignatureAdmin)

openedx/core/djangoapps/agreements/api.py

+"""
+Agreements API
+"""
+
+import logging
+
+from django.contrib.auth import get_user_model
+from django.core.exceptions import ObjectDoesNotExist
+from opaque_keys.edx.keys import CourseKey
+
+from openedx.core.djangoapps.agreements.models import IntegritySignature
+
+log = logging.getLogger(__name__)
+User = get_user_model()
+
+
+def create_integrity_signature(username, course_id):
+    """
+    Create an integrity signature. If a signature already exists, do not create a new one.
+
+    Arguments:
+        * username (str)
+        * course_id (str)
+
+    Returns:
+        * IntegritySignature object
+    """
+    user = User.objects.get(username=username)
+    course_key = CourseKey.from_string(course_id)
+    signature, created = IntegritySignature.objects.get_or_create(user=user, course_key=course_key)
+    if not created:
+        log.warning(
+            'Integrity signature already exists for user_id={user_id} and '
+            'course_id={course_id}'.format(user_id=user.id, course_id=course_id)
+        )
+    return signature
+
+
+def get_integrity_signature(username, course_id):
+    """
+    Get an integrity signature.
+
+    Arguments:
+        * username (str)
+        * course_id (str)
+
+    Returns:
+        * An IntegritySignature object, or None if one does not exist for the
+          user + course combination.
+    """
+    user = User.objects.get(username=username)
+    course_key = CourseKey.from_string(course_id)
+    try:
+        return IntegritySignature.objects.get(user=user, course_key=course_key)
+    except ObjectDoesNotExist:
+        return None
+
+
+def get_integrity_signatures_for_course(course_id):
+    """
+    Get all integrity signatures for a given course.
+
+    Arguments:
+        * course_id (str)
+
+    Returns:
+        * QuerySet of IntegritySignature objects (can be empty).
+    """
+    course_key = CourseKey.from_string(course_id)
+    return IntegritySignature.objects.filter(course_key=course_key)

openedx/core/djangoapps/agreements/migrations/0001_initial.py

+# Generated by Django 2.2.20 on 2021-05-07 16:53
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+import model_utils.fields
+import opaque_keys.edx.django.models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='IntegritySignature',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')),
+                ('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')),
+                ('course_key', opaque_keys.edx.django.models.CourseKeyField(db_index=True, max_length=255)),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'unique_together': {('user', 'course_key')},
+            },
+        ),
+    ]

openedx/core/djangoapps/agreements/models.py

+"""
+Agreements models
+"""
+
+from django.contrib.auth import get_user_model
+from django.db import models
+from model_utils.models import TimeStampedModel
+from opaque_keys.edx.django.models import CourseKeyField
+
+User = get_user_model()
+
+
+class IntegritySignature(TimeStampedModel):
+    """
+    This model represents an integrity signature for a user + course combination.
+
+    .. no_pii:
+    """
+    user = models.ForeignKey(User, db_index=True, on_delete=models.CASCADE)
+    course_key = CourseKeyField(max_length=255, db_index=True)
+
+    class Meta:
+        app_label = 'agreements'
+        unique_together = ('user', 'course_key')

openedx/core/djangoapps/agreements/tests/test_api.py

+"""
+Tests for the Agreements API
+"""
+import logging
+
+from testfixtures import LogCapture
+
+from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.djangoapps.agreements.api import (
+    create_integrity_signature,
+    get_integrity_signature,
+    get_integrity_signatures_for_course,
+)
+from openedx.core.djangolib.testing.utils import skip_unless_lms
+from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
+
+LOGGER_NAME = "openedx.core.djangoapps.agreements.api"
+
+
+@skip_unless_lms
+class TestIntegritySignatureApi(SharedModuleStoreTestCase):
+    """
+    Tests for the integrity signature API
+    """
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.user = UserFactory()
+        cls.course = CourseFactory()
+        cls.course_id = str(cls.course.id)
+
+    def test_create_integrity_signature(self):
+        """
+        Test to create an integrity signature
+        """
+        signature = create_integrity_signature(self.user.username, self.course_id)
+        self._assert_integrity_signature(signature)
+
+    def test_create_duplicate_integrity_signature(self):
+        """
+        Test that duplicate integrity signatures cannot be created
+        """
+        with LogCapture(LOGGER_NAME, level=logging.WARNING) as logger:
+            create_integrity_signature(self.user.username, self.course_id)
+            create_integrity_signature(self.user.username, self.course_id)
+            signature = get_integrity_signature(self.user.username, self.course_id)
+            self._assert_integrity_signature(signature)
+            logger.check((
+                LOGGER_NAME,
+                'WARNING',
+                (
+                    'Integrity signature already exists for user_id={user_id} and '
+                    'course_id={course_id}'.format(
+                        user_id=self.user.id, course_id=str(self.course_id)
+                    )
+                )
+            ))
+
+    def test_get_integrity_signature(self):
+        """
+        Test to get an integrity signature
+        """
+        create_integrity_signature(self.user.username, self.course_id)
+        signature = get_integrity_signature(self.user.username, self.course_id)
+        self._assert_integrity_signature(signature)
+
+    def test_get_nonexistent_integrity_signature(self):
+        """
+        Test that None is returned if an integrity signature does not exist
+        """
+        signature = get_integrity_signature(self.user.username, self.course_id)
+        self.assertIsNone(signature)
+
+    def test_get_integrity_signatures_for_course(self):
+        """
+        Test to get all integrity signatures for a course
+        """
+        create_integrity_signature(self.user.username, self.course_id)
+        second_user = UserFactory()
+        create_integrity_signature(second_user.username, self.course_id)
+        signatures = get_integrity_signatures_for_course(self.course_id)
+        self._assert_integrity_signature(signatures[0])
+        self.assertEqual(signatures[1].user, second_user)
+        self.assertEqual(signatures[1].course_key, self.course.id)
+
+    def test_get_integrity_signatures_for_course_empty(self):
+        """
+        Test that a course with no integrity signatures returns an empty queryset
+        """
+        signatures = get_integrity_signatures_for_course(self.course_id)
+        self.assertEqual(len(signatures), 0)
+
+    def _assert_integrity_signature(self, signature):
+        """
+        Helper function to assert the returned integrity signature has the correct
+        user and course key
+        """
+        self.assertEqual(signature.user, self.user)
+        self.assertEqual(signature.course_key, self.course.id)

openedx/core/djangoapps/agreements/toggles.py

+"""
+Toggles for the Agreements app
+"""
+
+from edx_toggles.toggles import WaffleFlag
+
+# .. toggle_name: agreements.enable_integrity_signature
+# .. toggle_implementation: WaffleFlag
+# .. toggle_default: False
+# .. toggle_description: Supports rollout of the integrity signature feature
+# .. toggle_use_cases: temporary, open_edx
+# .. toggle_creation_date: 2021-05-07
+# .. toggle_target_removal_date: None
+# .. toggle_warnings: None
+# .. toggle_tickets: MST-786
+
+ENABLE_INTEGRITY_SIGNATURE = WaffleFlag('agreements.enable_integrity_signature', __name__)
+
+
+def is_integrity_signature_enabled():
+    return ENABLE_INTEGRITY_SIGNATURE.is_enabled()

openedx/tests/settings.py

@@ -82,6 +82,7 @@ INSTALLED_APPS = (
     'openedx.core.djangoapps.theming.apps.ThemingConfig',
     'openedx.core.djangoapps.external_user_ids',
     'openedx.core.djangoapps.demographics',
+    'openedx.core.djangoapps.agreements',
 
     'lms.djangoapps.experiments',
     'openedx.features.content_type_gating',