134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
from unittest import mock
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase, override_settings
|
|
from django.urls import reverse
|
|
from wagtail.models import Page
|
|
|
|
from home.models.pages import CoursePage
|
|
from purchase.models import CoursePurchase, PurchasableProduct
|
|
|
|
|
|
@override_settings(
|
|
STRIPE_SECRET_KEY=None,
|
|
STRIPE_WEBHOOK_SECRET="test",
|
|
GITEA_URL=None,
|
|
)
|
|
class PurchaseStatusTests(TestCase):
|
|
"""Regression tests for purchase status tracking."""
|
|
|
|
def setUp(self):
|
|
self.user = get_user_model().objects.create_user(
|
|
username="alice",
|
|
email="alice@example.com",
|
|
password="pass12345",
|
|
)
|
|
root_page = Page.get_first_root_node()
|
|
self.course = CoursePage(title="Django Course", slug="django-course")
|
|
root_page.add_child(instance=self.course)
|
|
self.product = PurchasableProduct.objects.create(
|
|
name="Django Course",
|
|
course=self.course,
|
|
price_cents=1000,
|
|
currency="pln",
|
|
)
|
|
|
|
def _create_purchase(self, status, session_id="cs_test", refunded=False):
|
|
return CoursePurchase.objects.create(
|
|
user=self.user,
|
|
course=self.course,
|
|
status=status,
|
|
refunded=refunded,
|
|
stripe_checkout_session_id=session_id,
|
|
)
|
|
|
|
def _post_stripe_event(self, event):
|
|
with mock.patch(
|
|
"purchase.views.stripe.Webhook.construct_event", return_value=event
|
|
):
|
|
response = self.client.post(
|
|
reverse("stripe-webhook"),
|
|
data="{}",
|
|
content_type="application/json",
|
|
HTTP_STRIPE_SIGNATURE="test-signature",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_access_depends_on_paid_status(self):
|
|
purchase = self._create_purchase(CoursePurchase.Status.PENDING)
|
|
|
|
self.assertFalse(self.course._user_has_access(self.user))
|
|
|
|
purchase.status = CoursePurchase.Status.FAILED
|
|
purchase.save(update_fields=["status"])
|
|
self.assertFalse(self.course._user_has_access(self.user))
|
|
|
|
purchase.status = CoursePurchase.Status.PAID
|
|
purchase.refunded = False
|
|
purchase.save(update_fields=["status", "refunded"])
|
|
self.assertTrue(self.course._user_has_access(self.user))
|
|
|
|
def test_checkout_session_completed_unpaid_stays_pending(self):
|
|
purchase = self._create_purchase(
|
|
CoursePurchase.Status.PENDING, session_id="cs_completed"
|
|
)
|
|
|
|
self._post_stripe_event(
|
|
{
|
|
"type": "checkout.session.completed",
|
|
"data": {
|
|
"object": {
|
|
"id": "cs_completed",
|
|
"payment_status": "unpaid",
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
purchase.refresh_from_db()
|
|
self.assertEqual(purchase.status, CoursePurchase.Status.PENDING)
|
|
|
|
def test_async_payment_failed_marks_purchase_failed(self):
|
|
purchase = self._create_purchase(
|
|
CoursePurchase.Status.PENDING, session_id="cs_async_failed"
|
|
)
|
|
|
|
self._post_stripe_event(
|
|
{
|
|
"type": "checkout.session.async_payment_failed",
|
|
"data": {"object": {"id": "cs_async_failed"}},
|
|
}
|
|
)
|
|
|
|
purchase.refresh_from_db()
|
|
self.assertEqual(purchase.status, CoursePurchase.Status.FAILED)
|
|
self.assertFalse(self.course._user_has_access(self.user))
|
|
|
|
def test_payment_intent_failed_marks_purchase_failed_with_metadata_fallback(self):
|
|
purchase = self._create_purchase(
|
|
CoursePurchase.Status.PENDING, session_id="cs_pi_failed"
|
|
)
|
|
|
|
self._post_stripe_event(
|
|
{
|
|
"type": "payment_intent.payment_failed",
|
|
"data": {
|
|
"object": {
|
|
"id": "pi_failed",
|
|
"metadata": {
|
|
"user_id": str(self.user.id),
|
|
"client_reference_id": str(self.user.id),
|
|
"purchasable_id": str(self.product.id),
|
|
},
|
|
"receipt_email": self.user.email,
|
|
"charges": {"data": []},
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
purchase.refresh_from_db()
|
|
self.assertEqual(purchase.status, CoursePurchase.Status.FAILED)
|
|
self.assertFalse(self.course._user_has_access(self.user))
|