Added a _release Custom Django Action to SubscriptionSchedule

This would allow a User to use the Admin to select one or more than 1 SubscriptionSchedules and "Release" them, much in the same way they can choose to Cancel SubscriptionSchedules.

djstripe/admin.py

@@ -875,6 +875,22 @@ class SubscriptionScheduleAdmin(StripeModelAdmin):
     list_filter = ("status", "subscription", "customer")
     list_select_related = ("customer", "customer__subscriber", "subscription")
 
+    @admin.display(description="Release Selected Subscription Schedules")
+    def _release_subscription_schedule(self, request, queryset):
+        """Release a SubscriptionSchedule."""
+        context = self.get_admin_action_context(
+            queryset, "_release_subscription_schedule", CustomActionForm
+        )
+        return render(request, "djstripe/admin/confirm_action.html", context)
+
+    def get_actions(self, request):
+        # get all actions
+        actions = super().get_actions(request)
+        actions["_release_subscription_schedule"] = self.get_action(
+            "_release_subscription_schedule"
+        )
+        return actions
+
 
 @admin.register(models.TaxRate)
 class TaxRateAdmin(StripeModelAdmin):

djstripe/views.py

@@ -161,3 +161,12 @@ def _cancel(self, request, queryset):
                 messages.success(request, f"Successfully Canceled: {instance}")
             except stripe.error.InvalidRequestError as error:
                 messages.warning(request, error)
+
+    def _release_subscription_schedule(self, request, queryset):
+        """Release a SubscriptionSchedule."""
+        for subscription_schedule in queryset:
+            try:
+                instance = subscription_schedule.release()
+                messages.success(request, f"Successfully Released: {instance}")
+            except stripe.error.InvalidRequestError as error:
+                messages.warning(request, error)

tests/test_admin.py

@@ -30,6 +30,7 @@
     FAKE_PLAN,
     FAKE_PRODUCT,
     FAKE_SUBSCRIPTION,
+    FAKE_SUBSCRIPTION_SCHEDULE,
 )
 
 from .fields.models import TestCustomActionModel
@@ -1167,3 +1168,76 @@ def mock_plan_get(*args, **kwargs):
 
         # assert user got 200 status code
         assert response.status_code == 200
+
+
+class TestSubscriptionScheduleAdminCustomAction:
+    def test__release_subscription_schedule(
+        self,
+        admin_client,
+        monkeypatch,
+    ):
+        def mock_balance_transaction_get(*args, **kwargs):
+            return FAKE_BALANCE_TRANSACTION
+
+        def mock_subscription_get(*args, **kwargs):
+            return FAKE_SUBSCRIPTION
+
+        def mock_charge_get(*args, **kwargs):
+            return FAKE_CHARGE
+
+        def mock_payment_method_get(*args, **kwargs):
+            return FAKE_CARD_AS_PAYMENT_METHOD
+
+        def mock_payment_intent_get(*args, **kwargs):
+            return FAKE_PAYMENT_INTENT_I
+
+        def mock_product_get(*args, **kwargs):
+            return FAKE_PRODUCT
+
+        def mock_invoice_get(*args, **kwargs):
+            return FAKE_INVOICE
+
+        def mock_customer_get(*args, **kwargs):
+            return FAKE_CUSTOMER
+
+        def mock_plan_get(*args, **kwargs):
+            return FAKE_PLAN
+
+        # monkeypatch stripe retrieve calls to return
+        # the desired json response.
+        monkeypatch.setattr(
+            stripe.BalanceTransaction, "retrieve", mock_balance_transaction_get
+        )
+        monkeypatch.setattr(stripe.Subscription, "retrieve", mock_subscription_get)
+        monkeypatch.setattr(stripe.Charge, "retrieve", mock_charge_get)
+
+        monkeypatch.setattr(stripe.PaymentMethod, "retrieve", mock_payment_method_get)
+        monkeypatch.setattr(stripe.PaymentIntent, "retrieve", mock_payment_intent_get)
+        monkeypatch.setattr(stripe.Product, "retrieve", mock_product_get)
+
+        monkeypatch.setattr(stripe.Invoice, "retrieve", mock_invoice_get)
+        monkeypatch.setattr(stripe.Customer, "retrieve", mock_customer_get)
+
+        monkeypatch.setattr(stripe.Plan, "retrieve", mock_plan_get)
+
+        # create latest invoice
+        models.Invoice.sync_from_stripe_data(deepcopy(FAKE_INVOICE))
+
+        model = models.SubscriptionSchedule
+        subscription_schedule_fake = deepcopy(FAKE_SUBSCRIPTION_SCHEDULE)
+        instance = model.sync_from_stripe_data(subscription_schedule_fake)
+
+        # get the standard changelist_view url
+        change_url = reverse(
+            f"admin:{model._meta.app_label}_{model.__name__.lower()}_changelist"
+        )
+
+        data = {
+            "action": "_release_subscription_schedule",
+            helpers.ACTION_CHECKBOX_NAME: [instance.pk],
+        }
+
+        response = admin_client.post(change_url, data)
+
+        # assert user got 200 status code
+        assert response.status_code == 200

