如何测试Django post_save信号中事务提交后的执行逻辑
核心问题说明
你无法触发else分支不是Django测试事务机制导致的,是你判断「新建/更新对象」的业务逻辑本身存在错误:
- Django
post_save信号的触发时机是模型save()方法完成数据库INSERT/UPDATE操作之后 - 同事务内的查询可以读到当前事务未提交的写入数据,因此只要进入
post_save逻辑,Campaign.objects.filter(pk=instance.pk).exists()永远返回True,哪怕事务还没提交 - 你写的else分支在生产环境也永远不会执行,和测试环境无关
修正信号逻辑
Django的post_save信号自带created关键字参数,值为True时代表当前是新建对象的场景,值为False代表是更新已有对象的场景,不需要手动查库判断,修正后的信号代码如下:
@receiver(post_save, sender=Campaign, dispatch_uid="apps.writing.signals.create_handwriting") def create_handwriting(sender, instance, created, **kwargs): """新建或更新活动时,触发手写云函数(重新)生成手写图片""" if not created: 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))
正确编写测试用例
你已经用到的captureOnCommitCallbacks(execute=True)是Django TestCase专门提供的、用于测试transaction.on_commit逻辑的工具,不需要切换为速度更慢的TransactionTestCase,只需要把新建、更新两个场景拆分为独立测试用例即可:
class TestSignals(TestCase): def setUp(self): self.factory = RequestFactory() # 测试前连接信号 signals.post_save.connect( sender=Campaign, dispatch_uid="apps.writing.signals.create_handwriting", receiver=create_handwriting, ) # 测试结束后自动断开信号,避免影响其他用例 self.addCleanup( signals.post_save.disconnect, sender=Campaign, dispatch_uid="apps.writing.signals.create_handwriting", ) @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_new_campaign_triggers_create_logic( self, delete_campaign_pages, enqueue_handwriting_generation, log_campaign_progress ): enqueue_handwriting_generation.return_value = True log_campaign_progress.return_value = True with self.captureOnCommitCallbacks(execute=True): user = G(User) campaign = G(Campaign, user=user) # 断言新建场景逻辑调用正确 enqueue_handwriting_generation.assert_called_once_with(campaign_pk=campaign.pk) log_campaign_progress.assert_called_once_with(pk=campaign.pk, stage="campaign", status="t2h-created") # 新建场景不会触发删除已有页面的逻辑 delete_campaign_pages.assert_not_called() @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_edited_campaign_triggers_update_logic( self, delete_campaign_pages, enqueue_handwriting_generation, log_campaign_progress ): enqueue_handwriting_generation.return_value = True log_campaign_progress.return_value = True delete_campaign_pages.return_value = True # 提前创建测试活动 user = G(User) campaign = G(Campaign, user=user) # 重置mock,清除创建活动时产生的调用记录 log_campaign_progress.reset_mock() enqueue_handwriting_generation.reset_mock() delete_campaign_pages.reset_mock() with self.captureOnCommitCallbacks(execute=True): # 修改并保存已有活动,触发更新场景 campaign.title = "修改后的活动标题" campaign.save() # 断言更新场景逻辑调用正确 enqueue_handwriting_generation.assert_called_once_with(campaign_pk=campaign.pk) log_campaign_progress.assert_called_once_with(pk=campaign.pk, stage="campaign", status="t2h-edited") delete_campaign_pages.assert_called_once_with(campaign_pk=campaign.pk)
注意事项
- 测试更新场景时,必须在创建完测试对象后重置mock,否则创建对象时触发的信号调用会导致断言失败
- 如果使用factory_boy生成测试数据,确认工厂类没有默认断开
post_save信号,否则信号逻辑不会触发 - 不要通过事务内查库的方式判断对象是否为新建,同事务内可以读到未提交的写入,这类判断逻辑永远不会生效
内容的提问来源于stack exchange,提问作者John
相关产品推荐
相关产品推荐

