[HELP] Como criar testes automatizados para testar os webhooks do Stripe no Django?

Contextualizando

Implementei um checkout de pagamento com Django, Django Rest Framework e Stripe. Para o frontend, utilizei React + Next.

1. O primeiro passo do checkout de pagamento foi criar uma sessão de checkout a partir do backend (views.py), em seguida, retorno para o front-end o client_session_id que por sua vez renderizará o checkout "embedded" do stripe. Mais detalhes podem ser conferidos aqui: Stripe - Quickstart Embedded Checkout

2. Em seguida, criei os webhooks para verificar e salvar os dados do usuário assim que o pagamento for validado. Os eventos que estou ouvindo são: payment_intent.created, checkout.session.completed, payment_intent.succeeded, payment_intent.payment_failed e payment_intent.canceled

Respectivamente, pra cada evento acionado, a minha view vai alterar o status de pagamento do usuário e em caso de falha, enviar um e-mail. Demostrado no código a seguir:

# views.py

class WHUpdatePaymentStatusView(APIView):
    authentication_classes = (CsrfExemptSessionAuthentication, BasicAuthentication)

    def post(self, request, *args, **kwargs):
        try:
            payload_unicode = request.body.decode("utf-8")

            sig_header = request.META.get("HTTP_STRIPE_SIGNATURE")

            event = stripe.Webhook.construct_event(
                payload_unicode, sig_header, endpoint_secret
            )

            data = event.data.object

            if event.type == "payment_intent.created":
                user = User.objects.get(customer_payment_id=data["customer"])

                price_id = data["metadata"]["price_id"]
                cargo = Cargos.objects.get(price_id_payment=price_id)

                is_candidate_test = not data["livemode"]
                payment_status = "Pendente"
                data_inscricao = None

                try:
                    candidato = Candidato.objects.get(user=user, cargo=cargo)

                    candidato.payment_intent_id = data["id"]
                    candidato.status_pagamento = payment_status
                    candidato.data_inscricao = data_inscricao
                    candidato.updated_at = timezone.now()

                    candidato.save()
                except Candidato.DoesNotExist:
                    candidato = Candidato(
                        user=user,
                        cargo=cargo,
                        payment_intent_id=data["id"],
                        is_candidate_test=is_candidate_test,
                        status_pagamento=payment_status,
                        data_inscricao=data_inscricao,
                        updated_at=timezone.now(),
                    )

                    candidato.save()
                except Exception as e:
                    print(e)

            if event.type == "checkout.session.completed":
                candidato = Candidato.objects.get(
                    payment_intent_id=data["payment_intent"]
                )

                if (
                    candidato.status_pagamento == "Pendente"
                    and data["payment_status"] == "unpaid"
                ):
                    candidato.status_pagamento == "Processando"
                    candidato.updated_at = timezone.now()
                    candidato.save()
                elif (
                    candidato.status_pagamento == "Pendente"
                    and data["payment_status"] == "paid"
                ):
                    candidato.status_pagamento == "Pago"
                    candidato.data_inscricao = timezone.now()
                    candidato.updated_at = timezone.now()
                    candidato.save()

            if event.type == "payment_intent.succeeded":
                candidato = Candidato.objects.get(payment_intent_id=data["id"])

                if (
                    candidato.status_pagamento == "Processando"
                    or candidato.status_pagamento == "Pendente"
                ):
                    candidato.status_pagamento = "Pago"
                    candidato.data_inscricao = timezone.now()
                    candidato.updated_at = timezone.now()
                    candidato.save()

            if (
                event.type == "payment_intent.payment_failed"
                or event.type == "payment_intent.canceled"
            ):
                candidato = Candidato.objects.get(payment_intent_id=data["id"])

                candidato.status_pagamento = "Cancelado"
                candidato.updated_at = timezone.now()
                candidato.save()
                #Logica de enviar email aqui

Testes automatizados com o Django

Ao rodar o comando python manage.py test o django se encarrega de:

• Criar uma cópia do banco de dados com o nome: test_<nome_do_bd>
• Executar o código nessa instância de teste
• Em seguida, apagar ela

Para o meu conjunto de testes utilizo a classe APITestCase do rest_framework.test

O Problema

Seguindo a documentação do STRIPE, é recomendado na criação dos testes utilizar a PaymentItent API como no exemplo a seguir, pois, ao passar o atributo confirm=True o stripe cria todos os dados fakes necessários e completa o pagamento e assim, acionando os eventos relacionados:

import stripe
stripe.api_key = "<your_key>"

stripe.PaymentIntent.create(
  amount=1099,
  currency="usd",
  confirm=True  # esse carinha aqui
  payment_method_types=["card"],
)

Pois bem, chegamos ao problema.

Como o django cria uma instância do BD automaticamente só para rodar os testes, eu crio alguns registros durante o teste para o webhook manipular estes dados posteriormente.

Ex: Durante a execução do teste eu crio um novo Concurso (entidade
na minha API), Cargo (entidade na minha API), um Usuário
(entidade na minha API) vinculado a um cliente no stripe e por fim,
crio o PaymentIntent *(stripe API).

OBS: Como o código é grante, criei um GIST e ele está linkado no final da pergunta.

Até a finalização do fluxo de teste, o código funciona sem problemas.

O problema mesmo é que os eventos do stripe são assíncronos, ou seja, o webhook será executado após o teste, e, utiliza o banco de dados principal ao invés do banco de dados de teste criado automaticamente pelo django.

Isso acontece pois o escopo da execução da view (webhook - servidor rodando localmente) está fora do bloco de teste.

O que já tentei e não é viável:

Eu poderia mockar os dados que são lançados pelos eventos do Stripe
para o meu webhook e testar cada webhook que implementei, mas eu não
conseguiria simular a ordem de lançamento dos eventos originais do
stripe e nem criar novos dados de teste no stripe, que são cruciais
para mim.

Eu poderia utilizar o Stripe CLI e lançar individualmente cada evento
que eu gostaria de testar com dados de teste. Mas eu precisaria fazer
isso manualmente pra cada evento cada vez que eu quisesse testar a
minha view de webhook.

Eu pensei em criar um novo BD localmente com o nome
banco2 (exemplo) e utilizar a instância para testes.
Com isso, rodaria o servidor local apontando para esse BD. Mas ao
executar os testes, o django criaria uma instância a
partir do banco2 chamada de test_banco2
para executar o teste e voltaríamos a estaca zero.

Pensei em utilizar a configuração MIRROR dentro do DATABASES no settings.py para espelhar o banco de dados principal e utilizar a nova instância como teste, mas
pelo que entendi, a nova instância não poderia escrever novos dados,
pois ela é apenas um apontamento à instância de DB principal.

Talvez, a minha lógica de programação esteja equivocada, pois dependo de uma ordem específica dos envios dos eventos do stripe para o funcionamento adequado da criação e alteração dos status de pagamento no meu banco de dados.

Mais o que devo fazer a respeito? Há alguma forma de configurar um banco de dados de teste para rodar os testes nele e apontar o servidor local para o mesmo banco de dados? Se minha lógica estiver quebrada, como proceder? Ou ainda, o que mais eu poderia tentar?

Gist

Link para o o código relacionado -> GIST