"""Increment 2: admin payments dashboard, reporting, user history, receipt. Covers KPIs/revenue math, filters, search, the details drawer data, the derived timeline, user payment history (own-only), the receipt page (own-only + 404 for others), CSV export, analytics aggregation, authorization and empty states. The Increment 1 processing code is untouched and its tests still pass separately. """ import os import unittest import uuid from datetime import timedelta from decimal import Decimal 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 ( PaymentStatus, Subscription, SubscriptionStatus, User, UserProfile, UserRole, ) from app.models.base import utcnow from app.payments.models import Payment class _Base(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() self.admin = self._user("admin@c.edu", role=UserRole.SUPER_ADMIN, name="Ada Admin") self.buyer = self._user("buyer@c.edu", name="Bob Buyer") self.other = self._user("other@c.edu", name="Ola Other") def tearDown(self): db.session.remove() db.drop_all() self.ctx.pop() def _user(self, email, role=UserRole.STUDENT, name="User"): u = User(email=email, password_hash=generate_password_hash("Passw0rd!"), is_verified=True, role=role) db.session.add(u) db.session.commit() db.session.add(UserProfile(user_id=u.id, full_name=name)) db.session.commit() return u def _sub(self, user, status=SubscriptionStatus.ACTIVE, days=30): now = utcnow() s = Subscription(user_id=user.id, plan_name="Monthly", status=status, start_date=now, end_date=now + timedelta(days=days)) db.session.add(s) db.session.commit() return s def _payment(self, user, status, *, amount="99.00", plan="Monthly", provider="razorpay", pay_id=None, paid=False, sub=None): p = Payment( user_id=user.id, plan_name=plan, provider=provider, provider_order_id="order_" + uuid.uuid4().hex[:16], provider_payment_id=pay_id, amount=Decimal(amount), currency="INR", status=status, subscription_id=(sub.id if sub else None), paid_at=(utcnow() if paid else None), provider_signature=("sig" if status in (PaymentStatus.CAPTURED, PaymentStatus.REFUNDED) else None), ) db.session.add(p) db.session.commit() return p def _login(self, email): c = self.app.test_client() c.post("/login", data={"email": email, "password": "Passw0rd!"}) return c class AdminKpiRevenueTests(_Base): def test_kpis_and_revenue_math(self): from app.payments import admin_services as pa sub = self._sub(self.buyer) self._payment(self.buyer, PaymentStatus.CAPTURED, amount="99.00", paid=True, sub=sub, pay_id="pay_a") self._payment(self.other, PaymentStatus.CAPTURED, amount="999.00", plan="Yearly", paid=True, pay_id="pay_b") self._payment(self.buyer, PaymentStatus.FAILED, amount="99.00") self._payment(self.buyer, PaymentStatus.CREATED, amount="99.00") self._payment(self.other, PaymentStatus.REFUNDED, amount="99.00", paid=True, pay_id="pay_c") k = pa.payment_kpis() self.assertEqual(k["total_revenue"], Decimal("1098.00")) # 99 + 999 captured self.assertEqual(k["successful"], 2) self.assertEqual(k["failed"], 1) self.assertEqual(k["pending"], 1) self.assertEqual(k["refunded"], 1) self.assertEqual(k["active_subscribers"], 1) # success rate = captured / (captured + failed) = 2/3 self.assertAlmostEqual(k["success_rate"], 66.7, places=1) def test_revenue_analytics_shapes(self): from app.payments import admin_services as pa self._payment(self.buyer, PaymentStatus.CAPTURED, amount="99.00", paid=True, pay_id="p1") self._payment(self.other, PaymentStatus.CAPTURED, amount="999.00", plan="Yearly", paid=True, pay_id="p2") a = pa.revenue_analytics() self.assertEqual(len(a["revenue_over_time"]["labels"]), 30) # Refunds are external/synced-only — not a status metric on the dashboard. self.assertEqual(a["status_distribution"]["labels"], ["Successful", "Failed", "Pending"]) self.assertIn("Monthly", a["revenue_by_plan"]["labels"]) self.assertIn("Yearly", a["revenue_by_plan"]["labels"]) self.assertIn("Razorpay", a["provider_breakdown"]["labels"]) self.assertIn("Offline", a["provider_breakdown"]["labels"]) class AdminDashboardRouteTests(_Base): def test_dashboard_renders_with_data(self): self._payment(self.buyer, PaymentStatus.CAPTURED, paid=True, pay_id="pay_x") html = self._login("admin@c.edu").get("/admin/payments").get_data(as_text=True) self.assertIn("Total Revenue", html) self.assertIn("Payment Success Rate", html) self.assertIn("data-pay-row", html) # table row for drawer self.assertIn("chartRevenue", html) # revenue chart canvas self.assertIn("Bob Buyer", html) def test_filter_by_status(self): self._payment(self.buyer, PaymentStatus.CAPTURED, paid=True, pay_id="pay_ok") self._payment(self.other, PaymentStatus.FAILED) client = self._login("admin@c.edu") html = client.get("/admin/payments?status=failed").get_data(as_text=True) self.assertIn("Ola Other", html) self.assertNotIn('data-name="Bob Buyer"', html) def test_search_by_email_and_order(self): p = self._payment(self.buyer, PaymentStatus.CAPTURED, paid=True, pay_id="pay_srch") client = self._login("admin@c.edu") html = client.get("/admin/payments?q=buyer@c.edu").get_data(as_text=True) self.assertIn("Bob Buyer", html) html2 = client.get(f"/admin/payments?q={p.provider_order_id}").get_data(as_text=True) self.assertIn("Bob Buyer", html2) html3 = client.get("/admin/payments?q=nonexistent-xyz").get_data(as_text=True) self.assertIn("No payments match your filters", html3) def test_drawer_data_attributes_present(self): self._payment(self.buyer, PaymentStatus.CAPTURED, paid=True, pay_id="pay_drw") html = self._login("admin@c.edu").get("/admin/payments").get_data(as_text=True) self.assertIn('data-order="order_', html) self.assertIn('data-payment="pay_drw"', html) self.assertIn('data-webhook=', html) self.assertIn('data-verified="yes"', html) def test_empty_state_no_payments(self): html = self._login("admin@c.edu").get("/admin/payments").get_data(as_text=True) self.assertIn("No payments yet", html) def test_refund_ui_removed_but_backend_still_records(self): # A refunded payment exists (synced from Razorpay) ... self._payment(self.other, PaymentStatus.REFUNDED, paid=True, pay_id="pay_ref") html = self._login("admin@c.edu").get("/admin/payments").get_data(as_text=True) # ... but no refund KPI card or refund filter option is shown. self.assertNotIn("Refunded Payments", html) self.assertNotIn('value="refunded"', html) # Backend still tracks it (history + count preserved for sync/audit). from app.payments import admin_services as pa self.assertEqual(pa.payment_kpis()["refunded"], 1) self.assertNotIn(("refunded", "Refunded"), pa.STATUS_OPTIONS) def test_refunded_record_still_listed_as_history(self): # Refunded payments remain visible in the table as history records. self._payment(self.other, PaymentStatus.REFUNDED, paid=True, pay_id="pay_hist") html = self._login("admin@c.edu").get("/admin/payments").get_data(as_text=True) self.assertIn("pay_hist", html) self.assertIn("Ola Other", html) def test_export_csv(self): self._payment(self.buyer, PaymentStatus.CAPTURED, paid=True, pay_id="pay_csv") resp = self._login("admin@c.edu").get("/admin/payments/export.csv") self.assertEqual(resp.status_code, 200) self.assertIn("text/csv", resp.content_type) body = resp.get_data(as_text=True) self.assertIn("User,Email,Plan,Amount", body) self.assertIn("buyer@c.edu", body) self.assertNotIn("provider_signature", body) # never leak signatures def test_authorization_requires_admin(self): resp = self._login("buyer@c.edu").get("/admin/payments") self.assertNotEqual(resp.status_code, 200) # students blocked anon = self.app.test_client().get("/admin/payments") self.assertNotEqual(anon.status_code, 200) class TimelineTests(_Base): def test_timeline_derivation(self): from app.payments.admin_services import payment_timeline captured = self._payment(self.buyer, PaymentStatus.CAPTURED, paid=True, pay_id="pc", sub=self._sub(self.buyer)) labels = [s["label"] for s in payment_timeline(captured)] self.assertEqual(labels, ["Order created", "Payment captured", "Premium activated", "Notification sent"]) failed = self._payment(self.other, PaymentStatus.FAILED) self.assertEqual([s["label"] for s in payment_timeline(failed)], ["Order created", "Payment failed"]) refunded = self._payment(self.other, PaymentStatus.REFUNDED, paid=True, pay_id="pr") rlabels = [s["label"] for s in payment_timeline(refunded)] self.assertIn("Refund processed", rlabels) self.assertIn("Premium revoked", rlabels) class UserHistoryReceiptTests(_Base): def test_history_shows_own_payments_only(self): self._payment(self.buyer, PaymentStatus.CAPTURED, paid=True, pay_id="pay_mine") self._payment(self.other, PaymentStatus.CAPTURED, paid=True, pay_id="pay_theirs") html = self._login("buyer@c.edu").get("/app/payments/history").get_data(as_text=True) self.assertIn("pay_mine", html) self.assertNotIn("pay_theirs", html) def test_history_empty_state(self): html = self._login("buyer@c.edu").get("/app/payments/history").get_data(as_text=True) self.assertIn("No payments yet", html) def test_history_shows_active_subscription(self): self._sub(self.buyer) html = self._login("buyer@c.edu").get("/app/payments/history").get_data(as_text=True) self.assertIn("Current Premium Subscription", html) self.assertIn("remaining", html) def test_receipt_own_ok(self): p = self._payment(self.buyer, PaymentStatus.CAPTURED, paid=True, pay_id="pay_rcpt") resp = self._login("buyer@c.edu").get(f"/app/payments/receipt/{p.public_id}") self.assertEqual(resp.status_code, 200) body = resp.get_data(as_text=True) self.assertIn("Premium Membership Receipt", body) self.assertIn("pay_rcpt", body) self.assertIn("Bob Buyer", body) def test_receipt_other_user_404(self): p = self._payment(self.other, PaymentStatus.CAPTURED, paid=True, pay_id="pay_secret") resp = self._login("buyer@c.edu").get(f"/app/payments/receipt/{p.public_id}") self.assertEqual(resp.status_code, 404) if __name__ == "__main__": unittest.main()