feat: add mgmt command to replay transaction reversal events

This will allow a consumer in enterprise-access to update
related assignment record state for reversals that occurred
before the producer/consumer logic was running to synchronize
reversed state between ledger transactions and assignment records.
ENT-9212

enterprise_subsidy/apps/core/management/commands/produce_enterprise_ping_event.py

@@ -81,7 +81,6 @@ def add_arguments(self, parser):
     def handle(self, *args, **options):
         try:
             producer = create_producer()
-            # breakpoint()
             producer.send(
                 signal=ENTERPRISE_PING_SIGNAL,
                 topic=ENTERPRISE_CORE_TOPIC,

enterprise_subsidy/apps/subsidy/rules.py

@@ -88,18 +88,15 @@ def has_explicit_access_to_subsidy_learner(user, context):
 
 # Now, recombine the implicit and explicit rules for a given feature role using composition.  Also, waterfall the rules
 # by defining access levels which give "higher" levels access to their own level, as well as everything below.
-# pylint: disable=unsupported-binary-operation
 has_learner_level_access = (
     has_implicit_access_to_subsidy_operator | has_explicit_access_to_subsidy_operator |
     has_implicit_access_to_subsidy_admin | has_explicit_access_to_subsidy_admin |
     has_implicit_access_to_subsidy_learner | has_explicit_access_to_subsidy_learner
 )
-# pylint: disable=unsupported-binary-operation
 has_admin_level_access = (
     has_implicit_access_to_subsidy_operator | has_explicit_access_to_subsidy_operator |
     has_implicit_access_to_subsidy_admin | has_explicit_access_to_subsidy_admin
 )
-# pylint: disable=unsupported-binary-operation
 has_operator_level_access = has_implicit_access_to_subsidy_operator | has_explicit_access_to_subsidy_operator
 
 

enterprise_subsidy/apps/transaction/management/commands/replay_reversal_events.py

@@ -0,0 +1,60 @@
+"""
+Management command to replay openedx transaction reversal events for all
+currently-reversed ledger transactions.
+"""
+import logging
+
+from django.core.management.base import BaseCommand
+from openedx_events.event_bus import get_producer
+from openedx_ledger.models import Transaction, TransactionStateChoices
+
+from enterprise_subsidy.apps.core.event_bus import send_transaction_reversed_event
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Management command for writing Transaction Reversals from recent Enterprise unenrollments data.
+
+    ./manage.py write_reversals_from_enterprise_unenrollments
+    """
+
+    def add_arguments(self, parser):
+        """
+        Entry point for subclassed commands to add custom arguments.
+        """
+        parser.add_argument(
+            '--dry-run',
+            action='store_true',
+            dest='dry_run',
+            help=(
+                'If set, will only log a message about what events we would have produced,'
+                'without actually producing them.'
+            ),
+        )
+
+    def handle(self, *args, **options):
+        """
+        Finds all reversed transactions and emits a reversal event for each.
+        """
+        all_reversed_transactions = Transaction.objects.select_related('reversal').filter(
+            state=TransactionStateChoices.COMMITTED,
+            reversal__isnull=False,
+            reversal__state=TransactionStateChoices.COMMITTED,
+        )
+
+        for transaction_record in all_reversed_transactions:
+            if not options.get('dry_run'):
+                send_transaction_reversed_event(transaction_record)
+                logger.info(f'Sent reversal event for transaction {transaction_record.uuid}')
+            else:
+                logger.info(f'[DRY RUN] Would have sent reversal event for transaction {transaction_record.uuid}')
+
+        if not options.get('dry_run'):
+            # Retrieve the cached producer and tell it to prepare for shutdown before this command exits.
+            # This ensures that all messages in the send queue are flushed. Without this, this command
+            # will exit and drop all produced messages before they can be sent to the broker.
+            # See: https://github.com/openedx/event-bus-kafka/blob/main/edx_event_bus_kafka/internal/producer.py#L324
+            # and https://github.com/openedx/event-bus-kafka/blob/main/docs/decisions/0007-producer-polling.rst
+            get_producer().prepare_for_shutdown()

enterprise_subsidy/apps/transaction/tests/test_management.py

@@ -1086,3 +1086,47 @@ def test_backpopulate_transaction_parent_content_key_include_internal(
         assert self.transaction_to_backpopulate.parent_content_key == expected_parent_content_key
         assert self.internal_transaction_to_backpopulate.parent_content_key == expected_parent_content_key
         assert self.transaction_not_to_backpopulate.parent_content_key is None
+
+
+@mark.django_db
+@ddt.ddt
+class TestReplayReversalMgmtCommand(TestCase):
+    """
+    Test the replay_reversal_events mgmt command.
+    """
+    MOCK_PATH_PREFIX = 'enterprise_subsidy.apps.transaction.management.commands.replay_reversal_events'
+
+    def setUp(self):
+        super().setUp()
+        self.ledger = LedgerFactory()
+        self.transaction_a = TransactionFactory(ledger=self.ledger, quantity=100)
+        ReversalFactory(
+            transaction=self.transaction_a, idempotency_key=f'unenrollment-reversal-{self.transaction_a.uuid}',
+        )
+
+        self.transaction_b = TransactionFactory(ledger=self.ledger, quantity=200)
+        ReversalFactory(
+            transaction=self.transaction_b, idempotency_key=f'unenrollment-reversal-{self.transaction_b.uuid}',
+        )
+
+        # one un-reversed transaction
+        self.transaction_c = TransactionFactory(ledger=self.ledger, quantity=200)
+
+    @mock.patch(f'{MOCK_PATH_PREFIX}.send_transaction_reversed_event')
+    def test_command_dry_run(self, mock_send_event):
+        """
+        Test that no events are actually produced during a dry run.
+        """
+        call_command('replay_reversal_events', dry_run=True)
+        self.assertFalse(mock_send_event.called)
+
+    @mock.patch(f'{MOCK_PATH_PREFIX}.send_transaction_reversed_event')
+    def test_command_sends_events(self, mock_send_event):
+        """
+        Test that the command produces events for all reversed transactions.
+        """
+        call_command('replay_reversal_events', dry_run=False)
+        mock_send_event.assert_has_calls([
+            mock.call(self.transaction_a),
+            mock.call(self.transaction_b),
+        ], any_order=True)