tests/test_views.py

@@ -27,6 +27,7 @@
     FAKE_PLAN,
     FAKE_PRODUCT,
     FAKE_SUBSCRIPTION,
+    FAKE_SUBSCRIPTION_SCHEDULE,
 )
 
 from .fields.models import TestCustomActionModel
@@ -50,7 +51,13 @@ def dummy_get_response(self, request):
         return None
 
     @pytest.mark.parametrize(
-        "action_name", ["_resync_instances", "_sync_all_instances", "_cancel"]
+        "action_name",
+        [
+            "_resync_instances",
+            "_sync_all_instances",
+            "_cancel",
+            "_release_subscription_schedule",
+        ],
     )
     def test_get_form_kwargs(self, action_name, admin_user, monkeypatch):
 
@@ -87,7 +94,13 @@ def mock_get_model(*args, **kwargs):
         assert form_kwargs.get("action_name") == action_name
 
     @pytest.mark.parametrize(
-        "action_name", ["_resync_instances", "_sync_all_instances", "_cancel"]
+        "action_name",
+        [
+            "_resync_instances",
+            "_sync_all_instances",
+            "_cancel",
+            "_release_subscription_schedule",
+        ],
     )
     @pytest.mark.parametrize("is_admin_user", [True, False])
     def test_dispatch(self, is_admin_user, action_name, admin_user, monkeypatch):
@@ -137,7 +150,13 @@ def mock_get_model(*args, **kwargs):
             )
 
     @pytest.mark.parametrize(
-        "action_name", ["_resync_instances", "_sync_all_instances", "_cancel"]
+        "action_name",
+        [
+            "_resync_instances",
+            "_sync_all_instances",
+            "_cancel",
+            "_release_subscription_schedule",
+        ],
     )
     @pytest.mark.parametrize("djstripe_owner_account_exists", [False, True])
     def test_form_valid(self, djstripe_owner_account_exists, action_name, monkeypatch):
