Add Stripe Customer Portal session API for account billing.
Expose POST /api/finance/portal/ so authenticated users can open Stripe's hosted portal for plan changes, payment methods, and cancellations (chat_web_app#33 companion).
This commit is contained in:
@@ -0,0 +1,102 @@
|
||||
"""Tests for Stripe Customer Portal session API (mocked Stripe SDK)."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from django.test import override_settings
|
||||
from django.urls import reverse
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APITestCase
|
||||
|
||||
from chat_backend.tests.factories import make_company, make_user
|
||||
from finance.models import Invoice
|
||||
|
||||
|
||||
class CreateBillingPortalSessionViewTestCase(APITestCase):
|
||||
def setUp(self):
|
||||
self.company = make_company()
|
||||
self.user = make_user(company=self.company)
|
||||
self.client.force_authenticate(user=self.user)
|
||||
self.url = reverse("finance_portal")
|
||||
|
||||
def _create_invoice_with_customer(self, customer_id="cus_test_abc"):
|
||||
return Invoice.objects.create(
|
||||
user=self.user,
|
||||
company=self.company,
|
||||
amount_due=1000,
|
||||
amount_paid=1000,
|
||||
status=Invoice.Status.PAID,
|
||||
stripe_checkout_session_id="cs_portal_test",
|
||||
stripe_customer_id=customer_id,
|
||||
stripe_subscription_id="sub_test_abc",
|
||||
description="Chat Subscription",
|
||||
)
|
||||
|
||||
@override_settings(
|
||||
STRIPE_SECRET_KEY="sk_test_fake",
|
||||
STRIPE_PORTAL_RETURN_URL="http://localhost:3000/account/",
|
||||
)
|
||||
@patch("finance.services.stripe_service.stripe.billing_portal.Session.create")
|
||||
def test_creates_portal_session(self, mock_create):
|
||||
self._create_invoice_with_customer()
|
||||
mock_session = MagicMock()
|
||||
mock_session.url = "https://billing.stripe.com/p/session/test_portal"
|
||||
mock_create.return_value = mock_session
|
||||
|
||||
response = self.client.post(self.url, {}, format="json")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||||
self.assertEqual(
|
||||
response.data["portal_url"],
|
||||
"https://billing.stripe.com/p/session/test_portal",
|
||||
)
|
||||
mock_create.assert_called_once_with(
|
||||
customer="cus_test_abc",
|
||||
return_url="http://localhost:3000/account/",
|
||||
)
|
||||
|
||||
@override_settings(
|
||||
STRIPE_SECRET_KEY="sk_test_fake",
|
||||
STRIPE_PORTAL_RETURN_URL="http://localhost:3000/account/",
|
||||
)
|
||||
@patch("finance.services.stripe_service.stripe.billing_portal.Session.create")
|
||||
def test_accepts_custom_return_url(self, mock_create):
|
||||
self._create_invoice_with_customer()
|
||||
mock_session = MagicMock()
|
||||
mock_session.url = "https://billing.stripe.com/p/session/custom"
|
||||
mock_create.return_value = mock_session
|
||||
|
||||
response = self.client.post(
|
||||
self.url,
|
||||
{"return_url": "http://localhost:3000/account/#billing"},
|
||||
format="json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||||
self.assertEqual(
|
||||
mock_create.call_args.kwargs["return_url"],
|
||||
"http://localhost:3000/account/#billing",
|
||||
)
|
||||
|
||||
@override_settings(STRIPE_SECRET_KEY="sk_test_fake")
|
||||
def test_no_stripe_customer_returns_400(self):
|
||||
Invoice.objects.create(
|
||||
user=self.user,
|
||||
company=self.company,
|
||||
amount_due=1000,
|
||||
stripe_checkout_session_id="cs_no_customer",
|
||||
stripe_customer_id="",
|
||||
)
|
||||
response = self.client.post(self.url, {}, format="json")
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn("No Stripe customer", response.data["detail"])
|
||||
|
||||
@override_settings(STRIPE_SECRET_KEY="")
|
||||
def test_missing_stripe_key_returns_503(self):
|
||||
self._create_invoice_with_customer()
|
||||
response = self.client.post(self.url, {}, format="json")
|
||||
self.assertEqual(response.status_code, status.HTTP_503_SERVICE_UNAVAILABLE)
|
||||
|
||||
def test_unauthenticated_rejected(self):
|
||||
self.client.force_authenticate(user=None)
|
||||
response = self.client.post(self.url, {}, format="json")
|
||||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||
Reference in New Issue
Block a user