feat: bypass salesforce harness for Staging Integration

enterprise_access/apps/customer_billing/constants.py

@@ -153,3 +153,7 @@ class StripeSubscriptionStatus(StrEnum):
 
 CHECKOUT_LIFECYCLE_STATE_MONITORING_KEY = 'ssp_ci_lifecycle_change'
 CHECKOUT_LIFECYCLE_IS_ERROR_MONITORING_KEY = 'ssp_ci_lifecycle_is_error'
+
+# Feature flag to bypass Salesforce and directly trigger provisioning
+# from the invoice.paid webhook handler.
+BYPASS_SALESFORCE_PROVISIONING_FLAG = 'customer_billing.bypass_salesforce_for_provisioning'

enterprise_access/apps/customer_billing/stripe_event_handlers.py

@@ -7,6 +7,7 @@
 from uuid import UUID
 
 import stripe
+from django.conf import settings
 from django.utils import timezone
 from simple_history.utils import bulk_update_with_history
 
@@ -33,7 +34,9 @@
     send_trial_end_and_subscription_started_email_task,
     send_trial_ending_reminder_email_task
 )
+from enterprise_access.apps.customer_billing.toggles import bypass_salesforce_for_provisioning_enabled
 from enterprise_access.apps.customer_billing.utils import datetime_from_timestamp
+from enterprise_access.apps.provisioning.models import ProvisionNewCustomerWorkflow
 from enterprise_access.apps.track.segment import track_event
 
 logger = logging.getLogger(__name__)
@@ -381,6 +384,57 @@ def _valid_invoice_event_type(event: stripe.Event):
         return False
 
 
+def _bypass_salesforce_for_provisioning(checkout_intent: CheckoutIntent) -> None:
+    """
+    Directly trigger ``ProvisionNewCustomerWorkflow`` for the given ``checkout_intent``,
+    skipping the usual Salesforce-driven provisioning path.
+
+    Gated by the ``customer_billing.bypass_salesforce_for_provisioning`` waffle flag and
+    the ``settings.ALLOW_SALESFORCE_BYPASS`` hard guard. Intended only for end-to-end
+    testing in staging, where Salesforce is unavailable.
+    """
+    if not settings.ALLOW_SALESFORCE_BYPASS:
+        return
+    if not bypass_salesforce_for_provisioning_enabled():
+        return
+
+    logger.info(
+        'Bypassing Salesforce for checkout_intent uuid=%s: directly triggering provisioning workflow.',
+        checkout_intent.uuid,
+    )
+
+    workflow = None
+    try:
+        customer_request_dict = {
+            'name': checkout_intent.enterprise_name,
+            'slug': checkout_intent.enterprise_slug,
+            'country': checkout_intent.country,
+        }
+        admin_email_list = [checkout_intent.user.email]
+
+        workflow_input_dict = ProvisionNewCustomerWorkflow.generate_input_dict(
+            customer_request_dict,
+            admin_email_list,
+            None,
+            None,
+            None,
+            {},
+            {},
+            checkout_intent.ssp_product.slug,
+        )
+        workflow = ProvisionNewCustomerWorkflow.objects.create(input_data=workflow_input_dict)
+        workflow.execute()
+    except Exception as exc:  # pylint: disable=broad-except
+        logger.exception(
+            'Salesforce bypass provisioning failed for checkout_intent uuid=%s: %s',
+            checkout_intent.uuid, exc,
+        )
+        checkout_intent.mark_provisioning_error(str(exc), workflow=workflow)
+        return
+
+    checkout_intent.mark_as_fulfilled(workflow=workflow)
+
+
 def _handle_invoice_paid_status_updated(
     event: stripe.Event,
     checkout_intent: CheckoutIntent,
@@ -604,6 +658,9 @@ def invoice_paid(event: stripe.Event) -> None:
                 'Could not mark checkout intent %s as paid via invoice %s, because %s',
                 checkout_intent.uuid, invoice.id, exc,
             )
+            return
+
+        _bypass_salesforce_for_provisioning(checkout_intent)
 
     @on_stripe_event('invoice.created')
     @staticmethod

enterprise_access/apps/customer_billing/tests/test_stripe_event_handlers.py

@@ -11,7 +11,7 @@
 import ddt
 import stripe
 from django.contrib.auth.models import AbstractUser
