diff --git a/README.rst b/README.rst index 2fbc243..ca01d45 100644 --- a/README.rst +++ b/README.rst @@ -42,6 +42,13 @@ Below are the existing sink names and their corresponding object names (as neede ``CatalogPathway`` model and marks the pathway deleted in ``catalog_pathway``. - ``PathwayEnrollmentSink`` - Listens for the ``post_save`` Django signal on the openedx-core ``PathwayEnrollment`` model. Object name: ``pathway_enrollment`` +- ``StudentCompetencyStatusSink`` - Listens for the ``post_save`` Django signal on the openedx-core + ``StudentCompetencyStatus`` model (a learner's mastery status for one competency). Object name: + ``student_competency_status`` +- ``StudentCompetencyCriteriaGroupStatusSink`` - Listens for the ``post_save`` Django signal on the + openedx-core ``StudentCompetencyCriteriaGroupStatus`` model (a learner's mastery status for one + criteria group, with the group's competency, course and level). Object name: + ``student_competency_criteria_group_status`` - ``UserRetirementSink`` - Listen for the ``USER_RETIRE_LMS_MISC`` Django signal and remove the user PII information from ClickHouse. This is a special sink with no assigned object name. diff --git a/src/platform_plugin_aspects/settings/common.py b/src/platform_plugin_aspects/settings/common.py index 37ba256..1fe0417 100644 --- a/src/platform_plugin_aspects/settings/common.py +++ b/src/platform_plugin_aspects/settings/common.py @@ -111,6 +111,14 @@ def plugin_settings(settings): "module": "openedx_catalog.models", "model": "PathwayEnrollment", }, + "student_competency_status": { + "module": "openedx_learning.models", + "model": "StudentCompetencyStatus", + }, + "student_competency_criteria_group_status": { + "module": "openedx_learning.models", + "model": "StudentCompetencyCriteriaGroupStatus", + }, } # This is on by default here, off by default in tutor-contrib-aspects until we only diff --git a/src/platform_plugin_aspects/signals.py b/src/platform_plugin_aspects/signals.py index 6fbe06f..63045bc 100644 --- a/src/platform_plugin_aspects/signals.py +++ b/src/platform_plugin_aspects/signals.py @@ -14,6 +14,8 @@ ExternalIdSink, ObjectTagSink, PathwayEnrollmentSink, + StudentCompetencyCriteriaGroupStatusSink, + StudentCompetencyStatusSink, TagSink, TaxonomySink, UserProfileSink, @@ -408,3 +410,89 @@ def on_pathway_enrollment_saved( # pylint: disable=unused-argument # pragma: n _pathway_enrollment = get_model("pathway_enrollment") if _pathway_enrollment: # pragma: no cover post_save.connect(on_pathway_enrollment_saved_txn, sender=_pathway_enrollment) + + +def on_student_competency_status_saved_txn(*args, **kwargs): + """ + Handle student_competency_status saves in the middle of a transaction. + + The status is written inside the LMS grade transaction, so queuing the dump + before the commit could read a row that doesn't exist yet. This postpones + queuing the Celery task until after the transaction is committed. + + openedx-core raises a stored status with a bulk UPDATE and sends post_save + itself, so raises arrive here as well as creates. + """ + + def on_student_competency_status_saved( # pylint: disable=unused-argument # pragma: no cover + sender, instance, **kwargs + ): + """ + Receives post save signal and queues the dump job. + """ + # import here, because signal is registered at startup, but items in tasks are not yet able to be loaded + from platform_plugin_aspects.tasks import ( # pylint: disable=import-outside-toplevel + dump_data_to_clickhouse, + ) + + sink = StudentCompetencyStatusSink(None, None) + dump_data_to_clickhouse.delay( + sink_module=sink.__module__, + sink_name=sink.__class__.__name__, + object_id=str(instance.id), + ) + + transaction.on_commit(lambda: on_student_competency_status_saved(*args, **kwargs)) + + +# Connect the StudentCompetencyStatus.post_save signal handler only if we have a model to attach to. +# (prevents celery errors during tests) +_student_competency_status = get_model("student_competency_status") +if _student_competency_status: # pragma: no cover + post_save.connect( + on_student_competency_status_saved_txn, sender=_student_competency_status + ) + + +def on_student_competency_criteria_group_status_saved_txn(*args, **kwargs): + """ + Handle student_competency_criteria_group_status saves in the middle of a transaction. + + Group statuses are written by the rollup, inside a transaction per level, so + queuing the dump before the commit could read a row that doesn't exist yet. + openedx-core sends post_save itself when it raises a stored status. + """ + + def on_student_competency_criteria_group_status_saved( # pylint: disable=unused-argument # pragma: no cover + sender, instance, **kwargs + ): + """ + Receives post save signal and queues the dump job. + """ + # import here, because signal is registered at startup, but items in tasks are not yet able to be loaded + from platform_plugin_aspects.tasks import ( # pylint: disable=import-outside-toplevel + dump_data_to_clickhouse, + ) + + sink = StudentCompetencyCriteriaGroupStatusSink(None, None) + dump_data_to_clickhouse.delay( + sink_module=sink.__module__, + sink_name=sink.__class__.__name__, + object_id=str(instance.id), + ) + + transaction.on_commit( + lambda: on_student_competency_criteria_group_status_saved(*args, **kwargs) + ) + + +# Connect the StudentCompetencyCriteriaGroupStatus.post_save signal handler only if we have a +# model to attach to. (prevents celery errors during tests) +_student_competency_criteria_group_status = get_model( + "student_competency_criteria_group_status" +) +if _student_competency_criteria_group_status: # pragma: no cover + post_save.connect( + on_student_competency_criteria_group_status_saved_txn, + sender=_student_competency_criteria_group_status, + ) diff --git a/src/platform_plugin_aspects/sinks/__init__.py b/src/platform_plugin_aspects/sinks/__init__.py index 7f9f283..2d8539a 100644 --- a/src/platform_plugin_aspects/sinks/__init__.py +++ b/src/platform_plugin_aspects/sinks/__init__.py @@ -8,6 +8,10 @@ from .course_overview_sink import CourseOverviewSink, XBlockSink from .external_id_sink import ExternalIdSink from .pathway_enrollment_sink import PathwayEnrollmentSink +from .student_competency_status_sink import ( + StudentCompetencyCriteriaGroupStatusSink, + StudentCompetencyStatusSink, +) from .tag_sink import ObjectTagSink, TagSink, TaxonomySink from .user_profile_sink import UserProfileSink from .user_retire_sink import UserRetirementSink diff --git a/src/platform_plugin_aspects/sinks/serializers.py b/src/platform_plugin_aspects/sinks/serializers.py index 73caa0f..7964af0 100644 --- a/src/platform_plugin_aspects/sinks/serializers.py +++ b/src/platform_plugin_aspects/sinks/serializers.py @@ -279,6 +279,105 @@ def get_actor_id(self, instance): return get_or_create_xapi_actor_id(instance.user) +class StudentCompetencyStatusSerializer( + BaseSinkSerializer, serializers.ModelSerializer +): + """ + Serializer for the openedx-core StudentCompetencyStatus model. + + Statuses only rise, so ``created`` is when the learner first got a status for the + competency and ``modified`` on a Demonstrated row is when they demonstrated it. + """ + + actor_id = serializers.SerializerMethodField() + taxonomy_id = serializers.IntegerField(source="tag.taxonomy_id") + status = serializers.CharField(source="status.status") + + class Meta: + """Meta class for the StudentCompetencyStatusSerializer""" + + model = get_model("student_competency_status") + fields = [ + "id", + "actor_id", + "tag_id", + "taxonomy_id", + "status_id", + "status", + "created", + "modified", + "dump_id", + "time_last_dumped", + ] + + def get_actor_id(self, instance): + """Return the learner's xAPI actor id instead of any user identifier.""" + return get_or_create_xapi_actor_id(instance.user) + + +class StudentCompetencyCriteriaGroupStatusSerializer( + BaseSinkSerializer, serializers.ModelSerializer +): + """ + Serializer for the openedx-core StudentCompetencyCriteriaGroupStatus model. + + The group's competency, course and level are denormalized so ClickHouse needs no + criteria-group table: ``group_level`` is ``root`` (no parent), ``course`` (the + group scoping a competency to one course run) or ``leaf`` (below a course group). + ``course_key`` is the group's own course run, or for a leaf, its course group's. + """ + + actor_id = serializers.SerializerMethodField() + group_id = serializers.IntegerField(source="group.id") + parent_group_id = serializers.IntegerField(source="group.parent_id", allow_null=True) + tag_id = serializers.IntegerField(source="group.tag_id") + taxonomy_id = serializers.IntegerField(source="group.tag.taxonomy_id") + course_key = serializers.SerializerMethodField() + group_level = serializers.SerializerMethodField() + status = serializers.CharField(source="status.status") + + class Meta: + """Meta class for the StudentCompetencyCriteriaGroupStatusSerializer""" + + model = get_model("student_competency_criteria_group_status") + fields = [ + "id", + "actor_id", + "group_id", + "parent_group_id", + "tag_id", + "taxonomy_id", + "course_key", + "group_level", + "status_id", + "status", + "created", + "modified", + "dump_id", + "time_last_dumped", + ] + + def get_actor_id(self, instance): + """Return the learner's xAPI actor id instead of any user identifier.""" + return get_or_create_xapi_actor_id(instance.user) + + def get_group_level(self, instance): + """Return root, course or leaf for the group.""" + group = instance.group + if group.parent_id is None: + return "root" + if group.course_id is not None: + return "course" + return "leaf" + + def get_course_key(self, instance): + """Return the course key of the group's course run, or its nearest ancestor's; empty for a root.""" + group = instance.group + while group is not None and group.course_id is None: + group = group.parent + return str(group.course.course_key) if group is not None else "" + + class TagSerializer(BaseSinkSerializer, serializers.ModelSerializer): """Serializer for the Tag model.""" diff --git a/src/platform_plugin_aspects/sinks/student_competency_status_sink.py b/src/platform_plugin_aspects/sinks/student_competency_status_sink.py new file mode 100644 index 0000000..dc156c1 --- /dev/null +++ b/src/platform_plugin_aspects/sinks/student_competency_status_sink.py @@ -0,0 +1,45 @@ +"""Student Competency Status sinks: per competency and per criteria group""" + +from platform_plugin_aspects.sinks.base_sink import ModelBaseSink +from platform_plugin_aspects.sinks.serializers import ( + StudentCompetencyCriteriaGroupStatusSerializer, + StudentCompetencyStatusSerializer, +) + + +class StudentCompetencyStatusSink(ModelBaseSink): # pylint: disable=abstract-method + """ + Sink for the openedx-core StudentCompetencyStatus model: a learner's current mastery status for one competency + """ + + model = "student_competency_status" + unique_key = "id" + clickhouse_table_name = "student_competency_status" + timestamp_field = "time_last_dumped" + name = "Student Competency Status" + serializer_class = StudentCompetencyStatusSerializer + + def get_queryset(self, start_pk=None): + return super().get_queryset(start_pk).select_related("user", "tag", "status") + + +class StudentCompetencyCriteriaGroupStatusSink(ModelBaseSink): # pylint: disable=abstract-method + """ + Sink for the openedx-core StudentCompetencyCriteriaGroupStatus model: a learner's current + mastery status for one criteria group, which is the only place AttemptedNotDemonstrated is + stored and, for course-level groups, the learner's status in one course + """ + + model = "student_competency_criteria_group_status" + unique_key = "id" + clickhouse_table_name = "student_competency_criteria_group_status" + timestamp_field = "time_last_dumped" + name = "Student Competency Criteria Group Status" + serializer_class = StudentCompetencyCriteriaGroupStatusSerializer + + def get_queryset(self, start_pk=None): + return ( + super() + .get_queryset(start_pk) + .select_related("user", "status", "group__tag", "group__course") + ) diff --git a/src/platform_plugin_aspects/sinks/tests/test_serializers.py b/src/platform_plugin_aspects/sinks/tests/test_serializers.py index c34406a..4a71171 100644 --- a/src/platform_plugin_aspects/sinks/tests/test_serializers.py +++ b/src/platform_plugin_aspects/sinks/tests/test_serializers.py @@ -10,6 +10,8 @@ CourseOverviewSerializer, DateTimeJSONEncoder, PathwayEnrollmentSerializer, + StudentCompetencyCriteriaGroupStatusSerializer, + StudentCompetencyStatusSerializer, ) from test_utils.helpers import course_key_factory @@ -132,3 +134,78 @@ def test_get_actor_id(self, mock_get_or_create_xapi_actor_id): assert actor_id == "29e5b22b-b457-49fa-9633-6c0123305d18" mock_get_or_create_xapi_actor_id.assert_called_once_with(enrollment.user) + + +class TestStudentCompetencyStatusSerializer(TestCase): + """ + Test StudentCompetencyStatusSerializer + """ + + @patch("platform_plugin_aspects.sinks.serializers.get_or_create_xapi_actor_id") + def test_get_actor_id(self, mock_get_xapi_actor_id): + """ + Test that the learner is serialized as their xAPI actor id. + """ + mock_get_xapi_actor_id.return_value = "29e5b22b-b457-49fa-9633-6c0123305d18" + status = Mock() + + actor_id = StudentCompetencyStatusSerializer().get_actor_id(status) + + assert actor_id == "29e5b22b-b457-49fa-9633-6c0123305d18" + mock_get_xapi_actor_id.assert_called_once_with(status.user) + + +class TestStudentCompetencyCriteriaGroupStatusSerializer(TestCase): + """ + Test StudentCompetencyCriteriaGroupStatusSerializer + """ + + def setUp(self): + self.serializer = StudentCompetencyCriteriaGroupStatusSerializer() + # "parent" is a Mock constructor argument, so it is set as an attribute. + self.root = Mock(parent_id=None, course_id=None) + self.root.parent = None + self.course_group = Mock(parent_id=1, course_id=7) + self.course_group.parent = self.root + self.course_group.course.course_key = "course-v1:Org+Course+Run" + self.leaf = Mock(parent_id=2, course_id=None) + self.leaf.parent = self.course_group + + def test_group_level(self): + """ + Test that groups are serialized as root, course or leaf. + """ + levels = [ + self.serializer.get_group_level(Mock(group=group)) + for group in (self.root, self.course_group, self.leaf) + ] + + assert levels == ["root", "course", "leaf"] + + def test_course_key(self): + """ + Test that a leaf takes its course group's course key and a root has none. + """ + assert ( + self.serializer.get_course_key(Mock(group=self.course_group)) + == "course-v1:Org+Course+Run" + ) + assert ( + self.serializer.get_course_key(Mock(group=self.leaf)) + == "course-v1:Org+Course+Run" + ) + assert self.serializer.get_course_key(Mock(group=self.root)) == "" + + @patch("platform_plugin_aspects.sinks.serializers.get_or_create_xapi_actor_id") + def test_get_actor_id(self, mock_get_or_create_xapi_actor_id): + """ + Test that the learner is serialized as their xAPI actor id. + """ + mock_get_or_create_xapi_actor_id.return_value = "29e5b22b-b457-49fa-9633-6c0123305d18" + status = Mock() + + assert ( + self.serializer.get_actor_id(status) + == "29e5b22b-b457-49fa-9633-6c0123305d18" + ) + mock_get_or_create_xapi_actor_id.assert_called_once_with(status.user) diff --git a/src/platform_plugin_aspects/sinks/tests/test_student_competency_status_sink.py b/src/platform_plugin_aspects/sinks/tests/test_student_competency_status_sink.py new file mode 100644 index 0000000..f80dec7 --- /dev/null +++ b/src/platform_plugin_aspects/sinks/tests/test_student_competency_status_sink.py @@ -0,0 +1,40 @@ +""" +Test the student competency status sink module. +""" + +from unittest.mock import patch + +from platform_plugin_aspects.sinks import ( + StudentCompetencyCriteriaGroupStatusSink, + StudentCompetencyStatusSink, +) + + +@patch("platform_plugin_aspects.sinks.ModelBaseSink.get_queryset") +def test_get_queryset(mock_get_queryset): + """ + Test the get_queryset method. + """ + sink = StudentCompetencyStatusSink(None, None) + + sink.get_queryset() + + mock_get_queryset.assert_called_once_with(None) + mock_get_queryset.return_value.select_related.assert_called_once_with( + "user", "tag", "status" + ) + + +@patch("platform_plugin_aspects.sinks.ModelBaseSink.get_queryset") +def test_group_status_get_queryset(mock_get_queryset): + """ + Test the group status sink loads the group's tag and course with each row. + """ + sink = StudentCompetencyCriteriaGroupStatusSink(None, None) + + sink.get_queryset() + + mock_get_queryset.assert_called_once_with(None) + mock_get_queryset.return_value.select_related.assert_called_once_with( + "user", "status", "group__tag", "group__course" + ) diff --git a/src/platform_plugin_aspects/tests/test_signals.py b/src/platform_plugin_aspects/tests/test_signals.py index 9ef61b7..a957690 100644 --- a/src/platform_plugin_aspects/tests/test_signals.py +++ b/src/platform_plugin_aspects/tests/test_signals.py @@ -11,6 +11,8 @@ on_catalog_pathway_saved_txn, on_externalid_saved_txn, on_pathway_enrollment_saved_txn, + on_student_competency_criteria_group_status_saved_txn, + on_student_competency_status_saved_txn, on_user_retirement, receive_course_publish, ) @@ -88,6 +90,28 @@ def test_on_catalog_pathway_deleted(self, mock_transaction, mock_dump_task): object_id="7", ) + @patch("platform_plugin_aspects.signals.transaction") + def test_on_student_competency_status_saved(self, mock_transaction): + """ + Test that on_student_competency_status_saved_txn defers the dump to on_commit. + """ + instance = Mock() + sender = Mock() + on_student_competency_status_saved_txn(sender, instance) + + mock_transaction.on_commit.assert_called_once() + + @patch("platform_plugin_aspects.signals.transaction") + def test_on_student_competency_criteria_group_status_saved(self, mock_transaction): + """ + Test that on_student_competency_criteria_group_status_saved_txn defers the dump to on_commit. + """ + instance = Mock() + sender = Mock() + on_student_competency_criteria_group_status_saved_txn(sender, instance) + + mock_transaction.on_commit.assert_called_once() + @patch("platform_plugin_aspects.tasks.dump_data_to_clickhouse") def test_on_user_retirement(self, mock_dump_task): """