Тестирование сигнала django `post_save`, включающего вызовы функций, которые происходят после фиксации транзакции db

Когда выполняются тесты django, транзакции базы данных не фиксируются. Как мне протестировать событие, вызванное созданием объекта, но происходящее после фиксации транзакции базы данных?

У меня есть модель Campaign и приведенный ниже сигнал post_save. Используя Django TestCase, трудно утверждать, что функции внутри transaction.on_commit вызываются в случае, когда создается новый объект Campaign. Когда сигнал выполняется в тестовом контексте, он всегда думает, что редактируется существующий объект кампании, а не что он был создан заново. Поэтому я не могу протестировать ветвь else оператора if.

Как я могу проверить случай, когда Campaign.objects.filter(pk=instance.pk).exists() равно False?

Сигнал:

@receiver(post_save, sender=Campaign, dispatch_uid="apps.writing.signals.create_handwriting")
def create_handwriting(sender, instance, **kwargs):
    """Whenever a campaign is created or updated, trigger the handwriting cloud function to (re)generate the
    handwriting image.
    """

    if Campaign.objects.filter(pk=instance.pk).exists():
        transaction.on_commit(
            lambda: log_campaign_progress(pk=instance.pk, status="t2h-edited", stage="campaign")
        )
        transaction.on_commit(lambda: delete_campaign_pages(campaign_pk=instance.pk))
    else:
        transaction.on_commit(
            lambda: log_campaign_progress(pk=instance.pk, status="t2h-created", stage="campaign")
        )

    transaction.on_commit(lambda: enqueue_handwriting_generation(campaign_pk=instance.pk))

Тест:

class TestSignals(TestCase):
    def setUp(self):
        self.factory = RequestFactory()

    @mock.patch("lettergun.apps.writing.signals.log_campaign_progress")
    @mock.patch("lettergun.apps.writing.signals.enqueue_handwriting_generation")
    @mock.patch("lettergun.apps.writing.signals.delete_campaign_pages")
    def test_create_handwriting_edit_existing_campaign(
        self, delete_campaign_pages, enqueue_handwriting_generation, log_campaign_progress
    ):
        # disconnected in the factory so we need to reconnect it here
        signals.post_save.connect(
            sender=Campaign,
            dispatch_uid="apps.writing.signals.create_handwriting",
            receiver=create_handwriting,
        )
        enqueue_handwriting_generation.return_value = True
        log_campaign_progress.return_value = True

        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            user = G(User)
            campaign = G(Campaign, user=user)

        assert Campaign.objects.get(pk=campaign.pk)
        assert Campaign.objects.filter(pk=campaign.pk).exists()

        enqueue_handwriting_generation.assert_called_with(campaign_pk=campaign.pk)
        log_campaign_progress.assert_called_with(pk=campaign.pk, stage="campaign", status="t2h-edited")
        delete_campaign_pages.assert_called_with(campaign_pk=campaign.pk)

Вернуться на верх