扩展Django Social Auth管道，测试时阻止指定函数执行

我在使用Python Django Social Auth，扩展了认证管道的三个步骤：

• 用户创建前的部分管道（partial pipeline），用于请求部分数据；
• 自定义用户创建步骤，覆盖social.pipeline.user.create_user方法；
• 用户创建后的处理步骤。

当前管道配置：

SOCIAL_AUTH_PIPELINE = (
    'social_core.pipeline.social_auth.social_details',
    'social_core.pipeline.social_auth.social_uid',
    'social_core.pipeline.social_auth.social_user',
    'myapp.file.before_user_is_created',
    'myapp.file.create_user',
    'social_core.pipeline.social_auth.associate_user',
    'myapp.file.after_user_creation',
    'social_core.pipeline.social_auth.load_extra_data',
    'social_core.pipeline.user.user_details',
)

测试代码如下：

@mock.patch("social_core.backends.base.BaseAuth.request")
def test_complete(self, mock_request):
        url = reverse("social:complete", kwargs={"backend": "facebook"})
        url += "?code=2&state=1"
        mock_request.return_value.json.return_value = {"access_token": "123"}
        with mock.patch(
            "django.contrib.sessions.backends.base.SessionBase.set_expiry",
            side_effect=[OverflowError, None],
        ):
            response_1 = self.client.get(url)
            self.assertEqual(response_1.status_code, 302)
            self.assertEqual(response_1.url, "/before-user-is-created/")

            response_2 = self.client.post("/before-user-is-created/", {"some_keys": "some_values"})
            self.assertEqual(response_2.status_code, 302)
            self.assertEqual(response_2.url, "/social-auth/complete/facebook/")

            response_3 = self.client.post("/social-auth/complete/facebook/")
            return response_3

其中用户创建后的处理函数after_user_creation会调用function_called：

def after_user_creation(user, *args, **kwargs):
    ...
    function_called(something_from_user)

现在需要解决的问题是：function_called会在测试的response_2和response_3阶段执行，如何在测试中阻止这个函数运行？

解决方案 #

方法1：直接Mock function_called函数 #

通过unittest.mock直接拦截function_called的调用，让它不执行实际逻辑。关键是要确保Mock的路径和函数被调用的路径一致（即after_user_creation所在模块中引用function_called的路径）。

修改后的测试代码：

@mock.patch("social_core.backends.base.BaseAuth.request")
@mock.patch("myapp.file.function_called")  # 匹配实际调用路径
def test_complete(self, mock_function_called, mock_request):
        url = reverse("social:complete", kwargs={"backend": "facebook"})
        url += "?code=2&state=1"
        mock_request.return_value.json.return_value = {"access_token": "123"}
        
        # 可选：设置Mock返回值，不设置则默认返回None
        mock_function_called.return_value = None
        
        with mock.patch(
            "django.contrib.sessions.backends.base.SessionBase.set_expiry",
            side_effect=[OverflowError, None],
        ):
            response_1 = self.client.get(url)
            self.assertEqual(response_1.status_code, 302)
            self.assertEqual(response_1.url, "/before-user-is-created/")

            response_2 = self.client.post("/before-user-is-created/", {"some_keys": "some_values"})
            self.assertEqual(response_2.status_code, 302)
            self.assertEqual(response_2.url, "/social-auth/complete/facebook/")

            response_3 = self.client.post("/social-auth/complete/facebook/")
            
            # 可选：验证函数是否被调用（按需使用）
            # mock_function_called.assert_called_once_with(expected_argument)
            
            return response_3

方法2：临时移除管道中的after_user_creation步骤 #

如果不需要执行整个after_user_creation步骤，可以用override_settings临时覆盖管道配置，去掉该步骤。

修改后的测试代码：

from django.test import override_settings

@mock.patch("social_core.backends.base.BaseAuth.request")
@override_settings(
    SOCIAL_AUTH_PIPELINE = (
        'social_core.pipeline.social_auth.social_details',
        'social_core.pipeline.social_auth.social_uid',
        'social_core.pipeline.social_auth.social_user',
        'myapp.file.before_user_is_created',
        'myapp.file.create_user',
        'social_core.pipeline.social_auth.associate_user',
        # 移除after_user_creation步骤
        'social_core.pipeline.social_auth.load_extra_data',
        'social_core.pipeline.user.user_details',
    )
)
def test_complete(self, mock_request):
        # 原测试逻辑不变
        url = reverse("social:complete", kwargs={"backend": "facebook"})
        url += "?code=2&state=1"
        mock_request.return_value.json.return_value = {"access_token": "123"}
        with mock.patch(
            "django.contrib.sessions.backends.base.SessionBase.set_expiry",
            side_effect=[OverflowError, None],
        ):
            response_1 = self.client.get(url)
            self.assertEqual(response_1.status_code, 302)
            self.assertEqual(response_1.url, "/before-user-is-created/")

            response_2 = self.client.post("/before-user-is-created/", {"some_keys": "some_values"})
            self.assertEqual(response_2.status_code, 302)
            self.assertEqual(response_2.url, "/social-auth/complete/facebook/")

            response_3 = self.client.post("/social-auth/complete/facebook/")
            return response_3

选择建议 #

• 若仅需阻止function_called执行，保留after_user_creation的其他逻辑，选方法1；
• 若整个after_user_creation步骤都不需要在测试中运行，选方法2。

内容的提问来源于stack exchange，提问作者Tiago Peres