From c20dbdb2537dd4318718579a2ef247f3111da62a Mon Sep 17 00:00:00 2001 From: Felipe Montoya Date: Wed, 28 Oct 2026 18:22:04 -0500 Subject: [PATCH 1/4] feat: add catalog_pathway sink for openedx-core Catalog Pathways Co-Authored-By: Claude Opus 5.5 --- .../settings/common.py | 4 ++ src/platform_plugin_aspects/signals.py | 39 +++++++++++++++++++ src/platform_plugin_aspects/sinks/__init__.py | 1 + .../sinks/catalog_pathway_sink.py | 20 ++++++++++ .../sinks/serializers.py | 25 ++++++++++++ .../sinks/tests/test_catalog_pathway_sink.py | 22 +++++++++++ .../tests/test_signals.py | 12 ++++++ 7 files changed, 123 insertions(+) create mode 100644 src/platform_plugin_aspects/sinks/catalog_pathway_sink.py create mode 100644 src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py diff --git a/src/platform_plugin_aspects/settings/common.py b/src/platform_plugin_aspects/settings/common.py index fbd991bd..c45d2980 100644 --- a/src/platform_plugin_aspects/settings/common.py +++ b/src/platform_plugin_aspects/settings/common.py @@ -103,6 +103,10 @@ def plugin_settings(settings): "module": "openedx_tagging.models.base", "model": "ObjectTag", }, + "catalog_pathway": { + "module": "openedx_catalog.models", + "model": "CatalogPathway", + }, } # 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 5d1ef689..3ba5e928 100644 --- a/src/platform_plugin_aspects/signals.py +++ b/src/platform_plugin_aspects/signals.py @@ -8,6 +8,7 @@ from opaque_keys import InvalidKeyError from platform_plugin_aspects.sinks import ( + CatalogPathwaySink, CourseEnrollmentSink, ExternalIdSink, ObjectTagSink, @@ -303,3 +304,41 @@ def on_object_tag_deleted( # pylint: disable=unused-argument # pragma: no cove if _object_tag: # pragma: no cover post_save.connect(on_object_tag_saved_txn, sender=_object_tag) post_delete.connect(on_object_tag_deleted, sender=_object_tag) + + +def on_catalog_pathway_saved_txn(*args, **kwargs): + """ + Handle catalog_pathway saves in the middle of a transaction. + + Handle saves in the middle of a transaction. + If this gets fired before the transaction commits, the task may try to + query an id that doesn't exist yet and throw an error. This should postpone + queuing the Celery task until after the transaction is committed. + """ + + def on_catalog_pathway_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 = CatalogPathwaySink(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_catalog_pathway_saved(*args, **kwargs)) + + +# Connect the CatalogPathway.post_save signal handler only if we have a model to attach to. +# (prevents celery errors during tests) +_catalog_pathway = get_model("catalog_pathway") +if _catalog_pathway: # pragma: no cover + post_save.connect(on_catalog_pathway_saved_txn, sender=_catalog_pathway) diff --git a/src/platform_plugin_aspects/sinks/__init__.py b/src/platform_plugin_aspects/sinks/__init__.py index f3c059a6..747e012e 100644 --- a/src/platform_plugin_aspects/sinks/__init__.py +++ b/src/platform_plugin_aspects/sinks/__init__.py @@ -3,6 +3,7 @@ """ from .base_sink import BaseSink, ModelBaseSink +from .catalog_pathway_sink import CatalogPathwaySink from .course_enrollment_sink import CourseEnrollmentSink from .course_overview_sink import CourseOverviewSink, XBlockSink from .external_id_sink import ExternalIdSink diff --git a/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py b/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py new file mode 100644 index 00000000..911f83b7 --- /dev/null +++ b/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py @@ -0,0 +1,20 @@ +"""Catalog Pathway sink""" + +from platform_plugin_aspects.sinks.base_sink import ModelBaseSink +from platform_plugin_aspects.sinks.serializers import CatalogPathwaySerializer + + +class CatalogPathwaySink(ModelBaseSink): # pylint: disable=abstract-method + """ + Sink for the openedx-core CatalogPathway model + """ + + model = "catalog_pathway" + unique_key = "id" + clickhouse_table_name = "catalog_pathway" + timestamp_field = "time_last_dumped" + name = "Catalog Pathway" + serializer_class = CatalogPathwaySerializer + + def get_queryset(self, start_pk=None): + return super().get_queryset(start_pk).select_related("org", "category") diff --git a/src/platform_plugin_aspects/sinks/serializers.py b/src/platform_plugin_aspects/sinks/serializers.py index b20315d1..d994ec91 100644 --- a/src/platform_plugin_aspects/sinks/serializers.py +++ b/src/platform_plugin_aspects/sinks/serializers.py @@ -210,6 +210,31 @@ def get_course_key(self, obj): return str(obj.course_id) +class CatalogPathwaySerializer(BaseSinkSerializer, serializers.ModelSerializer): + """Serializer for the CatalogPathway model.""" + + org = serializers.CharField(source="org.short_name") + category_code = serializers.CharField(source="category.category_code") + category_title = serializers.CharField(source="category.title") + + class Meta: + """Meta class for the CatalogPathwaySerializer""" + + model = get_model("catalog_pathway") + fields = [ + "id", + "org", + "pathway_code", + "title", + "category_code", + "category_title", + "created", + "modified", + "dump_id", + "time_last_dumped", + ] + + class TagSerializer(BaseSinkSerializer, serializers.ModelSerializer): """Serializer for the Tag model.""" diff --git a/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py b/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py new file mode 100644 index 00000000..786dba40 --- /dev/null +++ b/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py @@ -0,0 +1,22 @@ +""" +Test the catalog pathway sink module. +""" + +from unittest.mock import patch + +from platform_plugin_aspects.sinks import CatalogPathwaySink + + +@patch("platform_plugin_aspects.sinks.ModelBaseSink.get_queryset") +def test_get_queryset(mock_get_queryset): + """ + Test the get_queryset method. + """ + sink = CatalogPathwaySink(None, None) + + sink.get_queryset() + + mock_get_queryset.assert_called_once_with(None) + mock_get_queryset.return_value.select_related.assert_called_once_with( + "org", "category" + ) diff --git a/src/platform_plugin_aspects/tests/test_signals.py b/src/platform_plugin_aspects/tests/test_signals.py index 5e1fe66f..02d47745 100644 --- a/src/platform_plugin_aspects/tests/test_signals.py +++ b/src/platform_plugin_aspects/tests/test_signals.py @@ -7,6 +7,7 @@ from django.test import TestCase from platform_plugin_aspects.signals import ( + on_catalog_pathway_saved_txn, on_externalid_saved_txn, on_user_retirement, receive_course_publish, @@ -41,6 +42,17 @@ def test_on_externalid_saved(self, mock_transaction): mock_transaction.on_commit.assert_called_once() + @patch("platform_plugin_aspects.signals.transaction") + def test_on_catalog_pathway_saved(self, mock_transaction): + """ + Test that on_catalog_pathway_saved_txn defers the dump to on_commit. + """ + instance = Mock() + sender = Mock() + on_catalog_pathway_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): """ From c7d330da2dcd77c953dedeefc6b87cff328e3e18 Mon Sep 17 00:00:00 2001 From: Felipe Montoya Date: Wed, 28 Oct 2026 18:24:15 -0500 Subject: [PATCH 2/4] feat: add pathway_enrollment sink keyed by the learner's xAPI actor_id Co-Authored-By: Claude Opus 5.5 --- .../settings/common.py | 4 ++ src/platform_plugin_aspects/signals.py | 39 +++++++++++++++++++ src/platform_plugin_aspects/sinks/__init__.py | 1 + .../sinks/pathway_enrollment_sink.py | 20 ++++++++++ .../sinks/serializers.py | 31 ++++++++++++++- .../tests/test_pathway_enrollment_sink.py | 20 ++++++++++ .../sinks/tests/test_serializers.py | 20 ++++++++++ .../tests/test_signals.py | 12 ++++++ .../tests/test_utils.py | 22 +++++++++++ src/platform_plugin_aspects/utils.py | 14 +++++++ 10 files changed, 182 insertions(+), 1 deletion(-) create mode 100644 src/platform_plugin_aspects/sinks/pathway_enrollment_sink.py create mode 100644 src/platform_plugin_aspects/sinks/tests/test_pathway_enrollment_sink.py diff --git a/src/platform_plugin_aspects/settings/common.py b/src/platform_plugin_aspects/settings/common.py index c45d2980..37ba256a 100644 --- a/src/platform_plugin_aspects/settings/common.py +++ b/src/platform_plugin_aspects/settings/common.py @@ -107,6 +107,10 @@ def plugin_settings(settings): "module": "openedx_catalog.models", "model": "CatalogPathway", }, + "pathway_enrollment": { + "module": "openedx_catalog.models", + "model": "PathwayEnrollment", + }, } # 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 3ba5e928..ad18ffba 100644 --- a/src/platform_plugin_aspects/signals.py +++ b/src/platform_plugin_aspects/signals.py @@ -12,6 +12,7 @@ CourseEnrollmentSink, ExternalIdSink, ObjectTagSink, + PathwayEnrollmentSink, TagSink, TaxonomySink, UserProfileSink, @@ -342,3 +343,41 @@ def on_catalog_pathway_saved( # pylint: disable=unused-argument # pragma: no c _catalog_pathway = get_model("catalog_pathway") if _catalog_pathway: # pragma: no cover post_save.connect(on_catalog_pathway_saved_txn, sender=_catalog_pathway) + + +def on_pathway_enrollment_saved_txn(*args, **kwargs): + """ + Handle pathway_enrollment saves in the middle of a transaction. + + Handle saves in the middle of a transaction. + If this gets fired before the transaction commits, the task may try to + query an id that doesn't exist yet and throw an error. This should postpone + queuing the Celery task until after the transaction is committed. + """ + + def on_pathway_enrollment_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 = PathwayEnrollmentSink(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_pathway_enrollment_saved(*args, **kwargs)) + + +# Connect the PathwayEnrollment.post_save signal handler only if we have a model to attach to. +# (prevents celery errors during tests) +_pathway_enrollment = get_model("pathway_enrollment") +if _pathway_enrollment: # pragma: no cover + post_save.connect(on_pathway_enrollment_saved_txn, sender=_pathway_enrollment) diff --git a/src/platform_plugin_aspects/sinks/__init__.py b/src/platform_plugin_aspects/sinks/__init__.py index 747e012e..e8636599 100644 --- a/src/platform_plugin_aspects/sinks/__init__.py +++ b/src/platform_plugin_aspects/sinks/__init__.py @@ -7,6 +7,7 @@ from .course_enrollment_sink import CourseEnrollmentSink from .course_overview_sink import CourseOverviewSink, XBlockSink from .external_id_sink import ExternalIdSink +from .pathway_enrollment_sink import PathwayEnrollmentSink 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/pathway_enrollment_sink.py b/src/platform_plugin_aspects/sinks/pathway_enrollment_sink.py new file mode 100644 index 00000000..80c1250c --- /dev/null +++ b/src/platform_plugin_aspects/sinks/pathway_enrollment_sink.py @@ -0,0 +1,20 @@ +"""Pathway Enrollment sink""" + +from platform_plugin_aspects.sinks.base_sink import ModelBaseSink +from platform_plugin_aspects.sinks.serializers import PathwayEnrollmentSerializer + + +class PathwayEnrollmentSink(ModelBaseSink): # pylint: disable=abstract-method + """ + Sink for the openedx-core PathwayEnrollment model + """ + + model = "pathway_enrollment" + unique_key = "id" + clickhouse_table_name = "pathway_enrollment" + timestamp_field = "time_last_dumped" + name = "Pathway Enrollment" + serializer_class = PathwayEnrollmentSerializer + + def get_queryset(self, start_pk=None): + return super().get_queryset(start_pk).select_related("user") diff --git a/src/platform_plugin_aspects/sinks/serializers.py b/src/platform_plugin_aspects/sinks/serializers.py index d994ec91..c4ba97cc 100644 --- a/src/platform_plugin_aspects/sinks/serializers.py +++ b/src/platform_plugin_aspects/sinks/serializers.py @@ -8,7 +8,11 @@ from pytz import UTC from rest_framework import serializers -from platform_plugin_aspects.utils import get_model, get_tags_for_block +from platform_plugin_aspects.utils import ( + get_model, + get_tags_for_block, + get_xapi_actor_id, +) class DateTimeJSONEncoder(json.JSONEncoder): @@ -235,6 +239,31 @@ class Meta: ] +class PathwayEnrollmentSerializer(BaseSinkSerializer, serializers.ModelSerializer): + """Serializer for the PathwayEnrollment model.""" + + actor_id = serializers.SerializerMethodField() + + class Meta: + """Meta class for the PathwayEnrollmentSerializer""" + + model = get_model("pathway_enrollment") + fields = [ + "id", + "catalog_pathway_id", + "actor_id", + "created", + "modified", + "is_active", + "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_xapi_actor_id(instance.user) + + class TagSerializer(BaseSinkSerializer, serializers.ModelSerializer): """Serializer for the Tag model.""" diff --git a/src/platform_plugin_aspects/sinks/tests/test_pathway_enrollment_sink.py b/src/platform_plugin_aspects/sinks/tests/test_pathway_enrollment_sink.py new file mode 100644 index 00000000..a597be4f --- /dev/null +++ b/src/platform_plugin_aspects/sinks/tests/test_pathway_enrollment_sink.py @@ -0,0 +1,20 @@ +""" +Test the pathway enrollment sink module. +""" + +from unittest.mock import patch + +from platform_plugin_aspects.sinks import PathwayEnrollmentSink + + +@patch("platform_plugin_aspects.sinks.ModelBaseSink.get_queryset") +def test_get_queryset(mock_get_queryset): + """ + Test the get_queryset method. + """ + sink = PathwayEnrollmentSink(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") diff --git a/src/platform_plugin_aspects/sinks/tests/test_serializers.py b/src/platform_plugin_aspects/sinks/tests/test_serializers.py index 8e281e86..d2f58258 100644 --- a/src/platform_plugin_aspects/sinks/tests/test_serializers.py +++ b/src/platform_plugin_aspects/sinks/tests/test_serializers.py @@ -8,6 +8,7 @@ BaseSinkSerializer, CourseOverviewSerializer, DateTimeJSONEncoder, + PathwayEnrollmentSerializer, ) from test_utils.helpers import course_key_factory @@ -94,3 +95,22 @@ def test_get_course_key(self): mock_id = Mock() mock_overview = Mock(id=mock_id) self.assertEqual(self.serializer.get_course_key(mock_overview), str(mock_id)) + + +class TestPathwayEnrollmentSerializer(TestCase): + """ + Test PathwayEnrollmentSerializer + """ + + @patch("platform_plugin_aspects.sinks.serializers.get_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" + enrollment = Mock() + + actor_id = PathwayEnrollmentSerializer().get_actor_id(enrollment) + + assert actor_id == "29e5b22b-b457-49fa-9633-6c0123305d18" + mock_get_xapi_actor_id.assert_called_once_with(enrollment.user) diff --git a/src/platform_plugin_aspects/tests/test_signals.py b/src/platform_plugin_aspects/tests/test_signals.py index 02d47745..90e447b9 100644 --- a/src/platform_plugin_aspects/tests/test_signals.py +++ b/src/platform_plugin_aspects/tests/test_signals.py @@ -9,6 +9,7 @@ from platform_plugin_aspects.signals import ( on_catalog_pathway_saved_txn, on_externalid_saved_txn, + on_pathway_enrollment_saved_txn, on_user_retirement, receive_course_publish, ) @@ -53,6 +54,17 @@ def test_on_catalog_pathway_saved(self, mock_transaction): mock_transaction.on_commit.assert_called_once() + @patch("platform_plugin_aspects.signals.transaction") + def test_on_pathway_enrollment_saved(self, mock_transaction): + """ + Test that on_pathway_enrollment_saved_txn defers the dump to on_commit. + """ + instance = Mock() + sender = Mock() + on_pathway_enrollment_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): """ diff --git a/src/platform_plugin_aspects/tests/test_utils.py b/src/platform_plugin_aspects/tests/test_utils.py index 97fc6df8..74fb4c24 100644 --- a/src/platform_plugin_aspects/tests/test_utils.py +++ b/src/platform_plugin_aspects/tests/test_utils.py @@ -16,6 +16,7 @@ get_model, get_tags_for_block, get_user_dashboard_locale, + get_xapi_actor_id, ) from test_utils.helpers import course_factory @@ -292,3 +293,24 @@ def test_get_user_dashboard_locale(self, mock_get_model): user = Mock() assert get_user_dashboard_locale(user) == "en" mock_get_model.assert_called_once() + + @patch("platform_plugin_aspects.utils.get_model") + def test_get_xapi_actor_id(self, mock_get_model): + """Test that get_xapi_actor_id returns the user's xapi ExternalId.""" + external_id = Mock(external_user_id="29e5b22b-b457-49fa-9633-6c0123305d18") + mock_get_model.return_value.add_new_user_id.return_value = (external_id, True) + user = Mock() + + assert get_xapi_actor_id(user) == "29e5b22b-b457-49fa-9633-6c0123305d18" + mock_get_model.assert_called_once_with("external_id") + mock_get_model.return_value.add_new_user_id.assert_called_once_with( + user, "xapi" + ) + + @patch("platform_plugin_aspects.utils.get_model") + def test_get_xapi_actor_id_missing_type(self, mock_get_model): + """Test that get_xapi_actor_id fails loudly without the xapi ExternalIdType.""" + mock_get_model.return_value.add_new_user_id.return_value = (None, False) + + with self.assertRaises(ValueError): + get_xapi_actor_id(Mock()) diff --git a/src/platform_plugin_aspects/utils.py b/src/platform_plugin_aspects/utils.py index 272c5e29..46e040fb 100644 --- a/src/platform_plugin_aspects/utils.py +++ b/src/platform_plugin_aspects/utils.py @@ -311,6 +311,20 @@ def get_tags_for_block(usage_key) -> set: return list(serialized_tags) +def get_xapi_actor_id(user): + """ + Return the learner's xAPI actor id, creating it if the learner has none yet. + + This is the id event-routing-backends sends as the xAPI actor (the + learner's ExternalId of type "xapi"), so sink rows keyed by it join the + xAPI tables and hold no PII. + """ + external_id, _ = get_model("external_id").add_new_user_id(user, "xapi") + if not external_id: + raise ValueError("The 'xapi' ExternalIdType does not exist.") + return str(external_id.external_user_id) + + def get_user_dashboard_locale(user): """ From 44f10d2361c94f10a6fed419cefe38c345338738 Mon Sep 17 00:00:00 2001 From: Felipe Montoya Date: Thu, 29 Oct 2026 11:42:26 -0500 Subject: [PATCH 3/4] fix: addressing comments from quality revision using agents Co-Authored-By: Claude Opus 5.5 --- README.rst | 4 ++++ src/platform_plugin_aspects/signals.py | 22 +++++++++++++++++ .../sinks/serializers.py | 10 +++++--- .../sinks/tests/test_serializers.py | 10 ++++---- .../tests/test_signals.py | 10 ++++++++ .../tests/test_utils.py | 24 +++++++++++++------ src/platform_plugin_aspects/utils.py | 19 +++++++++++---- 7 files changed, 81 insertions(+), 18 deletions(-) diff --git a/README.rst b/README.rst index 4b451966..fd70407c 100644 --- a/README.rst +++ b/README.rst @@ -36,6 +36,10 @@ Below are the existing sink names and their corresponding object names (as neede model and stores the user profile data in ClickHouse. Object name: ``user_profile`` - ``CourseEnrollmentSink`` - Listen for the ``ENROLL_STATUS_CHANGE`` event and stores the course enrollment data. Object name: ``course_enrollment`` +- ``CatalogPathwaySink`` - Listens for the ``post_save`` Django signal on the openedx-core + ``CatalogPathway`` model. Object name: ``catalog_pathway`` +- ``PathwayEnrollmentSink`` - Listens for the ``post_save`` Django signal on the openedx-core + ``PathwayEnrollment`` model. Object name: ``pathway_enrollment`` - ``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/signals.py b/src/platform_plugin_aspects/signals.py index ad18ffba..12f43a09 100644 --- a/src/platform_plugin_aspects/signals.py +++ b/src/platform_plugin_aspects/signals.py @@ -2,6 +2,8 @@ Signal handler functions, mapped to specific signals in apps.py. """ +import logging + from django.db import transaction from django.db.models.signals import post_delete, post_save from django.dispatch import Signal, receiver @@ -26,6 +28,8 @@ # Tests don't have the platform installed USER_RETIRE_LMS_MISC = Signal() +logger = logging.getLogger(__name__) + def receive_course_publish( # pylint: disable=unused-argument # pragma: no cover sender, course_key, **kwargs @@ -307,6 +311,22 @@ def on_object_tag_deleted( # pylint: disable=unused-argument # pragma: no cove post_delete.connect(on_object_tag_deleted, sender=_object_tag) +def on_pathway_model_deleted(sender, instance, **kwargs): + """ + Receives post delete signal for CatalogPathway and PathwayEnrollment. + + TODO: how deletions reach ClickHouse is not yet decided. Deleting the rows there is not + wanted, so for now the deletion is only logged and the ClickHouse row is left as it was. + Raising here is not an option: PathwayEnrollment is CASCADE-deleted with its user, so an + exception would block user deletion. + """ + logger.warning( + "%s %s was deleted; deletions are not propagated to ClickHouse yet.", + sender.__name__, + instance.id, + ) + + def on_catalog_pathway_saved_txn(*args, **kwargs): """ Handle catalog_pathway saves in the middle of a transaction. @@ -343,6 +363,7 @@ def on_catalog_pathway_saved( # pylint: disable=unused-argument # pragma: no c _catalog_pathway = get_model("catalog_pathway") if _catalog_pathway: # pragma: no cover post_save.connect(on_catalog_pathway_saved_txn, sender=_catalog_pathway) + post_delete.connect(on_pathway_model_deleted, sender=_catalog_pathway) def on_pathway_enrollment_saved_txn(*args, **kwargs): @@ -381,3 +402,4 @@ 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) + post_delete.connect(on_pathway_model_deleted, sender=_pathway_enrollment) diff --git a/src/platform_plugin_aspects/sinks/serializers.py b/src/platform_plugin_aspects/sinks/serializers.py index c4ba97cc..e9d0db45 100644 --- a/src/platform_plugin_aspects/sinks/serializers.py +++ b/src/platform_plugin_aspects/sinks/serializers.py @@ -10,8 +10,8 @@ from platform_plugin_aspects.utils import ( get_model, + get_or_create_xapi_actor_id, get_tags_for_block, - get_xapi_actor_id, ) @@ -240,7 +240,11 @@ class Meta: class PathwayEnrollmentSerializer(BaseSinkSerializer, serializers.ModelSerializer): - """Serializer for the PathwayEnrollment model.""" + """ + Serializer for the PathwayEnrollment model. + + The learner is stored as their xAPI actor id instead of user_id to keep the table PII free. + """ actor_id = serializers.SerializerMethodField() @@ -261,7 +265,7 @@ class Meta: def get_actor_id(self, instance): """Return the learner's xAPI actor id instead of any user identifier.""" - return get_xapi_actor_id(instance.user) + return get_or_create_xapi_actor_id(instance.user) class TagSerializer(BaseSinkSerializer, serializers.ModelSerializer): diff --git a/src/platform_plugin_aspects/sinks/tests/test_serializers.py b/src/platform_plugin_aspects/sinks/tests/test_serializers.py index d2f58258..fa702852 100644 --- a/src/platform_plugin_aspects/sinks/tests/test_serializers.py +++ b/src/platform_plugin_aspects/sinks/tests/test_serializers.py @@ -102,15 +102,17 @@ class TestPathwayEnrollmentSerializer(TestCase): Test PathwayEnrollmentSerializer """ - @patch("platform_plugin_aspects.sinks.serializers.get_xapi_actor_id") - def test_get_actor_id(self, mock_get_xapi_actor_id): + @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_xapi_actor_id.return_value = "29e5b22b-b457-49fa-9633-6c0123305d18" + mock_get_or_create_xapi_actor_id.return_value = ( + "29e5b22b-b457-49fa-9633-6c0123305d18" + ) enrollment = Mock() actor_id = PathwayEnrollmentSerializer().get_actor_id(enrollment) assert actor_id == "29e5b22b-b457-49fa-9633-6c0123305d18" - mock_get_xapi_actor_id.assert_called_once_with(enrollment.user) + mock_get_or_create_xapi_actor_id.assert_called_once_with(enrollment.user) diff --git a/src/platform_plugin_aspects/tests/test_signals.py b/src/platform_plugin_aspects/tests/test_signals.py index 90e447b9..9dfd69cd 100644 --- a/src/platform_plugin_aspects/tests/test_signals.py +++ b/src/platform_plugin_aspects/tests/test_signals.py @@ -10,6 +10,7 @@ on_catalog_pathway_saved_txn, on_externalid_saved_txn, on_pathway_enrollment_saved_txn, + on_pathway_model_deleted, on_user_retirement, receive_course_publish, ) @@ -65,6 +66,15 @@ def test_on_pathway_enrollment_saved(self, mock_transaction): mock_transaction.on_commit.assert_called_once() + @patch("platform_plugin_aspects.signals.logger") + def test_on_pathway_model_deleted(self, mock_logger): + """ + Test that on_pathway_model_deleted only logs the deletion. + """ + on_pathway_model_deleted(Mock(__name__="PathwayEnrollment"), Mock(id=1)) + + mock_logger.warning.assert_called_once() + @patch("platform_plugin_aspects.tasks.dump_data_to_clickhouse") def test_on_user_retirement(self, mock_dump_task): """ diff --git a/src/platform_plugin_aspects/tests/test_utils.py b/src/platform_plugin_aspects/tests/test_utils.py index 74fb4c24..bed8e620 100644 --- a/src/platform_plugin_aspects/tests/test_utils.py +++ b/src/platform_plugin_aspects/tests/test_utils.py @@ -14,9 +14,9 @@ generate_superset_context, get_ccx_courses, get_model, + get_or_create_xapi_actor_id, get_tags_for_block, get_user_dashboard_locale, - get_xapi_actor_id, ) from test_utils.helpers import course_factory @@ -295,22 +295,32 @@ def test_get_user_dashboard_locale(self, mock_get_model): mock_get_model.assert_called_once() @patch("platform_plugin_aspects.utils.get_model") - def test_get_xapi_actor_id(self, mock_get_model): - """Test that get_xapi_actor_id returns the user's xapi ExternalId.""" + def test_get_or_create_xapi_actor_id(self, mock_get_model): + """Test that get_or_create_xapi_actor_id returns the user's xapi ExternalId.""" external_id = Mock(external_user_id="29e5b22b-b457-49fa-9633-6c0123305d18") mock_get_model.return_value.add_new_user_id.return_value = (external_id, True) user = Mock() - assert get_xapi_actor_id(user) == "29e5b22b-b457-49fa-9633-6c0123305d18" + assert ( + get_or_create_xapi_actor_id(user) == "29e5b22b-b457-49fa-9633-6c0123305d18" + ) mock_get_model.assert_called_once_with("external_id") mock_get_model.return_value.add_new_user_id.assert_called_once_with( user, "xapi" ) @patch("platform_plugin_aspects.utils.get_model") - def test_get_xapi_actor_id_missing_type(self, mock_get_model): - """Test that get_xapi_actor_id fails loudly without the xapi ExternalIdType.""" + def test_get_or_create_xapi_actor_id_missing_type(self, mock_get_model): + """Test that get_or_create_xapi_actor_id fails loudly without the xapi ExternalIdType.""" mock_get_model.return_value.add_new_user_id.return_value = (None, False) with self.assertRaises(ValueError): - get_xapi_actor_id(Mock()) + get_or_create_xapi_actor_id(Mock()) + + @patch("platform_plugin_aspects.utils.get_model") + def test_get_or_create_xapi_actor_id_missing_model(self, mock_get_model): + """Test that get_or_create_xapi_actor_id fails loudly without the ExternalId model.""" + mock_get_model.return_value = None + + with self.assertRaises(ValueError): + get_or_create_xapi_actor_id(Mock()) diff --git a/src/platform_plugin_aspects/utils.py b/src/platform_plugin_aspects/utils.py index 46e040fb..0783abe6 100644 --- a/src/platform_plugin_aspects/utils.py +++ b/src/platform_plugin_aspects/utils.py @@ -39,6 +39,9 @@ def _(text): DEFAULT_USER_LANGUAGE = "en" +# ExternalIdType name event-routing-backends uses for the xAPI actor id. +XAPI_EXTERNAL_ID_TYPE = "xapi" + def generate_superset_context( context, @@ -311,17 +314,25 @@ def get_tags_for_block(usage_key) -> set: return list(serialized_tags) -def get_xapi_actor_id(user): +def get_or_create_xapi_actor_id(user): """ Return the learner's xAPI actor id, creating it if the learner has none yet. This is the id event-routing-backends sends as the xAPI actor (the learner's ExternalId of type "xapi"), so sink rows keyed by it join the - xAPI tables and hold no PII. + xAPI tables and hold no PII. Creating the ExternalId fires its post_save, + which queues an ExternalIdSink dump. + + Raises ValueError if the ExternalId model or the "xapi" ExternalIdType is missing. """ - external_id, _ = get_model("external_id").add_new_user_id(user, "xapi") + ExternalId = get_model("external_id") + if not ExternalId: + raise ValueError("The ExternalId model could not be loaded.") + external_id, _ = ExternalId.add_new_user_id(user, XAPI_EXTERNAL_ID_TYPE) if not external_id: - raise ValueError("The 'xapi' ExternalIdType does not exist.") + raise ValueError( + f"The '{XAPI_EXTERNAL_ID_TYPE}' ExternalIdType does not exist." + ) return str(external_id.external_user_id) From 450277b8fb475d9421e4090fa775ce7b627f6dc6 Mon Sep 17 00:00:00 2001 From: Felipe Montoya Date: Thu, 29 Oct 2026 17:25:39 -0500 Subject: [PATCH 4/4] feat: add tombstone sink for hard-deleted Catalog Pathways Co-Authored-By: Claude Opus 5.5 --- README.rst | 2 + src/platform_plugin_aspects/signals.py | 41 +++++---- src/platform_plugin_aspects/sinks/__init__.py | 2 +- .../sinks/catalog_pathway_sink.py | 60 +++++++++++++ .../sinks/serializers.py | 11 +++ .../sinks/tests/test_catalog_pathway_sink.py | 88 ++++++++++++++++++- .../sinks/tests/test_serializers.py | 16 ++++ .../tests/test_signals.py | 27 ++++-- 8 files changed, 219 insertions(+), 28 deletions(-) diff --git a/README.rst b/README.rst index fd70407c..2fbc2430 100644 --- a/README.rst +++ b/README.rst @@ -38,6 +38,8 @@ Below are the existing sink names and their corresponding object names (as neede the course enrollment data. Object name: ``course_enrollment`` - ``CatalogPathwaySink`` - Listens for the ``post_save`` Django signal on the openedx-core ``CatalogPathway`` model. Object name: ``catalog_pathway`` +- ``CatalogPathwayTombstoneSink`` - Listens for the ``post_delete`` Django signal on the + ``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`` - ``UserRetirementSink`` - Listen for the ``USER_RETIRE_LMS_MISC`` Django signal and diff --git a/src/platform_plugin_aspects/signals.py b/src/platform_plugin_aspects/signals.py index 12f43a09..6fbe06f6 100644 --- a/src/platform_plugin_aspects/signals.py +++ b/src/platform_plugin_aspects/signals.py @@ -2,8 +2,6 @@ Signal handler functions, mapped to specific signals in apps.py. """ -import logging - from django.db import transaction from django.db.models.signals import post_delete, post_save from django.dispatch import Signal, receiver @@ -11,6 +9,7 @@ from platform_plugin_aspects.sinks import ( CatalogPathwaySink, + CatalogPathwayTombstoneSink, CourseEnrollmentSink, ExternalIdSink, ObjectTagSink, @@ -28,8 +27,6 @@ # Tests don't have the platform installed USER_RETIRE_LMS_MISC = Signal() -logger = logging.getLogger(__name__) - def receive_course_publish( # pylint: disable=unused-argument # pragma: no cover sender, course_key, **kwargs @@ -311,20 +308,29 @@ def on_object_tag_deleted( # pylint: disable=unused-argument # pragma: no cove post_delete.connect(on_object_tag_deleted, sender=_object_tag) -def on_pathway_model_deleted(sender, instance, **kwargs): +def on_catalog_pathway_deleted( # pylint: disable=unused-argument + sender, instance, **kwargs +): """ - Receives post delete signal for CatalogPathway and PathwayEnrollment. - - TODO: how deletions reach ClickHouse is not yet decided. Deleting the rows there is not - wanted, so for now the deletion is only logged and the ClickHouse row is left as it was. - Raising here is not an option: PathwayEnrollment is CASCADE-deleted with its user, so an - exception would block user deletion. + Receives post delete signal for CatalogPathway and queues a tombstone dump on commit. """ - logger.warning( - "%s %s was deleted; deletions are not propagated to ClickHouse yet.", - sender.__name__, - instance.id, - ) + # Django clears instance.pk after post_delete, so the id is read now, not in on_commit. + object_id = str(instance.id) + + def queue_tombstone(): + # 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 = CatalogPathwayTombstoneSink(None, None) + dump_data_to_clickhouse.delay( + sink_module=sink.__module__, + sink_name=sink.__class__.__name__, + object_id=object_id, + ) + + transaction.on_commit(queue_tombstone, robust=True) def on_catalog_pathway_saved_txn(*args, **kwargs): @@ -363,7 +369,7 @@ def on_catalog_pathway_saved( # pylint: disable=unused-argument # pragma: no c _catalog_pathway = get_model("catalog_pathway") if _catalog_pathway: # pragma: no cover post_save.connect(on_catalog_pathway_saved_txn, sender=_catalog_pathway) - post_delete.connect(on_pathway_model_deleted, sender=_catalog_pathway) + post_delete.connect(on_catalog_pathway_deleted, sender=_catalog_pathway) def on_pathway_enrollment_saved_txn(*args, **kwargs): @@ -402,4 +408,3 @@ 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) - post_delete.connect(on_pathway_model_deleted, sender=_pathway_enrollment) diff --git a/src/platform_plugin_aspects/sinks/__init__.py b/src/platform_plugin_aspects/sinks/__init__.py index e8636599..7f9f283f 100644 --- a/src/platform_plugin_aspects/sinks/__init__.py +++ b/src/platform_plugin_aspects/sinks/__init__.py @@ -3,7 +3,7 @@ """ from .base_sink import BaseSink, ModelBaseSink -from .catalog_pathway_sink import CatalogPathwaySink +from .catalog_pathway_sink import CatalogPathwaySink, CatalogPathwayTombstoneSink from .course_enrollment_sink import CourseEnrollmentSink from .course_overview_sink import CourseOverviewSink, XBlockSink from .external_id_sink import ExternalIdSink diff --git a/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py b/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py index 911f83b7..dccf6061 100644 --- a/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py +++ b/src/platform_plugin_aspects/sinks/catalog_pathway_sink.py @@ -1,5 +1,9 @@ """Catalog Pathway sink""" +import json + +import requests + from platform_plugin_aspects.sinks.base_sink import ModelBaseSink from platform_plugin_aspects.sinks.serializers import CatalogPathwaySerializer @@ -16,5 +20,61 @@ class CatalogPathwaySink(ModelBaseSink): # pylint: disable=abstract-method name = "Catalog Pathway" serializer_class = CatalogPathwaySerializer + # Unlike the other sink tables, time_last_dumped is DateTime64 here because it is the + # ReplacingMergeTree version column; the serializer writes it with a UTC offset, which + # only "best_effort" parses. + CLICKHOUSE_BULK_INSERT_PARAMS = { + **ModelBaseSink.CLICKHOUSE_BULK_INSERT_PARAMS, + "date_time_input_format": "best_effort", + } + def get_queryset(self, start_pk=None): return super().get_queryset(start_pk).select_related("org", "category") + + +# pylint: disable-next=abstract-method +class CatalogPathwayTombstoneSink(CatalogPathwaySink): + """ + Sink for deleted openedx-core CatalogPathways + + The pathway is gone from MySQL by the time this runs, so instead of serializing it, the + pathway's latest row in ClickHouse is copied with is_deleted set. A pathway that was never + dumped gets no row. + """ + + name = "Catalog Pathway Tombstone" + + def dump(self, item_id, many=False, initial=None): + """ + Write a tombstone row for the deleted pathway with id item_id. + """ + table = f"{self.ch_database}.{self.clickhouse_table_name}" + params = { + "query": ( + f"INSERT INTO {table} " + "SELECT * REPLACE (1 AS is_deleted, generateUUIDv4() AS dump_id, " + "now64(6) AS time_last_dumped) " + # Picked in a subquery: next to the REPLACE, ORDER BY time_last_dumped would + # resolve to the now64(6) alias and copy an arbitrary version. + f"FROM (SELECT * FROM {table} WHERE id = {int(item_id)} " + "ORDER BY time_last_dumped DESC LIMIT 1)" + ), + } + request = requests.Request( + "POST", + self.ch_url, + params=params, + auth=self.ch_auth, + ) + self.log.info(f"Now dumping {self.name} {item_id} to ClickHouse") + # Retried once: the pathway is gone from MySQL, so a lost tombstone can't be backfilled. + try: + response = self._send_clickhouse_request(request) + except requests.RequestException: + response = self._send_clickhouse_request(request) + summary = json.loads(response.headers.get("X-ClickHouse-Summary", "{}")) + if summary.get("written_rows") == "0": + self.log.warning( + f"No {self.name} written for {item_id}, it was never dumped" + ) + self.log.info(f"Completed dumping {self.name} {item_id} to ClickHouse") diff --git a/src/platform_plugin_aspects/sinks/serializers.py b/src/platform_plugin_aspects/sinks/serializers.py index e9d0db45..73caa0fa 100644 --- a/src/platform_plugin_aspects/sinks/serializers.py +++ b/src/platform_plugin_aspects/sinks/serializers.py @@ -220,6 +220,7 @@ class CatalogPathwaySerializer(BaseSinkSerializer, serializers.ModelSerializer): org = serializers.CharField(source="org.short_name") category_code = serializers.CharField(source="category.category_code") category_title = serializers.CharField(source="category.title") + is_deleted = serializers.SerializerMethodField() class Meta: """Meta class for the CatalogPathwaySerializer""" @@ -236,8 +237,18 @@ class Meta: "modified", "dump_id", "time_last_dumped", + "is_deleted", ] + def get_is_deleted(self, instance): # pylint: disable=unused-argument + """ + Saved pathways are never deleted; CatalogPathwayTombstoneSink writes deletions. + + 0 rather than False: the column is the UInt8 is_deleted of a ReplacingMergeTree, and + ClickHouse won't parse a CSV "False" as a UInt8. + """ + return 0 + class PathwayEnrollmentSerializer(BaseSinkSerializer, serializers.ModelSerializer): """ diff --git a/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py b/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py index 786dba40..8a13a147 100644 --- a/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py +++ b/src/platform_plugin_aspects/sinks/tests/test_catalog_pathway_sink.py @@ -2,9 +2,20 @@ Test the catalog pathway sink module. """ -from unittest.mock import patch +import logging +from unittest.mock import Mock, patch -from platform_plugin_aspects.sinks import CatalogPathwaySink +import pytest +import requests +import responses +from responses.registries import OrderedRegistry + +from platform_plugin_aspects.sinks import ( + CatalogPathwaySink, + CatalogPathwayTombstoneSink, +) + +log = logging.getLogger(__name__) @patch("platform_plugin_aspects.sinks.ModelBaseSink.get_queryset") @@ -20,3 +31,76 @@ def test_get_queryset(mock_get_queryset): mock_get_queryset.return_value.select_related.assert_called_once_with( "org", "category" ) + + +@responses.activate( # pylint: disable=unexpected-keyword-arg,no-value-for-parameter + registry=OrderedRegistry +) +def test_tombstone_dump(): + """ + Test that the tombstone sink copies the pathway's latest row with is_deleted set. + """ + tombstone = responses.post( + "https://foo.bar/", + match=[ + responses.matchers.query_param_matcher( + { + "query": "INSERT INTO cool_data.catalog_pathway " + "SELECT * REPLACE (1 AS is_deleted, generateUUIDv4() AS dump_id, " + "now64(6) AS time_last_dumped) " + "FROM (SELECT * FROM cool_data.catalog_pathway WHERE id = 7 " + "ORDER BY time_last_dumped DESC LIMIT 1)", + } + ) + ], + headers={"X-ClickHouse-Summary": '{"written_rows":"1"}'}, + ) + mock_log = Mock() + + CatalogPathwayTombstoneSink(None, mock_log).dump("7") + + assert tombstone.call_count == 1 + mock_log.warning.assert_not_called() + + +@responses.activate( # pylint: disable=unexpected-keyword-arg,no-value-for-parameter + registry=OrderedRegistry +) +def test_tombstone_dump_retries_once(): + """ + Test that a failed tombstone is retried once, and a second failure raises. + """ + responses.post("https://foo.bar/", body=requests.ConnectionError()) + responses.post("https://foo.bar/") + + CatalogPathwayTombstoneSink(None, log).dump("7") + + responses.post("https://foo.bar/", body=requests.ConnectionError()) + responses.post("https://foo.bar/", body=requests.ConnectionError()) + + with pytest.raises(requests.ConnectionError): + CatalogPathwayTombstoneSink(None, log).dump("7") + + +@responses.activate # pylint: disable=unexpected-keyword-arg,no-value-for-parameter +def test_tombstone_dump_warns_when_nothing_written(): + """ + Test that a pathway that was never dumped logs a warning. + """ + responses.post( + "https://foo.bar/", + headers={"X-ClickHouse-Summary": '{"written_rows":"0"}'}, + ) + mock_log = Mock() + + CatalogPathwayTombstoneSink(None, mock_log).dump("7") + + mock_log.warning.assert_called_once() + + +def test_tombstone_dump_rejects_non_integer_id(): + """ + Test that an id that isn't an integer never reaches the query. + """ + with pytest.raises(ValueError): + CatalogPathwayTombstoneSink(None, log).dump("1 OR 1=1") diff --git a/src/platform_plugin_aspects/sinks/tests/test_serializers.py b/src/platform_plugin_aspects/sinks/tests/test_serializers.py index fa702852..c34406a7 100644 --- a/src/platform_plugin_aspects/sinks/tests/test_serializers.py +++ b/src/platform_plugin_aspects/sinks/tests/test_serializers.py @@ -6,6 +6,7 @@ from platform_plugin_aspects.sinks.serializers import ( BaseSinkSerializer, + CatalogPathwaySerializer, CourseOverviewSerializer, DateTimeJSONEncoder, PathwayEnrollmentSerializer, @@ -97,6 +98,21 @@ def test_get_course_key(self): self.assertEqual(self.serializer.get_course_key(mock_overview), str(mock_id)) +class TestCatalogPathwaySerializer(TestCase): + """ + Test CatalogPathwaySerializer + """ + + def test_is_deleted(self): + """ + Test that saved pathways are written with is_deleted 0, in the last column. + """ + serializer = CatalogPathwaySerializer() + + self.assertEqual(serializer.get_is_deleted(Mock()), 0) + self.assertEqual(CatalogPathwaySerializer.Meta.fields[-1], "is_deleted") + + class TestPathwayEnrollmentSerializer(TestCase): """ Test PathwayEnrollmentSerializer diff --git a/src/platform_plugin_aspects/tests/test_signals.py b/src/platform_plugin_aspects/tests/test_signals.py index 9dfd69cd..9ef61b77 100644 --- a/src/platform_plugin_aspects/tests/test_signals.py +++ b/src/platform_plugin_aspects/tests/test_signals.py @@ -2,18 +2,19 @@ Tests for signal handlers. """ -from unittest.mock import Mock, patch +from unittest.mock import ANY, Mock, patch from django.test import TestCase from platform_plugin_aspects.signals import ( + on_catalog_pathway_deleted, on_catalog_pathway_saved_txn, on_externalid_saved_txn, on_pathway_enrollment_saved_txn, - on_pathway_model_deleted, on_user_retirement, receive_course_publish, ) +from platform_plugin_aspects.sinks import CatalogPathwayTombstoneSink from platform_plugin_aspects.sinks.user_retire_sink import UserRetirementSink @@ -66,14 +67,26 @@ def test_on_pathway_enrollment_saved(self, mock_transaction): mock_transaction.on_commit.assert_called_once() - @patch("platform_plugin_aspects.signals.logger") - def test_on_pathway_model_deleted(self, mock_logger): + @patch("platform_plugin_aspects.tasks.dump_data_to_clickhouse") + @patch("platform_plugin_aspects.signals.transaction") + def test_on_catalog_pathway_deleted(self, mock_transaction, mock_dump_task): """ - Test that on_pathway_model_deleted only logs the deletion. + Test that on_catalog_pathway_deleted queues the tombstone sink on commit, with the id + read before Django clears it. """ - on_pathway_model_deleted(Mock(__name__="PathwayEnrollment"), Mock(id=1)) + instance = Mock(id=7) + on_catalog_pathway_deleted(Mock(), instance) + instance.id = None + + mock_dump_task.delay.assert_not_called() + mock_transaction.on_commit.assert_called_once_with(ANY, robust=True) + mock_transaction.on_commit.call_args.args[0]() - mock_logger.warning.assert_called_once() + mock_dump_task.delay.assert_called_once_with( + sink_module=CatalogPathwayTombstoneSink.__module__, + sink_name="CatalogPathwayTombstoneSink", + object_id="7", + ) @patch("platform_plugin_aspects.tasks.dump_data_to_clickhouse") def test_on_user_retirement(self, mock_dump_task):