"""Razorpay premium payment integration tests. The Razorpay SDK/network is never touched: the isolated adapter functions in ``app.payments.razorpay_service`` are patched, so these tests exercise the real order/verify/webhook orchestration, security and idempotency without credentials. Covers: server-side order creation, successful payment + activation, invalid signature, webhook capture (authoritative), duplicate callback/webhook idempotency, payment failure, refund revocation, unknown orders, and that the existing manual premium workflow is unaffected. """ import json import os import unittest from decimal import Decimal from unittest import mock os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from werkzeug.security import generate_password_hash from app import create_app from app.extensions import db from app.models import ( Notification, NotificationType, PaymentStatus, Subscription, SubscriptionStatus, User, UserRole, ) from app.models.base import utcnow from app.payments.models import Payment _ADAPTER = "app.payments.razorpay_service" class _PaymentTestBase(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.app.config.update( RAZORPAY_KEY_ID="rzp_test_key", RAZORPAY_KEY_SECRET="rzp_test_secret", RAZORPAY_WEBHOOK_SECRET="whsec", PAYMENTS_ENABLED=True, ) self.ctx = self.app.app_context() self.ctx.push() db.create_all() self.user = self._user("buyer@c.edu") def tearDown(self): db.session.remove() db.drop_all() self.ctx.pop() def _user(self, email, role=UserRole.STUDENT): u = User(email=email, password_hash=generate_password_hash("Passw0rd!"), is_verified=True, role=role) db.session.add(u) db.session.commit() return u def _fake_order(self, *a, **k): # Mirrors the Razorpay order dict shape the adapter returns. return {"id": "order_TESTid001", "amount": k.get("amount", 9900), "currency": "INR"} def _order_for(self, plan="Monthly"): from app.payments import services with mock.patch(f"{_ADAPTER}.create_order", side_effect=self._fake_order): return services.create_premium_order(self.user, plan) def _active_subs(self, user=None): user = user or self.user return db.session.scalars( db.select(Subscription).where( Subscription.user_id == user.id, Subscription.status == SubscriptionStatus.ACTIVE, ) ).all() def _premium_notifications(self): return db.session.scalars( db.select(Notification).where( Notification.user_id == self.user.id, Notification.notification_type == NotificationType.PREMIUM, ) ).all() class OrderCreationTests(_PaymentTestBase): def test_order_records_pending_payment_with_server_amount(self): from app.payments import services with mock.patch(f"{_ADAPTER}.create_order", side_effect=self._fake_order) as m: ctx = services.create_premium_order(self.user, "Monthly") payment = db.session.scalar(db.select(Payment)) self.assertIsNotNone(payment) self.assertEqual(payment.status, PaymentStatus.CREATED) # Default seeded Monthly plan is ₹99 → 9900 paise, computed server-side. self.assertEqual(payment.amount, Decimal("99.00")) self.assertEqual(m.call_args.args[0], 9900) # amount_minor to gateway self.assertEqual(ctx["amount"], 9900) self.assertEqual(ctx["order_id"], "order_TESTid001") self.assertEqual(ctx["key_id"], "rzp_test_key") self.assertNotIn("rzp_test_secret", json.dumps(ctx)) # never leak the secret def test_create_order_route_ignores_client_amount(self): client = self._login() with mock.patch(f"{_ADAPTER}.create_order", side_effect=self._fake_order): resp = client.post("/app/payments/create-order", json={"plan": "Monthly", "amount": 1}) self.assertEqual(resp.status_code, 200) # Amount is derived from the plan, never from the posted body. self.assertEqual(resp.get_json()["amount"], 9900) def test_create_order_unknown_plan_rejected(self): client = self._login() resp = client.post("/app/payments/create-order", json={"plan": "Nope"}) self.assertEqual(resp.status_code, 400) self.assertEqual(resp.get_json()["error"], "plan_unavailable") def _login(self): c = self.app.test_client() c.post("/login", data={"email": "buyer@c.edu", "password": "Passw0rd!"}) return c class VerifyCallbackTests(_PaymentTestBase): def test_successful_verify_activates_premium(self): from app.payments import services self._order_for("Monthly") with mock.patch(f"{_ADAPTER}.verify_payment_signature", return_value=True): payment = services.verify_callback(self.user, "order_TESTid001", "pay_1", "sig") self.assertEqual(payment.status, PaymentStatus.CAPTURED) self.assertEqual(payment.provider_payment_id, "pay_1") self.assertIsNotNone(payment.paid_at) self.assertIsNotNone(payment.subscription_id) self.assertEqual(len(self._active_subs()), 1) titles = [n.title for n in self._premium_notifications()] self.assertIn("Premium Activated", titles) self.assertIn("Payment Successful", titles) def test_verify_is_idempotent(self): from app.payments import services self._order_for("Monthly") with mock.patch(f"{_ADAPTER}.verify_payment_signature", return_value=True): services.verify_callback(self.user, "order_TESTid001", "pay_1", "sig") services.verify_callback(self.user, "order_TESTid001", "pay_1", "sig") self.assertEqual(len(self._active_subs()), 1) # no double activation payments = db.session.scalars(db.select(Payment)).all() self.assertEqual(len(payments), 1) def test_invalid_signature_marks_failed_and_does_not_activate(self): from app.payments import services from app.payments.razorpay_service import SignatureError self._order_for("Monthly") with mock.patch(f"{_ADAPTER}.verify_payment_signature", return_value=False): with self.assertRaises(SignatureError): services.verify_callback(self.user, "order_TESTid001", "pay_x", "bad") payment = db.session.scalar(db.select(Payment)) self.assertEqual(payment.status, PaymentStatus.FAILED) self.assertEqual(len(self._active_subs()), 0) def test_verify_route_success(self): client = self.app.test_client() client.post("/login", data={"email": "buyer@c.edu", "password": "Passw0rd!"}) self._order_for("Monthly") with mock.patch(f"{_ADAPTER}.verify_payment_signature", return_value=True): resp = client.post("/app/payments/verify", json={ "razorpay_order_id": "order_TESTid001", "razorpay_payment_id": "pay_1", "razorpay_signature": "sig"}) self.assertEqual(resp.status_code, 200) self.assertTrue(resp.get_json()["success"]) self.assertEqual(len(self._active_subs()), 1) def test_verify_route_invalid_signature_400(self): client = self.app.test_client() client.post("/login", data={"email": "buyer@c.edu", "password": "Passw0rd!"}) self._order_for("Monthly") with mock.patch(f"{_ADAPTER}.verify_payment_signature", return_value=False): resp = client.post("/app/payments/verify", json={ "razorpay_order_id": "order_TESTid001", "razorpay_payment_id": "pay_1", "razorpay_signature": "bad"}) self.assertEqual(resp.status_code, 400) self.assertEqual(len(self._active_subs()), 0) class WebhookTests(_PaymentTestBase): def _post_webhook(self, payload, *, valid=True): client = self.app.test_client() with mock.patch(f"{_ADAPTER}.verify_webhook_signature", return_value=valid): return client.post( "/app/payments/webhook/razorpay", data=json.dumps(payload), content_type="application/json", headers={"X-Razorpay-Signature": "sig"}, ) def _captured_payload(self, order_id="order_TESTid001", payment_id="pay_1"): return {"event": "payment.captured", "payload": {"payment": {"entity": {"id": payment_id, "order_id": order_id}}}} def test_webhook_capture_activates_authoritatively(self): self._order_for("Monthly") # only CREATED — no browser callback arrived resp = self._post_webhook(self._captured_payload()) self.assertEqual(resp.status_code, 200) payment = db.session.scalar(db.select(Payment)) self.assertEqual(payment.status, PaymentStatus.CAPTURED) self.assertEqual(len(self._active_subs()), 1) def test_duplicate_webhook_is_idempotent(self): self._order_for("Monthly") self._post_webhook(self._captured_payload()) self._post_webhook(self._captured_payload()) # redelivery self.assertEqual(len(self._active_subs()), 1) # never double-activates def test_webhook_invalid_signature_rejected(self): self._order_for("Monthly") resp = self._post_webhook(self._captured_payload(), valid=False) self.assertEqual(resp.status_code, 400) payment = db.session.scalar(db.select(Payment)) self.assertEqual(payment.status, PaymentStatus.CREATED) self.assertEqual(len(self._active_subs()), 0) def test_webhook_payment_failed_marks_failed(self): self._order_for("Monthly") payload = {"event": "payment.failed", "payload": {"payment": {"entity": { "id": "pay_1", "order_id": "order_TESTid001", "error_description": "card declined"}}}} resp = self._post_webhook(payload) self.assertEqual(resp.status_code, 200) payment = db.session.scalar(db.select(Payment)) self.assertEqual(payment.status, PaymentStatus.FAILED) self.assertEqual(len(self._active_subs()), 0) def test_webhook_refund_revokes_premium(self): # Capture first (sets provider_payment_id), then refund by that id. self._order_for("Monthly") self._post_webhook(self._captured_payload(payment_id="pay_9")) self.assertEqual(len(self._active_subs()), 1) refund = {"event": "refund.processed", "payload": {"refund": {"entity": {"id": "rfnd_1", "payment_id": "pay_9"}}}} resp = self._post_webhook(refund) self.assertEqual(resp.status_code, 200) payment = db.session.scalar(db.select(Payment)) self.assertEqual(payment.status, PaymentStatus.REFUNDED) self.assertEqual(len(self._active_subs()), 0) # premium revoked self.assertIn("Refund Processed", [n.title for n in self._premium_notifications()]) def test_webhook_unknown_order_ignored(self): resp = self._post_webhook(self._captured_payload(order_id="order_unknown")) self.assertEqual(resp.status_code, 200) self.assertIsNone(db.session.scalar(db.select(Payment))) self.assertEqual(len(self._active_subs()), 0) class ManualFlowUnaffectedTests(_PaymentTestBase): def test_manual_activation_still_works_without_payment_row(self): # The manual admin activation path is unchanged and creates no Payment. from app.premium.services import activate_premium, is_user_premium activate_premium(self.user, duration_days=30, plan_name="monthly") self.assertTrue(is_user_premium(self.user)) self.assertIsNone(db.session.scalar(db.select(Payment))) if __name__ == "__main__": unittest.main()