shoppy/tpv/tests/test_redsys.py

import base64
import json
from decimal import Decimal
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from django.conf import settings

from tpv.models import PaymentTransaction
from tpv.redsys import RedsysClient
from tpv.settings import ERROR_CODES


class TestRedsysTPV(APITestCase):
    def test_redsys_client(self):
        amount_to_pay = Decimal("10.00")
        transaction = PaymentTransaction.objects.create(amount=amount_to_pay)

        items = [
            {
                "id": 1,
                "name": "Papacolas",
                "description": "Las mejores papacolas",
                "type": "INV",
                "amount_to_pay": str(amount_to_pay),
            }
        ]

        transaction.metadata["items"] = items
        transaction.save()

        client = RedsysClient()
        merchant_parameters = client.get_merchant_parameters_for_transaction(transaction)

        assert merchant_parameters.get("DS_MERCHANT_ORDER") == transaction.hash.hex
        assert merchant_parameters.get("DS_MERCHANT_AMOUNT") == str(
            int(amount_to_pay) * 100
        )
        assert (
            merchant_parameters.get("DS_MERCHANT_TERMINAL") == settings.REDSYS_TERMINAL
        )
        assert (
            merchant_parameters.get("DS_MERCHANT_MERCHANTCODE")
            == settings.REDSYS_MERCHANT_CODE
        )

        client.get_body_for_transaction(transaction)

    def test_redsys_webhook(self):
        amount_to_pay = Decimal("10.00")
        transaction = PaymentTransaction.objects.create(amount=amount_to_pay)

        items = [
            {
                "id": 1,
                "name": "Papacolas",
                "description": "Las mejores papacolas",
                "type": "INV",
                "amount_to_pay": str(amount_to_pay),
            }
        ]
        transaction.metadata["items"] = items
        transaction.save()

        response = self.client.get(
            reverse("tpv:order_created", kwargs={"transaction": transaction.hash})
        )
        assert response.status_code == status.HTTP_200_OK

        redsys_response_data = {
            "Ds_MerchantCode": "999008881",
            "Ds_Terminal": "001",
            "Ds_Order": transaction.hash.hex,
            "Ds_Amount": str(transaction.amount_integer),
            "Ds_Currency": "978",
            "Ds_Date": "01/01/2024",
            "Ds_Hour": "00:00",
            "Ds_SecurePayment": "1",
            "Ds_Card_Number": "454881******1156",
            "Ds_Card_Country": "724",
            "Ds_Response": "0000",
            "Ds_MerchantData": "",
            "Ds_TransactionType": "0",
            "Ds_ConsumerLanguage": "1",
            "Ds_AuthorisationCode": "182670",
            "Ds_Card_Brand": "1",
            "Ds_ProcessedPayMethod": "80",
            "Ds_ECI": "05",
            "Ds_Response_Description": "OPERACION AUTORIZADA",
        }
        redsys_response_data_str = json.dumps(redsys_response_data)
        b64_merchant_params = base64.b64encode(
            redsys_response_data_str.encode()
        ).decode()

        response = self.client.post(
            reverse("tpv:webhook", kwargs={"transaction": transaction.hash}),
            data={
                "Ds_MerchantParameters": b64_merchant_params,
                "Ds_Signature": settings.REDSYS_SHARED_SECRET,
                "Ds_SignatureVersion": "HMAC_SHA256_V1",
            },
        )

        assert response.status_code == status.HTTP_200_OK

        response = self.client.get(
            reverse("tpv:ok", kwargs={"transaction": transaction.hash})
        )
        assert response.status_code == status.HTTP_200_OK

    def test_redsys_webhook_payment_error(self):
        amount_to_pay = Decimal("10.00")
        transaction = PaymentTransaction.objects.create(amount=amount_to_pay)

        items = [
            {
                "id": 1,
                "name": "Papacolas",
                "description": "Las mejores papacolas",
                "type": "INV",
                "amount_to_pay": str(amount_to_pay),
            }
        ]

        transaction.metadata["items"] = items

        transaction.save()

        response = self.client.get(
            reverse("tpv:order_created", kwargs={"transaction": transaction.hash})
        )
        assert response.status_code == status.HTTP_200_OK

        redsys_response_data = {
            "Ds_MerchantCode": "999008881",
            "Ds_Terminal": "001",
            "Ds_Order": transaction.hash.hex,
            "Ds_Amount": "1000",
            "Ds_Currency": "978",
            "Ds_Date": "01/01/2024",
            "Ds_Hour": "00:00",
            "Ds_SecurePayment": "1",
            "Ds_Card_Number": "454881******1156",
            "Ds_Card_Country": "724",
            "Ds_Response": "0184",
            "Ds_MerchantData": "",
            "Ds_TransactionType": "0",
            "Ds_ConsumerLanguage": "1",
            "Ds_AuthorisationCode": "182670",
            "Ds_Card_Brand": "1",
            "Ds_ProcessedPayMethod": "80",
            "Ds_ECI": "05",
            "Ds_Response_Description": "OPERACION AUTORIZADA",
        }

        redsys_response_data_str = json.dumps(redsys_response_data)
        b64_merchant_params = base64.b64encode(
            redsys_response_data_str.encode()
        ).decode()

        response = self.client.post(
            reverse("tpv:webhook", kwargs={"transaction": transaction.hash}),
            data={
                "Ds_MerchantParameters": b64_merchant_params,
                "Ds_Signature": settings.REDSYS_SHARED_SECRET,
                "Ds_SignatureVersion": "HMAC_SHA256_V1",
            },
        )

        assert response.status_code == status.HTTP_409_CONFLICT
        transaction.refresh_from_db()
        assert str(ERROR_CODES.get("0184")) in transaction.observations
        assert transaction.status == PaymentTransaction.StatusChoices.ERROR

        response = self.client.get(
            reverse("tpv:ko", kwargs={"transaction": transaction.hash})
        )
        assert response.status_code == status.HTTP_200_OK