fix: CCX LTI configuration compatibility (#391)

* fix: CCX LTI configuration compatibility

Co-authored-by: Squirrel18 <daniel.quiroga@edunext.co>
Co-authored-by: alexjmpb <alexander.mendoza@edunext.co>

* feat: Update version and changelog

---------

Co-authored-by: Squirrel18 <daniel.quiroga@edunext.co>
Co-authored-by: alexjmpb <alexander.mendoza@edunext.co>

Author: 3 people

CHANGELOG.rst

@@ -16,6 +16,10 @@ Please See the `releases tab <https://github.com/openedx/xblock-lti-consumer/rel
 Unreleased
 ~~~~~~~~~~
 
+9.6.1 - 2023-06-28
+------------------
+* Fix CCX LTI configuration compatibility
+
 9.6.0 - 2023-08-01
 ------------------
 * Added support for Django 4.2

lti_consumer/__init__.py

@@ -4,4 +4,4 @@
 from .apps import LTIConsumerApp
 from .lti_xblock import LtiConsumerXBlock
 
-__version__ = '9.6.0'
+__version__ = '9.6.1'

lti_consumer/models.py

@@ -11,6 +11,7 @@
 
 from jsonfield import JSONField
 from Cryptodome.PublicKey import RSA
+from ccx_keys.locator import CCXBlockUsageLocator
 from opaque_keys.edx.django.models import CourseKeyField, UsageKeyField
 from opaque_keys.edx.keys import CourseKey
 from config_models.models import ConfigurationModel
@@ -29,6 +30,7 @@
     get_lti_deeplinking_response_url,
     get_lti_nrps_context_membership_url,
     choose_lti_1p3_redirect_uris,
+    model_to_dict,
 )
 
 log = logging.getLogger(__name__)
@@ -270,6 +272,42 @@ def clean(self):
         if consumer is None:
             raise ValidationError(_("Invalid LTI configuration."))
 
+    def sync_configurations(self):
+        """Syncronize main/children configurations.
+
+        This method will synchronize the field values of main/children configurations.
+        On a configuration with a CCX location, it will copy the values from the main course configuration,
+        otherwise, it will try to query any children configuration and update their fields using
+        the current configuration values.
+        """
+        EXCLUDED_FIELDS = ['id', 'config_id', 'location']
+
+        if isinstance(self.location, CCXBlockUsageLocator):
+            # Query main configuration using main location.
+            main_config = LtiConfiguration.objects.filter(location=self.location.to_block_locator()).first()
+            # Copy fields from main configuration.
+            for field in model_to_dict(main_config, EXCLUDED_FIELDS).items():
+                setattr(self, field[0], field[1])
+        else:
+            try:
+                # Query child CCX configurations using location block ID and CCX namespace.
+                child_configs = LtiConfiguration.objects.filter(
+                    location__endswith=str(self.location).split('@')[-1],
+                ).filter(
+                    location__startswith=CCXBlockUsageLocator.CANONICAL_NAMESPACE,
+                ).exclude(id=self.pk)
+                # Copy fields to child CCX configurations.
+                child_configs.update(**model_to_dict(self, EXCLUDED_FIELDS))
+            except IndexError:
+                log.exception(
+                    f'Failed to query children CCX LTI configurations: '
+                    f'Failed to parse main LTI configuration location: {self.location}',
+                )
+
+    def save(self, *args, **kwargs):
+        self.sync_configurations()
+        super().save(*args, **kwargs)
+
     def _generate_lti_1p3_keys_if_missing(self):
         """
         Generate LTI 1.3 RSA256 keys if missing.

lti_consumer/tests/unit/test_models.py

@@ -3,7 +3,7 @@
 """
 from contextlib import contextmanager
 from datetime import datetime, timedelta
-from unittest.mock import patch
+from unittest.mock import patch, call
 
 import ddt
 from Cryptodome.PublicKey import RSA
@@ -12,6 +12,7 @@
 from django.utils import timezone
 from edx_django_utils.cache import RequestCache
 from jwkest.jwk import RSAKey
+from ccx_keys.locator import CCXBlockUsageLocator
 from opaque_keys.edx.locator import CourseLocator
 
 from lti_consumer.lti_xblock import LtiConsumerXBlock
@@ -415,6 +416,80 @@ def test_get_redirect_uris_for_db_model_returns_expected(
 
         assert self.lti_1p3_config_db.get_lti_1p3_redirect_uris() == expected
 
+    @patch.object(LtiConfiguration, 'sync_configurations')
+    def test_save(self, sync_configurations_mock):
+        """Test save method."""
+        self.assertEqual(self.lti_1p3_config.save(), None)
+        sync_configurations_mock.assert_called_once_with()
+
+    @patch('lti_consumer.models.isinstance', return_value=True)
+    @patch.object(LtiConfiguration.objects, 'filter')
+    @patch('lti_consumer.models.model_to_dict')
+    @patch('lti_consumer.models.setattr')
+    def test_sync_configurations_with_ccx_location(
+        self,
+        setattr_mock,
+        model_to_dict_mock,
+        filter_mock,
+        isinstance_mock,
+    ):
+        """
+        Test sync_configurations method with CCX location.
+        """
+        model_to_dict_mock.return_value = {'test': 'test'}
+        self.lti_1p3_config.location = 'ccx-block-v1:course+test+2020+ccx@1+type@problem+block@test'
+
+        self.assertEqual(self.lti_1p3_config.sync_configurations(), None)
+        isinstance_mock.assert_called_once_with(self.lti_1p3_config.location, CCXBlockUsageLocator)
+        filter_mock.assert_has_calls([
+            call(location=self.lti_1p3_config.location.to_block_locator()),
+            call().first(),
+        ])
+        model_to_dict_mock.assert_called_once_with(filter_mock.return_value.first(), ['id', 'config_id', 'location'])
+        setattr_mock.assert_called_once_with(self.lti_1p3_config, 'test', 'test')
+
+    @patch('lti_consumer.models.isinstance', return_value=False)
+    @patch.object(LtiConfiguration.objects, 'filter')
+    @patch('lti_consumer.models.model_to_dict')
+    def test_sync_configurations_with_location(
+        self,
+        model_to_dict_mock,
+        filter_mock,
+        isinstance_mock,
+    ):
+        """
+        Test sync_configurations method with location.
+        """
+        self.assertEqual(self.lti_1p3_config.sync_configurations(), None)
+        isinstance_mock.assert_called_once_with(self.lti_1p3_config.location, CCXBlockUsageLocator)
+        filter_mock.assert_has_calls([
+            call(location__endswith=str(self.lti_1p3_config.location).split('@')[-1]),
+            call().filter(location__startswith=CCXBlockUsageLocator.CANONICAL_NAMESPACE),
+            call().filter().exclude(id=self.lti_1p3_config.pk),
+            call().filter().exclude().update(**model_to_dict_mock),
+        ])
+        model_to_dict_mock.assert_called_once_with(self.lti_1p3_config, ['id', 'config_id', 'location'])
+
+    @patch('lti_consumer.models.isinstance', return_value=False)
+    @patch.object(LtiConfiguration.objects, 'filter', side_effect=IndexError())
+    @patch('lti_consumer.models.log.exception')
+    def test_sync_configurations_with_invalid_location(
+        self,
+        log_exception_mock,
+        filter_mock,
+        isinstance_mock,
+    ):
+        """
+        Test sync_configurations method with invalid location.
+        """
+        self.assertEqual(self.lti_1p3_config.sync_configurations(), None)
+        isinstance_mock.assert_called_once_with(self.lti_1p3_config.location, CCXBlockUsageLocator)
+        filter_mock.assert_called_once_with(location__endswith=str(self.lti_1p3_config.location).split('@')[-1])
+        log_exception_mock.assert_called_once_with(
+            f'Failed to query children CCX LTI configurations: '
+            f'Failed to parse main LTI configuration location: {self.lti_1p3_config.location}'
+        )
+
 
 class TestLtiAgsLineItemModel(TestCase):
     """

lti_consumer/tests/unit/test_utils.py

@@ -13,6 +13,7 @@
     get_lti_1p3_launch_data_cache_key,
     cache_lti_1p3_launch_data,
     get_data_from_cache,
+    model_to_dict,
 )
 
 LAUNCH_URL = "http://tool.launch"
@@ -137,3 +138,37 @@ def test_choose_lti_1p3_redirect_uri_returns_expected(self, launch_url, deep_lin
         )
 
         assert result == expected
+
+
+class TestModelToDict(TestCase):
+    """
+    Tests for the model_to_dict function.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.model_object = Mock()
+
+    @patch('lti_consumer.utils.copy.deepcopy', return_value={'test': 'test', '_test': 'test'})
+    def test_with_exclude_argument(self, deepcopy_mock):
+        """
+        Test model_to_dict function with exclude argument.
+        """
+        self.assertEqual(model_to_dict(self.model_object, ['test']), {})
+        deepcopy_mock.assert_called_once_with(self.model_object.__dict__)
+
+    @patch('lti_consumer.utils.copy.deepcopy', side_effect=AttributeError())
+    def test_with_attribute_error(self, deepcopy_mock):
+        """
+        Test model_to_dict function with AttributeError exception.
+        """
+        self.assertEqual(model_to_dict(self.model_object), {})
+        deepcopy_mock.assert_called_once_with(self.model_object.__dict__)
+
+    @patch('lti_consumer.utils.copy.deepcopy', side_effect=TypeError())
+    def test_with_type_error(self, deepcopy_mock):
+        """
+        Test model_to_dict function with TypeError exception.
+        """
+        self.assertEqual(model_to_dict(self.model_object), {})
+        deepcopy_mock.assert_called_once_with(self.model_object.__dict__)

lti_consumer/utils.py

@@ -1,6 +1,7 @@
 """
 Utility functions for LTI Consumer block
 """
+import copy
 import logging
 from importlib import import_module
 from urllib.parse import urlencode
@@ -332,3 +333,27 @@ def check_token_claim(token, claim_key, expected_value=None, invalid_claim_error
     if expected_value and claim_value != expected_value:
         msg = invalid_claim_error_msg if invalid_claim_error_msg else f"The claim {claim_key} value is invalid."
         raise InvalidClaimValue(msg)
+
+
+def model_to_dict(model_object, exclude=None):
+    """
+    Get dictionary from model object.
+
+    This function will create a copy of a model object in a dictionary,
+    with all private and excluded fields removed.
+    """
+    if not exclude:
+        exclude = []
+
+    try:
+        # Copy model object fields.
+        object_fields = copy.deepcopy(model_object.__dict__)
+
+        # Remove private and excluded fields.
+        for key in list(object_fields):
+            if key.startswith('_') or key in exclude:
+                object_fields.pop(key, None)
+
+        return object_fields
+    except (AttributeError, TypeError):
+        return {}

requirements/base.in

@@ -13,6 +13,7 @@ xblock-utils
 pycryptodomex
 pyjwkest
 edx-opaque-keys[django]
+edx-ccx-keys                 # Opaque keys for Custom Courses.
 django-filter
 jsonfield
 django-config-models         # Configuration models for Django allowing config management with auditing

requirements/base.txt

@@ -55,6 +55,8 @@ django-waffle==4.0.0
     # via edx-django-utils
 djangorestframework==3.14.0
     # via django-config-models
+edx-ccx-keys==1.2.1
+    # via -r requirements/base.in
 edx-django-utils==5.7.0
     # via django-config-models
 edx-opaque-keys[django]==2.4.0
@@ -138,6 +140,7 @@ simplejson==3.19.1
 six==1.16.0
     # via
     #   bleach
+    #   edx-ccx-keys
     #   fs
     #   fs-s3fs
     #   pyjwkest

requirements/ci.txt

@@ -143,6 +143,8 @@ docutils==0.20.1
     # via
     #   -r requirements/test.txt
     #   readme-renderer
+edx-ccx-keys==1.2.1
+    # via -r requirements/test.txt
 edx-django-utils==5.7.0
     # via
     #   -r requirements/test.txt
@@ -412,6 +414,7 @@ six==1.16.0
     #   -r requirements/test.txt
     #   -r requirements/tox.txt
     #   bleach
+    #   edx-ccx-keys
     #   edx-lint
     #   fs
     #   fs-s3fs

requirements/dev.txt

@@ -78,6 +78,8 @@ djangorestframework==3.14.0
     # via
     #   -r requirements/base.txt
     #   django-config-models
+edx-ccx-keys==1.2.1
+    # via -r requirements/base.txt
 edx-django-utils==5.7.0
     # via
     #   -r requirements/base.txt
@@ -202,6 +204,7 @@ six==1.16.0
     # via
     #   -r requirements/base.txt
     #   bleach
+    #   edx-ccx-keys
     #   fs
     #   fs-s3fs
     #   pyjwkest