1+ import json
2+ import time
3+ from datetime import datetime
4+ from unittest .mock import Mock , call , patch
5+
6+ import stripe
7+ from django .conf import settings
8+ from django .test import TestCase
9+ from django .urls import reverse
10+ from rest_framework import status
11+ from rest_framework .test import APIRequestFactory
12+
13+ from billing .constants import DEFAULT_FREE_PLAN , PlanName
14+ from billing .models import Plan
15+ from billing .views import StripeHTTPHeaders , StripeWebhookHandler
16+ from codecov_auth .tests .factories import OwnerFactory
17+ from core .tests .factories import RepositoryFactory
18+ from services .task import TaskService
19+
20+
21+ class MockPaymentIntent :
22+ def __init__ (self , noCard = False ):
23+ self .status = "succeeded"
24+ self .id = "pi_123"
25+ self .next_action = None
26+ if noCard :
27+ self .payment_method = None
28+ else :
29+ self .payment_method = {
30+ "card" : {
31+ "brand" : "visa" ,
32+ "last4" : "1234"
33+ }
34+ }
35+
36+
37+ class MockSubscription :
38+ def __init__ (self , owner , params ):
39+ self .plan = Mock ()
40+ self .plan .id = params ["new_plan" ]
41+ self .customer = owner .stripe_customer_id
42+ self .id = params ["subscription_id" ]
43+ self .quantity = params ["new_quantity" ]
44+
45+
46+ class StripeWebhookHandlerTests (TestCase ):
47+ def setUp (self ):
48+ self .owner = OwnerFactory (
49+ 50+ username = "owner" ,
51+ plan = PlanName .CODECOV_PRO_MONTHLY .value ,
52+ stripe_customer_id = "cus_123" ,
53+ stripe_subscription_id = "sub_123" ,
54+ )
55+ self .handler = StripeWebhookHandler ()
56+
57+ def _send_event (self , payload , errorSig = None ):
58+ timestamp = time .time_ns ()
59+
60+ request = APIRequestFactory ().post (
61+ reverse ("stripe-webhook" ), data = payload , format = "json"
62+ )
63+
64+ return self .client .post (
65+ reverse ("stripe-webhook" ),
66+ ** {
67+ StripeHTTPHeaders .SIGNATURE : errorSig
68+ or "t={},v1={}" .format (
69+ timestamp ,
70+ stripe .WebhookSignature ._compute_signature (
71+ "{}.{}" .format (timestamp , request .body .decode ("utf-8" )),
72+ settings .STRIPE_ENDPOINT_SECRET ,
73+ ),
74+ )
75+ },
76+ data = payload ,
77+ format = "json" ,
78+ )
79+
80+ def add_second_owner (self ):
81+ self .other_owner = OwnerFactory (
82+ stripe_customer_id = "cus_123" ,
83+ stripe_subscription_id = "sub_123" ,
84+ )
85+
86+ @patch ("services.task.TaskService.send_email" )
87+ def test_invoice_payment_succeeded_emails_delinquents (self , mocked_send_email ):
88+ non_admin = OwnerFactory (
email = "[email protected] " )
89+ admin_1 = OwnerFactory (
email = "[email protected] " )
90+ admin_2 = OwnerFactory (
email = "[email protected] " )
91+ self .owner .admins = [admin_1 .ownerid , admin_2 .ownerid ]
92+ self .owner .plan_activated_users = [non_admin .ownerid ]
93+ self .
owner .
email = "[email protected] " 94+ self .owner .delinquent = True
95+ self .owner .save ()
96+ self .add_second_owner ()
97+ self .other_owner .delinquent = False
98+ self .other_owner .save ()
99+
100+ response = self ._send_event (
101+ payload = {
102+ "type" : "invoice.payment_succeeded" ,
103+ "data" : {
104+ "object" : {
105+ "customer" : self .owner .stripe_customer_id ,
106+ "subscription" : self .owner .stripe_subscription_id ,
107+ "total" : 24000 ,
108+ "hosted_invoice_url" : "https://stripe.com" ,
109+ }
110+ },
111+ }
112+ )
113+
114+ self .owner .refresh_from_db ()
115+ self .other_owner .refresh_from_db ()
116+ self .assertEqual (response .status_code , status .HTTP_204_NO_CONTENT )
117+ self .assertFalse (self .owner .delinquent )
118+ self .assertFalse (self .other_owner .delinquent )
119+
120+ expected_calls = [
121+ call (
122+ to_addr = self .owner .email ,
123+ subject = "You're all set" ,
124+ template_name = "success-after-failed-payment" ,
125+ amount = 240 ,
126+ cta_link = "https://stripe.com" ,
127+ date = datetime .now ().strftime ("%B %-d, %Y" ),
128+ ),
129+ call (
130+ to_addr = admin_1 .email ,
131+ subject = "You're all set" ,
132+ template_name = "success-after-failed-payment" ,
133+ amount = 240 ,
134+ cta_link = "https://stripe.com" ,
135+ date = datetime .now ().strftime ("%B %-d, %Y" ),
136+ ),
137+ call (
138+ to_addr = admin_2 .email ,
139+ subject = "You're all set" ,
140+ template_name = "success-after-failed-payment" ,
141+ amount = 240 ,
142+ cta_link = "https://stripe.com" ,
143+ date = datetime .now ().strftime ("%B %-d, %Y" ),
144+ ),
145+ ]
146+
147+ mocked_send_email .assert_has_calls (expected_calls )
148+
149+ @patch ("services.billing.stripe.PaymentIntent.retrieve" )
150+ def test_invoice_payment_failed_sets_owner_delinquent (self , retrieve_paymentintent_mock ):
151+ self .owner .delinquent = False
152+ self .owner .save ()
153+
154+ retrieve_paymentintent_mock .return_value = stripe .PaymentIntent .construct_from (
155+ {
156+ "status" : "requires_action" ,
157+ "next_action" : {"type" : "verify_with_microdeposits" },
158+ },
159+ "payment_intent_asdf" ,
160+ )
161+
162+ response = self ._send_event (
163+ payload = {
164+ "type" : "invoice.payment_failed" ,
165+ "data" : {
166+ "object" : {
167+ "customer" : self .owner .stripe_customer_id ,
168+ "subscription" : self .owner .stripe_subscription_id ,
169+ "total" : 24000 ,
170+ "hosted_invoice_url" : "https://stripe.com" ,
171+ "payment_intent" : "payment_intent_asdf" ,
172+ "default_payment_method" : {},
173+ }
174+ },
175+ }
176+ )
177+
178+ self .owner .refresh_from_db ()
179+ self .assertEqual (response .status_code , status .HTTP_204_NO_CONTENT )
180+ self .assertTrue (self .owner .delinquent )
181+
182+ @patch ("services.task.TaskService.send_email" )
183+ @patch ("services.billing.stripe.PaymentIntent.retrieve" )
184+ def test_invoice_payment_failed_sends_email_to_admins (
185+ self ,
186+ retrieve_paymentintent_mock ,
187+ mocked_send_email ,
188+ ):
189+ non_admin = OwnerFactory (
email = "[email protected] " )
190+ admin_1 = OwnerFactory (
email = "[email protected] " )
191+ admin_2 = OwnerFactory (
email = "[email protected] " )
192+ self .owner .admins = [admin_1 .ownerid , admin_2 .ownerid ]
193+ self .owner .plan_activated_users = [non_admin .ownerid ]
194+ self .
owner .
email = "[email protected] " 195+ self .owner .save ()
196+
197+ retrieve_paymentintent_mock .return_value = MockPaymentIntent ()
198+
199+ response = self ._send_event (
200+ payload = {
201+ "type" : "invoice.payment_failed" ,
202+ "data" : {
203+ "object" : {
204+ "customer" : self .owner .stripe_customer_id ,
205+ "subscription" : self .owner .stripe_subscription_id ,
206+ "total" : 24000 ,
207+ "hosted_invoice_url" : "https://stripe.com" ,
208+ "payment_intent" : "payment_intent_asdf" ,
209+ "default_payment_method" : {},
210+ }
211+ },
212+ }
213+ )
214+
215+ self .owner .refresh_from_db ()
216+ self .assertEqual (response .status_code , status .HTTP_204_NO_CONTENT )
217+ self .assertTrue (self .owner .delinquent )
218+
219+ expected_calls = [
220+ call (
221+ to_addr = self .owner .email ,
222+ subject = "Your Codecov payment failed" ,
223+ template_name = "failed-payment" ,
224+ name = self .owner .username ,
225+ amount = 240 ,
226+ card_type = "visa" ,
227+ last_four = "1234" ,
228+ cta_link = "https://stripe.com" ,
229+ date = datetime .now ().strftime ("%B %-d, %Y" ),
230+ ),
231+ call (
232+ to_addr = admin_1 .email ,
233+ subject = "Your Codecov payment failed" ,
234+ template_name = "failed-payment" ,
235+ name = admin_1 .username ,
236+ amount = 240 ,
237+ card_type = "visa" ,
238+ last_four = "1234" ,
239+ cta_link = "https://stripe.com" ,
240+ date = datetime .now ().strftime ("%B %-d, %Y" ),
241+ ),
242+ call (
243+ to_addr = admin_2 .email ,
244+ subject = "Your Codecov payment failed" ,
245+ template_name = "failed-payment" ,
246+ name = admin_2 .username ,
247+ amount = 240 ,
248+ card_type = "visa" ,
249+ last_four = "1234" ,
250+ cta_link = "https://stripe.com" ,
251+ date = datetime .now ().strftime ("%B %-d, %Y" ),
252+ ),
253+ ]
254+
255+ mocked_send_email .assert_has_calls (expected_calls )
256+
257+ def test_customer_subscription_deleted_sets_plan_to_free (self ):
258+ self .owner .plan = PlanName .CODECOV_PRO_YEARLY .value
259+ self .owner .plan_user_count = 20
260+ self .owner .save ()
261+
262+ self ._send_event (
263+ payload = {
264+ "type" : "customer.subscription.deleted" ,
265+ "data" : {
266+ "object" : {
267+ "id" : self .owner .stripe_subscription_id ,
268+ "customer" : self .owner .stripe_customer_id ,
269+ "plan" : {"name" : self .owner .plan },
270+ "status" : "active" ,
271+ }
272+ },
273+ }
274+ )
275+ self .owner .refresh_from_db ()
276+
277+ self .assertEqual (self .owner .plan , DEFAULT_FREE_PLAN )
278+ self .assertEqual (self .owner .plan_user_count , 1 )
279+ self .assertIsNone (self .owner .plan_activated_users )
280+ self .assertIsNone (self .owner .stripe_subscription_id )
281+
282+ def test_customer_subscription_deleted_deactivates_all_repos (self ):
283+ RepositoryFactory (author = self .owner , activated = True , active = True )
284+ RepositoryFactory (author = self .owner , activated = True , active = True )
285+ RepositoryFactory (author = self .owner , activated = True , active = True )
286+
287+ self .assertEqual (
288+ self .owner .repository_set .filter (activated = True , active = True ).count (), 3
289+ )
290+
291+ self ._send_event (
292+ payload = {
293+ "type" : "customer.subscription.deleted" ,
294+ "data" : {
295+ "object" : {
296+ "id" : self .owner .stripe_subscription_id ,
297+ "customer" : self .owner .stripe_customer_id ,
298+ "plan" : {"name" : PlanName .CODECOV_PRO_MONTHLY .value },
299+ "status" : "active" ,
300+ }
301+ },
302+ }
303+ )
304+
305+ self .assertEqual (
306+ self .owner .repository_set .filter (activated = False , active = False ).count (), 3
307+ )
0 commit comments