paymentservice/payments/tests.py

import json
import hmac
import hashlib
import base64
import time
from decimal import Decimal
from unittest.mock import patch, MagicMock
from django.test import TestCase
from django.urls import reverse
from rest_framework.test import APIClient
from rest_framework import status

from payments.models import PaymentGatewayConfig, PaymentOrder, PaymentAttempt, PaymentRefund, WebhookLog
from payments.gateways.cashfree import CashfreeGateway
from payments.engine import PaymentEngine


class CashfreeIntegrationTests(TestCase):
    def setUp(self):
        self.client = APIClient()
        self.secret_key = "cfsk_ma_test_secret_for_unit_tests"
        self.config = PaymentGatewayConfig.objects.create(
            name="CASHFREE",
            display_name="Cashfree Payments",
            is_active=True,
            priority=1,
            health_status="HEALTHY",
            app_id="TEST_APP_ID_123",
            secret_key=self.secret_key,
            environment="sandbox"
        )

    @patch("requests.post")
    def test_create_order_success(self, mock_post):
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.content = b'{"cf_order_id": 998811, "payment_session_id": "session_test_abc123"}'
        mock_response.json.return_value = {
            "cf_order_id": 998811,
            "payment_session_id": "session_test_abc123"
        }
        mock_post.return_value = mock_response

        url = reverse("payment-order-create")
        payload = {
            "order_id": "TEST_TRDX_001",
            "amount": "1499.00",
            "currency": "INR",
            "customer_id": "cust_42",
            "customer_name": "Vignesh Tech",
            "customer_email": "vignesh@tradhox.com",
            "customer_phone": "9876543210",
            "return_url": "https://tradhox.com/return?order_id={order_id}"
        }

        response = self.client.post(url, data=payload, format="json")
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertTrue(response.data["success"])
        self.assertEqual(response.data["order_id"], "TEST_TRDX_001")
        self.assertEqual(response.data["active_gateway"], "CASHFREE")
        self.assertEqual(response.data["payment_session_id"], "session_test_abc123")

        # Verify DB records
        order = PaymentOrder.objects.get(order_id="TEST_TRDX_001")
        self.assertEqual(order.status, "ACTIVE")
        self.assertEqual(order.active_gateway, "CASHFREE")

        attempt = PaymentAttempt.objects.filter(order=order).first()
        self.assertIsNotNone(attempt)
        self.assertEqual(attempt.gateway_name, "CASHFREE")
        self.assertEqual(attempt.payment_session_id, "session_test_abc123")

    def test_webhook_signature_verification_and_order_paid(self):
        # Create an existing active order
        order = PaymentOrder.objects.create(
            order_id="TEST_TRDX_002",
            amount=Decimal("499.00"),
            currency="INR",
            customer_id="cust_99",
            status="ACTIVE",
            active_gateway="CASHFREE"
        )
        attempt = PaymentAttempt.objects.create(
            order=order,
            gateway_name="CASHFREE",
            gateway_order_id="998822",
            status="INITIATED"
        )

        webhook_payload = {
            "type": "PAYMENT_SUCCESS_WEBHOOK",
            "event_time": "2026-10-18T21:10:00Z",
            "data": {
                "order": {
                    "order_id": "TEST_TRDX_002",
                    "order_amount": 499.00,
                    "order_currency": "INR"
                },
                "payment": {
                    "cf_payment_id": "cf_pay_776655",
                    "payment_status": "SUCCESS",
                    "payment_amount": 499.00,
                    "payment_currency": "INR",
                    "payment_group": "upi",
                    "bank_reference": "UTR123456789"
                }
            }
        }

        raw_body = json.dumps(webhook_payload).encode("utf-8")
        timestamp = str(int(time.time()))
        to_sign = timestamp.encode("utf-8") + raw_body
        signature = base64.b64encode(
            hmac.new(self.secret_key.encode("utf-8"), to_sign, hashlib.sha256).digest()
        ).decode("utf-8")

        url = reverse("cashfree-webhook")
        response = self.client.post(
            url,
            data=raw_body,
            content_type="application/json",
            HTTP_X_WEBHOOK_SIGNATURE=signature,
            HTTP_X_WEBHOOK_TIMESTAMP=timestamp
        )

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data.get("status"), "acknowledged")

        # Verify DB state
        order.refresh_from_db()
        self.assertEqual(order.status, "PAID")

        attempt.refresh_from_db()
        self.assertEqual(attempt.status, "SUCCESS")
        self.assertEqual(attempt.gateway_payment_id, "cf_pay_776655")
        self.assertEqual(attempt.bank_reference, "UTR123456789")

        # Verify WebhookLog
        log = WebhookLog.objects.filter(gateway_name="CASHFREE").first()
        self.assertIsNotNone(log)
        self.assertTrue(log.signature_valid)
        self.assertTrue(log.processed)

    @patch("requests.get")
    def test_gateway_health_check(self, mock_get):
        mock_response = MagicMock()
        mock_response.status_code = 401  # Server reachable
        mock_get.return_value = mock_response

        url = reverse("payment-engine-health")
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data["gateways"]), 1)
        self.assertEqual(response.data["gateways"][0]["name"], "CASHFREE")
        self.assertEqual(response.data["gateways"][0]["health_status"], "HEALTHY")