-from django.test import TestCase
+from django.test import TestCase, override_settings
 from django.utils import timezone
 
 from enterprise_access.apps.core.tests.factories import UserFactory
@@ -48,6 +48,7 @@
     GetCreateTrialSubscriptionPlanStep
 )
 from enterprise_access.apps.provisioning.tests.factories import ProvisionNewCustomerWorkflowFactory
+from enterprise_access.apps.workflow.exceptions import UnitOfWorkException
 
 
 def _rand_numeric_string():
@@ -417,6 +418,169 @@ def test_invoice_paid_handles_salesforce_style_invoice_payload(self):
         self.assertEqual(event_data.summary.invoice_quantity, 6)
         self.assertEqual(event_data.summary.invoice_currency, 'usd')
 
+    def _trial_invoice_paid_event(self, stripe_customer_id):
+        """Build a trial-path (``invoice.total == 0``) invoice.paid event for ``self.checkout_intent``."""
+        self.checkout_intent.stripe_customer_id = stripe_customer_id
+        self.checkout_intent.save()
+
+        invoice_data = _build_salesforce_style_invoice_data(
+            invoice_id='in_bypass_test_001',
+            subscription_id='sub_bypass_test_001',
+            customer_id=stripe_customer_id,
+            customer_email=self.user.email,
+            customer_name='Test User',
+            checkout_intent_id=self.checkout_intent.id,
+            checkout_intent_uuid=self.checkout_intent.uuid,
+            enterprise_customer_slug='test-enterprise',
+            enterprise_customer_name='Test Enterprise',
+            lms_user_id=str(self.user.lms_user_id),
+            catalog_title='Open Courses',
+            catalog_query_id=30,
+            period_start=int(timezone.now().timestamp()),
+            total=0,
+            amount_paid=0,
+        )
+        return self._create_mock_stripe_event('invoice.paid', invoice_data)
+
+    @override_settings(ALLOW_SALESFORCE_BYPASS=True)
+    @mock.patch(
+        'enterprise_access.apps.customer_billing.stripe_event_handlers.bypass_salesforce_for_provisioning_enabled'
+    )
+    @mock.patch('enterprise_access.apps.provisioning.models.ProvisionNewCustomerWorkflow.objects.create')
+    @mock.patch('enterprise_access.apps.provisioning.models.ProvisionNewCustomerWorkflow.generate_input_dict')
+    def test_invoice_paid_bypasses_salesforce_when_enabled(
+        self,
+        mock_generate_input_dict,
+        mock_workflow_create,
+        mock_flag_enabled,
+    ):
+        """When the waffle flag and setting are both on, provisioning is triggered directly."""
+        mock_flag_enabled.return_value = True
+        mock_workflow = ProvisionNewCustomerWorkflowFactory.build()
+        mock_workflow.save()
+        mock_workflow.execute = mock.Mock()
+        mock_workflow_create.return_value = mock_workflow
+
+        mock_event = self._trial_invoice_paid_event('cus_bypass_enabled')
+
+        StripeEventHandler.dispatch(mock_event)
+
+        mock_generate_input_dict.assert_called_once()
+        call_args = mock_generate_input_dict.call_args.args
+        self.assertEqual(call_args[0]['slug'], self.checkout_intent.enterprise_slug)
+        self.assertEqual(call_args[1], [self.user.email])
+        self.assertEqual(call_args[-1], self.checkout_intent.ssp_product.slug)
+
+        mock_workflow_create.assert_called_once_with(input_data=mock_generate_input_dict.return_value)
+        mock_workflow.execute.assert_called_once()
+
+        self.checkout_intent.refresh_from_db()
+        self.assertEqual(self.checkout_intent.state, CheckoutIntentState.FULFILLED)
+        self.assertEqual(self.checkout_intent.workflow, mock_workflow)
+
+    @mock.patch(
+        'enterprise_access.apps.customer_billing.stripe_event_handlers.bypass_salesforce_for_provisioning_enabled'
+    )
+    @mock.patch('enterprise_access.apps.provisioning.models.ProvisionNewCustomerWorkflow.objects.create')
+    def test_invoice_paid_does_not_bypass_salesforce_when_setting_disabled(
+        self,
+        mock_workflow_create,
+        mock_flag_enabled,
+    ):
+        """Even with the waffle flag on, the hard settings guard must also be enabled."""
+        mock_flag_enabled.return_value = True
+
+        mock_event = self._trial_invoice_paid_event('cus_bypass_setting_off')
+
+        StripeEventHandler.dispatch(mock_event)
+
+        mock_workflow_create.assert_not_called()
+        self.checkout_intent.refresh_from_db()
+        self.assertEqual(self.checkout_intent.state, CheckoutIntentState.PAID)
+
+    @override_settings(ALLOW_SALESFORCE_BYPASS=True)
+    @mock.patch(
+        'enterprise_access.apps.customer_billing.stripe_event_handlers.bypass_salesforce_for_provisioning_enabled'
+    )
+    @mock.patch('enterprise_access.apps.provisioning.models.ProvisionNewCustomerWorkflow.objects.create')
+    def test_invoice_paid_does_not_bypass_salesforce_when_flag_disabled(
+        self,
+        mock_workflow_create,
+        mock_flag_enabled,
+    ):
+        """Even with the setting on, the waffle flag must also be enabled."""
+        mock_flag_enabled.return_value = False
+
+        mock_event = self._trial_invoice_paid_event('cus_bypass_flag_off')
+
+        StripeEventHandler.dispatch(mock_event)
+
+        mock_workflow_create.assert_not_called()
+        self.checkout_intent.refresh_from_db()
+        self.assertEqual(self.checkout_intent.state, CheckoutIntentState.PAID)
+
+    @mock.patch('enterprise_access.apps.provisioning.models.ProvisionNewCustomerWorkflow.objects.create')
+    def test_invoice_paid_does_not_bypass_salesforce_when_mark_as_paid_raises(self, mock_workflow_create):
+        """If ``mark_as_paid`` raises a ValueError, provisioning must not be triggered."""
+        self.checkout_intent.state = CheckoutIntentState.PAID
+        self.checkout_intent.stripe_customer_id = 'cus_original'
+        self.checkout_intent.save()
+
+        invoice_data = _build_salesforce_style_invoice_data(
+            invoice_id='in_bypass_mismatch_001',
+            subscription_id='sub_bypass_mismatch_001',
+            customer_id='cus_mismatched',
+            customer_email=self.user.email,
+            customer_name='Test User',
+            checkout_intent_id=self.checkout_intent.id,
+            checkout_intent_uuid=self.checkout_intent.uuid,
+            enterprise_customer_slug='test-enterprise',
+            enterprise_customer_name='Test Enterprise',
+            lms_user_id=str(self.user.lms_user_id),
+            catalog_title='Open Courses',
+            catalog_query_id=30,
+            period_start=int(timezone.now().timestamp()),
+            total=0,
+            amount_paid=0,
+        )
+        mock_event = self._create_mock_stripe_event('invoice.paid', invoice_data)
+
+        StripeEventHandler.dispatch(mock_event)
+
+        mock_workflow_create.assert_not_called()
+        self.checkout_intent.refresh_from_db()
+        self.assertEqual(self.checkout_intent.stripe_customer_id, 'cus_original')
+
+    @override_settings(ALLOW_SALESFORCE_BYPASS=True)
+    @mock.patch(
+        'enterprise_access.apps.customer_billing.stripe_event_handlers.bypass_salesforce_for_provisioning_enabled'
+    )
+    @mock.patch('enterprise_access.apps.provisioning.models.ProvisionNewCustomerWorkflow.objects.create')
+    @mock.patch('enterprise_access.apps.provisioning.models.ProvisionNewCustomerWorkflow.generate_input_dict')
+    def test_invoice_paid_bypass_marks_provisioning_error_on_workflow_failure(
+        self,
+        mock_generate_input_dict,
+        mock_workflow_create,
+        mock_flag_enabled,
+    ):
+        """If the bypassed workflow fails, the checkout intent is marked as errored, not fulfilled."""
+        mock_flag_enabled.return_value = True
+        mock_workflow = ProvisionNewCustomerWorkflowFactory.build()
+        mock_workflow.save()
+        mock_workflow.execute = mock.Mock()
+        mock_workflow_create.return_value = mock_workflow
+        mock_workflow.execute.side_effect = UnitOfWorkException('boom')
+
+        mock_event = self._trial_invoice_paid_event('cus_bypass_workflow_error')
+
+        StripeEventHandler.dispatch(mock_event)
+
+        mock_generate_input_dict.assert_called_once()
+        self.checkout_intent.refresh_from_db()
+        self.assertEqual(self.checkout_intent.state, CheckoutIntentState.ERRORED_PROVISIONING)
+        self.assertIn('boom', self.checkout_intent.last_provisioning_error)
+        self.assertEqual(self.checkout_intent.workflow, mock_workflow)
+
     @ddt.data(
         # Happy path: correct parent type at lines.data[0].parent.type
         {

enterprise_access/apps/customer_billing/tests/test_toggles.py

@@ -0,0 +1,20 @@
+"""Tests for customer_billing toggles."""
+from unittest import mock
+
+from django.test import TestCase
+
+from enterprise_access.apps.customer_billing.toggles import bypass_salesforce_for_provisioning_enabled
+
+
+class BypassSalesforceForProvisioningEnabledTests(TestCase):
+    """Tests for bypass_salesforce_for_provisioning_enabled."""
+
+    @mock.patch('enterprise_access.apps.customer_billing.toggles.BYPASS_SALESFORCE_FOR_PROVISIONING')
+    def test_returns_true_when_flag_enabled(self, mock_flag):
+        mock_flag.is_enabled.return_value = True
+        self.assertTrue(bypass_salesforce_for_provisioning_enabled())
+
+    @mock.patch('enterprise_access.apps.customer_billing.toggles.BYPASS_SALESFORCE_FOR_PROVISIONING')
+    def test_returns_false_when_flag_disabled(self, mock_flag):
+        mock_flag.is_enabled.return_value = False
+        self.assertFalse(bypass_salesforce_for_provisioning_enabled())

enterprise_access/apps/customer_billing/toggles.py

@@ -0,0 +1,28 @@
+"""Feature toggles for the customer_billing app."""
+
+from edx_toggles.toggles import WaffleFlag
+
+from enterprise_access.apps.customer_billing.constants import BYPASS_SALESFORCE_PROVISIONING_FLAG
+
+CUSTOMER_BILLING_LOG_PREFIX = '[customer_billing] '
+
+
+# .. toggle_name: customer_billing.bypass_salesforce_for_provisioning
+# .. toggle_implementation: WaffleFlag
+# .. toggle_default: False
+# .. toggle_description: When enabled (and settings.ALLOW_SALESFORCE_BYPASS is True), the
+#     invoice.paid Stripe webhook handler skips waiting for Salesforce and directly triggers
+#     ProvisionNewCustomerWorkflow. Intended for end-to-end testing in staging.
+# .. toggle_use_cases: temporary
+# .. toggle_creation_date: 2026-07-21
+# .. toggle_target_removal_date: 2026-10-21
+BYPASS_SALESFORCE_FOR_PROVISIONING = WaffleFlag(
+    BYPASS_SALESFORCE_PROVISIONING_FLAG,
+    __name__,
+    CUSTOMER_BILLING_LOG_PREFIX,
+)
+
+
+def bypass_salesforce_for_provisioning_enabled():
+    """Return whether the invoice.paid handler should bypass Salesforce and provision directly."""
+    return BYPASS_SALESFORCE_FOR_PROVISIONING.is_enabled()

enterprise_access/settings/base.py

@@ -543,6 +543,11 @@ def root(*path_fragments):
 # Budget deactivation settings
 ALLOW_BUDGET_DEACTIVATION_WITH_SPEND = False
 
+# Hard guard for the customer_billing.bypass_salesforce_for_provisioning waffle flag.
+# Must be explicitly enabled (e.g. in stage) in addition to the waffle flag before the
+# invoice.paid webhook handler will bypass Salesforce and directly trigger provisioning.
+ALLOW_SALESFORCE_BYPASS = False
+
 BRAZE_ASSIGNMENT_REMINDER_POST_LOGISTRATION_NOTIFICATION_CAMPAIGN = ''
 BRAZE_ASSIGNMENT_NUDGE_EXEC_ED_ACCEPTED_ASSIGNMENT_CAMPAIGN = ''
 BRAZE_ASSIGNMENT_CANCELLED_NOTIFICATION_CAMPAIGN = ''