Send email to admins on failed payment

Author: spalmurray

billing/tests/test_views.py

@@ -1,5 +1,6 @@
 import time
-from unittest.mock import patch
+from datetime import datetime
+from unittest.mock import call, patch
 
 import stripe
 from django.conf import settings
@@ -142,6 +143,11 @@ def test_invoice_payment_failed_sets_owner_delinquent_true(self):
                     "object": {
                         "customer": self.owner.stripe_customer_id,
                         "subscription": self.owner.stripe_subscription_id,
+                        "default_payment_method": {
+                            "card": {"brand": "visa", "last4": 1234}
+                        },
+                        "total": 24000,
+                        "hosted_invoice_url": "https://stripe.com",
                     }
                 },
             }
@@ -165,6 +171,11 @@ def test_invoice_payment_failed_sets_multiple_owners_delinquent_true(self):
                     "object": {
                         "customer": self.owner.stripe_customer_id,
                         "subscription": self.owner.stripe_subscription_id,
+                        "default_payment_method": {
+                            "card": {"brand": "visa", "last4": 1234}
+                        },
+                        "total": 24000,
+                        "hosted_invoice_url": "https://stripe.com",
                     }
                 },
             }
@@ -176,6 +187,118 @@ def test_invoice_payment_failed_sets_multiple_owners_delinquent_true(self):
         assert self.owner.delinquent is True
         assert self.other_owner.delinquent is True
 
+    @patch("services.task.TaskService.send_email")
+    def test_invoice_payment_failed_sends_email_to_admins(self, mocked_send_email):
+        non_admin = OwnerFactory(email="[email protected]")
+        admin_1 = OwnerFactory(email="[email protected]")
+        admin_2 = OwnerFactory(email="[email protected]")
+        self.owner.admins = [admin_1.ownerid, admin_2.ownerid]
+        self.owner.plan_activated_users = [non_admin.ownerid]
+        self.owner.save()
+
+        response = self._send_event(
+            payload={
+                "type": "invoice.payment_failed",
+                "data": {
+                    "object": {
+                        "customer": self.owner.stripe_customer_id,
+                        "subscription": self.owner.stripe_subscription_id,
+                        "default_payment_method": {
+                            "card": {"brand": "visa", "last4": 1234}
+                        },
+                        "total": 24000,
+                        "hosted_invoice_url": "https://stripe.com",
+                    }
+                },
+            }
+        )
+
+        self.owner.refresh_from_db()
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert self.owner.delinquent is True
+
+        expected_calls = [
+            call(
+                to_addr=admin_1.email,
+                subject="Your Codecov payment failed",
+                template_name="failed-payment",
+                name=admin_1.username,
+                amount=240,
+                card_type="visa",
+                last_four=1234,
+                cta_link="https://stripe.com",
+                date=datetime.now().strftime("%B %-d, %Y"),
+            ),
+            call(
+                to_addr=admin_2.email,
+                subject="Your Codecov payment failed",
+                template_name="failed-payment",
+                name=admin_2.username,
+                amount=240,
+                card_type="visa",
+                last_four=1234,
+                cta_link="https://stripe.com",
+                date=datetime.now().strftime("%B %-d, %Y"),
+            ),
+        ]
+        mocked_send_email.assert_has_calls(expected_calls)
+
+    @patch("services.task.TaskService.send_email")
+    def test_invoice_payment_failed_sends_email_to_admins_no_card(
+        self, mocked_send_email
+    ):
+        non_admin = OwnerFactory(email="[email protected]")
+        admin_1 = OwnerFactory(email="[email protected]")
+        admin_2 = OwnerFactory(email="[email protected]")
+        self.owner.admins = [admin_1.ownerid, admin_2.ownerid]
+        self.owner.plan_activated_users = [non_admin.ownerid]
+        self.owner.save()
+
+        response = self._send_event(
+            payload={
+                "type": "invoice.payment_failed",
+                "data": {
+                    "object": {
+                        "customer": self.owner.stripe_customer_id,
+                        "subscription": self.owner.stripe_subscription_id,
+                        "default_payment_method": None,
+                        "total": 24000,
+                        "hosted_invoice_url": "https://stripe.com",
+                    }
+                },
+            }
+        )
+
+        self.owner.refresh_from_db()
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert self.owner.delinquent is True
+
+        expected_calls = [
+            call(
+                to_addr=admin_1.email,
+                subject="Your Codecov payment failed",
+                template_name="failed-payment",
+                name=admin_1.username,
+                amount=240,
+                card_type=None,
+                last_four=None,
+                cta_link="https://stripe.com",
+                date=datetime.now().strftime("%B %-d, %Y"),
+            ),
+            call(
+                to_addr=admin_2.email,
+                subject="Your Codecov payment failed",
+                template_name="failed-payment",
+                name=admin_2.username,
+                amount=240,
+                card_type=None,
+                last_four=None,
+                cta_link="https://stripe.com",
+                date=datetime.now().strftime("%B %-d, %Y"),
+            ),
+        ]
+        mocked_send_email.assert_has_calls(expected_calls)
+
     def test_customer_subscription_deleted_sets_plan_to_free(self):
         self.owner.plan = "users-inappy"
         self.owner.plan_user_count = 20

billing/views.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import datetime
 from typing import List
 
 import stripe
@@ -12,6 +13,7 @@
 
 from codecov_auth.models import Owner
 from plan.service import PlanService
+from services.task.task import TaskService
 
 from .constants import StripeHTTPHeaders, StripeWebhookEvents
 
@@ -63,6 +65,36 @@ def invoice_payment_failed(self, invoice: stripe.Invoice) -> None:
         owners.update(delinquent=True)
         self._log_updated(list(owners))
 
+        # Send failed payment email to all owner admins
+
+        admins: QuerySet[Owner] = Owner.objects.filter(
+            pk__in={admin for owner in owners for admin in owner.admins}
+        )
+
+        task_service = TaskService()
+        card = (
+            invoice.default_payment_method.card
+            if invoice.default_payment_method
+            else None
+        )
+        template_vars = {
+            "amount": invoice.total / 100,
+            "card_type": card.brand if card else None,
+            "last_four": card.last4 if card else None,
+            "cta_link": invoice.hosted_invoice_url,
+            "date": datetime.now().strftime("%B %-d, %Y"),
+        }
+
+        for admin in admins:
+            if admin.email is not None:
+                task_service.send_email(
+                    to_addr=admin.email,
+                    subject="Your Codecov payment failed",
+                    template_name="failed-payment",
+                    name=admin.username,
+                    **template_vars,
+                )
+
     def customer_subscription_deleted(self, subscription: stripe.Subscription) -> None:
         log.info(
             "Customer Subscription Deleted - Setting free plan and deactivating repos for stripe customer",

services/task/task.py

@@ -391,15 +391,21 @@ def preprocess_upload(self, repoid, commitid, report_code):
         ).apply_async()
 
     def send_email(
-        self, ownerid, template_name: str, from_addr: str, subject: str, **kwargs
+        self,
+        to_addr: str,
+        subject: str,
+        template_name: str,
+        from_addr: str | None = None,
+        **kwargs,
     ):
+        # Templates can be found in worker/templates
         self._create_signature(
             "app.tasks.send_email.SendEmail",
             kwargs=dict(
-                ownerid=ownerid,
+                to_addr=to_addr,
+                subject=subject,
                 template_name=template_name,
                 from_addr=from_addr,
-                subject=subject,
                 **kwargs,
             ),
         ).apply_async()