@@ -202,7 +221,13 @@ def mock_request_handler(*args, **kwargs):
         assert response.url == "/admin/fields/testcustomactionmodel/"
 
     @pytest.mark.parametrize(
-        "action_name", ["_resync_instances", "_sync_all_instances", "_cancel"]
+        "action_name",
+        [
+            "_resync_instances",
+            "_sync_all_instances",
+            "_cancel",
+            "_release_subscription_schedule",
+        ],
     )
     @pytest.mark.parametrize("djstripe_owner_account_exists", [False, True])
     def test_form_invalid(
@@ -663,3 +688,187 @@ def mock_subscription_cancel(*args, **keywargs):
         with pytest.warns(None, match=r"some random error message"):
             # Invoke the Custom Actions
             view._cancel(request, [instance])
+
+    def test__release_subscription_schedule(  # noqa: C901
+        self,
+        monkeypatch,
+    ):
+        def mock_balance_transaction_get(*args, **kwargs):
+            return FAKE_BALANCE_TRANSACTION
+
+        def mock_subscription_get(*args, **kwargs):
+            return FAKE_SUBSCRIPTION
+
+        def mock_charge_get(*args, **kwargs):
+            return FAKE_CHARGE
+
+        def mock_payment_method_get(*args, **kwargs):
+            return FAKE_CARD_AS_PAYMENT_METHOD
+
+        def mock_payment_intent_get(*args, **kwargs):
+            return FAKE_PAYMENT_INTENT_I
+
+        def mock_product_get(*args, **kwargs):
+            return FAKE_PRODUCT
+
+        def mock_invoice_get(*args, **kwargs):
+            return FAKE_INVOICE
+
+        def mock_customer_get(*args, **kwargs):
+            return FAKE_CUSTOMER
+
+        def mock_plan_get(*args, **kwargs):
+            return FAKE_PLAN
+
+        # monkeypatch stripe retrieve calls to return
+        # the desired json response.
+        monkeypatch.setattr(
+            stripe.BalanceTransaction, "retrieve", mock_balance_transaction_get
+        )
+        monkeypatch.setattr(stripe.Subscription, "retrieve", mock_subscription_get)
+        monkeypatch.setattr(stripe.Charge, "retrieve", mock_charge_get)
+
+        monkeypatch.setattr(stripe.PaymentMethod, "retrieve", mock_payment_method_get)
+        monkeypatch.setattr(stripe.PaymentIntent, "retrieve", mock_payment_intent_get)
+        monkeypatch.setattr(stripe.Product, "retrieve", mock_product_get)
+
+        monkeypatch.setattr(stripe.Invoice, "retrieve", mock_invoice_get)
+        monkeypatch.setattr(stripe.Customer, "retrieve", mock_customer_get)
+
+        monkeypatch.setattr(stripe.Plan, "retrieve", mock_plan_get)
+
+        # create latest invoice
+        models.Invoice.sync_from_stripe_data(deepcopy(FAKE_INVOICE))
+
+        model = models.SubscriptionSchedule
+        subscription_schedule_fake = deepcopy(FAKE_SUBSCRIPTION_SCHEDULE)
+        instance = model.sync_from_stripe_data(subscription_schedule_fake)
+
+        # monkeypatch subscription_schedule.release()
+        def mock_subscription_schedule_release(*args, **keywargs):
+            return instance
+
+        monkeypatch.setattr(instance, "release", mock_subscription_schedule_release)
+
+        data = {
+            "action": "_release_subscription_schedule",
+            helpers.ACTION_CHECKBOX_NAME: [instance.pk],
+        }
+
+        kwargs = {
+            "action_name": "_release_subscription_schedule",
+            "model_name": model.__name__.lower(),
+        }
+
+        # get the custom action POST url
+        change_url = reverse("djstripe:djstripe_custom_action", kwargs=kwargs)
+
+        request = RequestFactory().post(change_url, data=data, follow=True)
+
+        # Add the session/message middleware to the request
+        SessionMiddleware(self.dummy_get_response).process_request(request)
+        MessageMiddleware(self.dummy_get_response).process_request(request)
+
+        view = ConfirmCustomAction()
+        view.setup(request, **kwargs)
+
+        # Invoke the Custom Actions
+        view._release_subscription_schedule(request, [instance])
+
+        # assert correct Success messages are emmitted
+        messages_sent_dictionary = {
+            m.message: m.level_tag for m in messages.get_messages(request)
+        }
+
+        # assert correct success message was emmitted
+        assert (
+            messages_sent_dictionary.get(f"Successfully Released: {instance}")
+            == "success"
+        )
+
+    def test__release_subscription_schedule_stripe_invalid_request_error(  # noqa: C901
+        self,
+        monkeypatch,
+    ):
+        def mock_balance_transaction_get(*args, **kwargs):
+            return FAKE_BALANCE_TRANSACTION
+
+        def mock_subscription_get(*args, **kwargs):
+            return FAKE_SUBSCRIPTION
+
+        def mock_charge_get(*args, **kwargs):
+            return FAKE_CHARGE
+
+        def mock_payment_method_get(*args, **kwargs):
+            return FAKE_CARD_AS_PAYMENT_METHOD
+
+        def mock_payment_intent_get(*args, **kwargs):
+            return FAKE_PAYMENT_INTENT_I
+
+        def mock_product_get(*args, **kwargs):
+            return FAKE_PRODUCT
+
+        def mock_invoice_get(*args, **kwargs):
+            return FAKE_INVOICE
+
+        def mock_customer_get(*args, **kwargs):
+            return FAKE_CUSTOMER
+
+        def mock_plan_get(*args, **kwargs):
+            return FAKE_PLAN
+
+        # monkeypatch stripe retrieve calls to return
+        # the desired json response.
+        monkeypatch.setattr(
+            stripe.BalanceTransaction, "retrieve", mock_balance_transaction_get
+        )
+        monkeypatch.setattr(stripe.Subscription, "retrieve", mock_subscription_get)
+        monkeypatch.setattr(stripe.Charge, "retrieve", mock_charge_get)
+
+        monkeypatch.setattr(stripe.PaymentMethod, "retrieve", mock_payment_method_get)
+        monkeypatch.setattr(stripe.PaymentIntent, "retrieve", mock_payment_intent_get)
+        monkeypatch.setattr(stripe.Product, "retrieve", mock_product_get)
+
+        monkeypatch.setattr(stripe.Invoice, "retrieve", mock_invoice_get)
+        monkeypatch.setattr(stripe.Customer, "retrieve", mock_customer_get)
+
+        monkeypatch.setattr(stripe.Plan, "retrieve", mock_plan_get)
+
+        # create latest invoice
+        models.Invoice.sync_from_stripe_data(deepcopy(FAKE_INVOICE))
+
+        model = models.SubscriptionSchedule
+        subscription_schedule_fake = deepcopy(FAKE_SUBSCRIPTION_SCHEDULE)
+        instance = model.sync_from_stripe_data(subscription_schedule_fake)
+
+        # monkeypatch subscription_schedule.release()
+        def mock_subscription_schedule_release(*args, **keywargs):
+            raise stripe.error.InvalidRequestError({}, "some random error message")
+
+        monkeypatch.setattr(instance, "release", mock_subscription_schedule_release)
+
+        data = {
+            "action": "_release_subscription_schedule",
+            helpers.ACTION_CHECKBOX_NAME: [instance.pk],
+        }
+
+        kwargs = {
+            "action_name": "_release_subscription_schedule",
+            "model_name": model.__name__.lower(),
+        }
+
+        # get the custom action POST url
+        change_url = reverse("djstripe:djstripe_custom_action", kwargs=kwargs)
+
+        request = RequestFactory().post(change_url, data=data, follow=True)
+
+        # Add the session/message middleware to the request
+        SessionMiddleware(self.dummy_get_response).process_request(request)
+        MessageMiddleware(self.dummy_get_response).process_request(request)
+
+        view = ConfirmCustomAction()
+        view.setup(request, **kwargs)
+
+        with pytest.warns(None, match=r"some random error message"):
+            # Invoke the Custom Actions
+            view._release_subscription_schedule(request, [instance])