nch请求 #31

Closed
p9r7lebpj wants to merge 2 commits from nch_branch into master

@ -1,88 +1,44 @@
# 导入日志模块,用于记录系统运行时的信息和错误
import logging import logging
# 导入Django的时间工具用于获取当前时间
from django.utils import timezone from django.utils import timezone
# 导入自定义工具:缓存工具和获取博客设置的方法
from djangoblog.utils import cache, get_blog_setting from djangoblog.utils import cache, get_blog_setting
# 导入当前应用的模型:分类和文章模型
from .models import Category, Article from .models import Category, Article
# 创建日志记录器,用于记录当前模块的日志信息
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def seo_processor(requests): def seo_processor(requests):
"""
自定义上下文处理器用于在所有模板中全局共享SEO相关的配置和数据
上下文处理器会在每次请求时被调用返回的字典会自动注入到所有模板中
Args:
requests: HttpRequest对象包含当前请求的信息如协议主机等
Returns:
dict: 包含网站配置导航数据等的字典供模板全局使用
"""
# 定义缓存键,用于标识当前处理器的缓存数据
key = 'seo_processor' key = 'seo_processor'
# 尝试从缓存中获取数据,减少数据库查询和计算开销
value = cache.get(key) value = cache.get(key)
# 如果缓存中存在数据,直接返回缓存内容
if value: if value:
return value return value
else: else:
# 缓存未命中时,记录日志并重新计算数据
logger.info('set processor cache.') logger.info('set processor cache.')
# 获取博客的全局设置(从数据库或其他配置源)
setting = get_blog_setting() setting = get_blog_setting()
# 构建需要返回给模板的数据集
value = { value = {
# 网站名称(用于页面标题等)
'SITE_NAME': setting.site_name, 'SITE_NAME': setting.site_name,
# 是否显示谷歌广告
'SHOW_GOOGLE_ADSENSE': setting.show_google_adsense, 'SHOW_GOOGLE_ADSENSE': setting.show_google_adsense,
# 谷歌广告代码
'GOOGLE_ADSENSE_CODES': setting.google_adsense_codes, 'GOOGLE_ADSENSE_CODES': setting.google_adsense_codes,
# 网站SEO描述用于meta标签
'SITE_SEO_DESCRIPTION': setting.site_seo_description, 'SITE_SEO_DESCRIPTION': setting.site_seo_description,
# 网站描述(用于页面展示)
'SITE_DESCRIPTION': setting.site_description, 'SITE_DESCRIPTION': setting.site_description,
# 网站关键词用于meta标签提升SEO
'SITE_KEYWORDS': setting.site_keywords, 'SITE_KEYWORDS': setting.site_keywords,
# 网站基础URL协议+域名如https://example.com/
'SITE_BASE_URL': requests.scheme + '://' + requests.get_host() + '/', 'SITE_BASE_URL': requests.scheme + '://' + requests.get_host() + '/',
# 文章摘要长度(用于列表页显示)
'ARTICLE_SUB_LENGTH': setting.article_sub_length, 'ARTICLE_SUB_LENGTH': setting.article_sub_length,
# 导航栏显示的分类列表(从数据库查询所有分类)
'nav_category_list': Category.objects.all(), 'nav_category_list': Category.objects.all(),
# 导航栏显示的页面列表(筛选类型为"页面"且状态为"已发布"的文章)
'nav_pages': Article.objects.filter( 'nav_pages': Article.objects.filter(
type='p', # 'p'表示页面page区别于普通文章article type='p',
status='p'), # 'p'表示已发布published status='p'),
# 是否开启网站评论功能
'OPEN_SITE_COMMENT': setting.open_site_comment, 'OPEN_SITE_COMMENT': setting.open_site_comment,
# 网站备案号
'BEIAN_CODE': setting.beian_code, 'BEIAN_CODE': setting.beian_code,
# 网站统计代码如百度统计、Google Analytics
'ANALYTICS_CODE': setting.analytics_code, 'ANALYTICS_CODE': setting.analytics_code,
# 公安备案号
"BEIAN_CODE_GONGAN": setting.gongan_beiancode, "BEIAN_CODE_GONGAN": setting.gongan_beiancode,
# 是否显示公安备案号
"SHOW_GONGAN_CODE": setting.show_gongan_code, "SHOW_GONGAN_CODE": setting.show_gongan_code,
# 当前年份(用于页脚版权信息等)
"CURRENT_YEAR": timezone.now().year, "CURRENT_YEAR": timezone.now().year,
# 全局页头代码如自定义CSS、JS
"GLOBAL_HEADER": setting.global_header, "GLOBAL_HEADER": setting.global_header,
# 全局页脚代码
"GLOBAL_FOOTER": setting.global_footer, "GLOBAL_FOOTER": setting.global_footer,
# 评论是否需要审核后才显示
"COMMENT_NEED_REVIEW": setting.comment_need_review, "COMMENT_NEED_REVIEW": setting.comment_need_review,
} }
# 将数据存入缓存有效期为10小时60秒*60分*10小时
# 减少重复计算和数据库查询,提升性能
cache.set(key, value, 60 * 60 * 10) cache.set(key, value, 60 * 60 * 10)
return value return value

@ -6,40 +6,29 @@ from django.utils.translation import gettext_lazy as _
def disable_commentstatus(modeladmin, request, queryset): def disable_commentstatus(modeladmin, request, queryset):
queryset.update(is_enable=False) queryset.update(is_enable=False)
# 批量禁用评论将选中评论的is_enable字段设为False
def enable_commentstatus(modeladmin, request, queryset): def enable_commentstatus(modeladmin, request, queryset):
queryset.update(is_enable=True) queryset.update(is_enable=True)
# 批量启用评论将选中评论的is_enable字段设为True
disable_commentstatus.short_description = _('Disable comments') disable_commentstatus.short_description = _('Disable comments')
enable_commentstatus.short_description = _('Enable comments') enable_commentstatus.short_description = _('Enable comments')
# 设置动作在后台下拉菜单中的显示名称
class CommentAdmin(admin.ModelAdmin): class CommentAdmin(admin.ModelAdmin):
list_per_page = 20 list_per_page = 20
# 每页显示20条评论
list_display = ( list_display = (
'id', 'id',
'body', 'body',
'link_to_userinfo', # 自定义方法显示用户信息链接 'link_to_userinfo',
'link_to_article', # 自定义方法显示文章链接 'link_to_article',
'is_enable', 'is_enable',
'creation_time') 'creation_time')
# 列表页面显示的字段
list_display_links = ('id', 'body', 'is_enable') list_display_links = ('id', 'body', 'is_enable')
# 这些字段可以作为链接点击进入编辑页面
list_filter = ('is_enable',) list_filter = ('is_enable',)
# 右侧过滤器:按启用状态筛选
exclude = ('creation_time', 'last_modify_time') exclude = ('creation_time', 'last_modify_time')
# 编辑页面中排除这些字段(自动生成的时间戳)
actions = [disable_commentstatus, enable_commentstatus] actions = [disable_commentstatus, enable_commentstatus]
# 注册可用的批量动作
def link_to_userinfo(self, obj): def link_to_userinfo(self, obj):
info = (obj.author._meta.app_label, obj.author._meta.model_name) info = (obj.author._meta.app_label, obj.author._meta.model_name)
@ -47,15 +36,12 @@ class CommentAdmin(admin.ModelAdmin):
return format_html( return format_html(
u'<a href="%s">%s</a>' % u'<a href="%s">%s</a>' %
(link, obj.author.nickname if obj.author.nickname else obj.author.email)) (link, obj.author.nickname if obj.author.nickname else obj.author.email))
# 生成指向用户编辑页面的链接,显示昵称(如果没有则显示邮箱)
def link_to_article(self, obj): def link_to_article(self, obj):
info = (obj.article._meta.app_label, obj.article._meta.model_name) info = (obj.article._meta.app_label, obj.article._meta.model_name)
link = reverse('admin:%s_%s_change' % info, args=(obj.article.id,)) link = reverse('admin:%s_%s_change' % info, args=(obj.article.id,))
return format_html( return format_html(
u'<a href="%s">%s</a>' % (link, obj.article.title)) u'<a href="%s">%s</a>' % (link, obj.article.title))
# 生成指向文章编辑页面的链接,显示文章标题
link_to_userinfo.short_description = _('User') link_to_userinfo.short_description = _('User')
link_to_article.short_description = _('Article') link_to_article.short_description = _('Article')
# 设置自定义方法在列表中的列标题

@ -9,63 +9,31 @@ from blog.models import Article
# Create your models here. # Create your models here.
class Comment(models.Model): class Comment(models.Model):
"""
评论模型类
用于存储博客文章的评论信息支持多级回复功能
"""
# 评论正文最大长度300字符在管理界面显示为'正文'
body = models.TextField('正文', max_length=300) body = models.TextField('正文', max_length=300)
# 评论创建时间默认值为当前时间使用Django的国际化支持
creation_time = models.DateTimeField(_('creation time'), default=now) creation_time = models.DateTimeField(_('creation time'), default=now)
# 评论最后修改时间,默认值为当前时间
last_modify_time = models.DateTimeField(_('last modify time'), default=now) last_modify_time = models.DateTimeField(_('last modify time'), default=now)
# 评论作者,关联用户模型,删除用户时级联删除其所有评论
author = models.ForeignKey( author = models.ForeignKey(
settings.AUTH_USER_MODEL, settings.AUTH_USER_MODEL,
verbose_name=_('author'), verbose_name=_('author'),
on_delete=models.CASCADE) on_delete=models.CASCADE)
# 所属文章,关联文章模型,删除文章时级联删除所有相关评论
article = models.ForeignKey( article = models.ForeignKey(
Article, Article,
verbose_name=_('article'), verbose_name=_('article'),
on_delete=models.CASCADE) on_delete=models.CASCADE)
# 父级评论,支持评论回复功能,允许为空表示顶级评论
parent_comment = models.ForeignKey( parent_comment = models.ForeignKey(
'self', # 自关联,指向同一个模型 'self',
verbose_name=_('parent comment'), verbose_name=_('parent comment'),
blank=True, # 在表单中允许为空 blank=True,
null=True, # 在数据库中允许为NULL null=True,
on_delete=models.CASCADE) # 删除父评论时级联删除子评论 on_delete=models.CASCADE)
# 评论是否启用(审核通过),默认未启用,不允许为空
is_enable = models.BooleanField(_('enable'), is_enable = models.BooleanField(_('enable'),
default=False, blank=False, null=False) default=False, blank=False, null=False)
class Meta: class Meta:
"""
模型的元数据配置
"""
# 按ID降序排列新的评论显示在前面
ordering = ['-id'] ordering = ['-id']
# 单数名称,在管理界面显示
verbose_name = _('comment') verbose_name = _('comment')
# 复数名称,与单数相同
verbose_name_plural = verbose_name verbose_name_plural = verbose_name
# 指定获取最新记录时使用的字段
get_latest_by = 'id' get_latest_by = 'id'
def __str__(self): def __str__(self):
""" return self.body
对象的字符串表示方法
在Django管理界面和其他显示对象的地方使用
"""
return self.body # 直接返回评论正文作为显示内容

@ -37,7 +37,7 @@ class OAuthUserAdmin(admin.ModelAdmin):
# 右侧筛选器配置 # 右侧筛选器配置
list_filter = ('author', 'type',) list_filter = ('author', 'type',)
# 只读字段列表(初始为空) # 只读字段列表,开始为空
readonly_fields = [] readonly_fields = []
def get_readonly_fields(self, request, obj=None): def get_readonly_fields(self, request, obj=None):

@ -11,5 +11,5 @@ class OauthConfig(AppConfig):
""" """
# 指定应用的Python路径Django使用这个名称来识别应用 # 指定应用的Python路径Django使用这个名称来识别应用
# 这应该与应用目录的名称保持一致 # 这应该与应用目录的名称保持相同
name = 'oauth' name = 'oauth'

@ -9,7 +9,7 @@ class RequireEmailForm(forms.Form):
通常在第三方OAuth服务没有返回邮箱信息时使用 通常在第三方OAuth服务没有返回邮箱信息时使用
""" """
# 邮箱字段,标签显示为'电子邮箱',必填字段 # 邮箱字段,标签显示为'电子邮箱'
email = forms.EmailField(label='电子邮箱', required=True) email = forms.EmailField(label='电子邮箱', required=True)
# OAuth用户ID隐藏字段用于关联OAuth用户记录 # OAuth用户ID隐藏字段用于关联OAuth用户记录

@ -12,7 +12,7 @@ class OAuthUser(models.Model):
用于存储通过第三方OAuth服务登录的用户信息 用于存储通过第三方OAuth服务登录的用户信息
""" """
# 关联到系统的本地用户,允许为空初次OAuth登录时可能还未关联 # 关联到系统的本地用户,允许为空
author = models.ForeignKey( author = models.ForeignKey(
settings.AUTH_USER_MODEL, settings.AUTH_USER_MODEL,
verbose_name=_('author'), verbose_name=_('author'),

@ -13,43 +13,81 @@ from oauth.oauthmanager import BaseOauthManager
# Create your tests here. # Create your tests here.
class OAuthConfigTest(TestCase): class OAuthConfigTest(TestCase):
"""
OAuth配置基础测试类
测试OAuth登录流程的基本功能
"""
def setUp(self): def setUp(self):
self.client = Client() """
self.factory = RequestFactory() 测试初始化方法在每个测试方法执行前运行
"""
self.client = Client() # Django测试客户端用于模拟HTTP请求
self.factory = RequestFactory() # 用于创建请求对象的工厂
def test_oauth_login_test(self): def test_oauth_login_test(self):
"""
测试OAuth登录流程
验证微博OAuth登录的跳转和授权流程
"""
# 创建微博OAuth配置
c = OAuthConfig() c = OAuthConfig()
c.type = 'weibo' c.type = 'weibo'
c.appkey = 'appkey' c.appkey = 'appkey'
c.appsecret = 'appsecret' c.appsecret = 'appsecret'
c.save() c.save()
# 测试OAuth登录请求应该重定向到微博授权页面
response = self.client.get('/oauth/oauthlogin?type=weibo') response = self.client.get('/oauth/oauthlogin?type=weibo')
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302) # 验证重定向状态码
self.assertTrue("api.weibo.com" in response.url) self.assertTrue("api.weibo.com" in response.url) # 验证跳转到微博授权页面
# 测试授权回调处理,应该重定向到首页
response = self.client.get('/oauth/authorize?type=weibo&code=code') response = self.client.get('/oauth/authorize?type=weibo&code=code')
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302) # 验证重定向状态码
self.assertEqual(response.url, '/') self.assertEqual(response.url, '/') # 验证跳转到首页
class OauthLoginTest(TestCase): class OauthLoginTest(TestCase):
"""
OAuth登录详细测试类
测试各种OAuth服务提供商的登录流程
"""
def setUp(self) -> None: def setUp(self) -> None:
self.client = Client() """
self.factory = RequestFactory() 测试初始化方法
self.apps = self.init_apps() """
self.client = Client() # Django测试客户端
self.factory = RequestFactory() # 请求工厂
self.apps = self.init_apps() # 初始化所有OAuth应用配置
def init_apps(self): def init_apps(self):
"""
初始化所有支持的OAuth应用配置
为每种OAuth服务创建测试配置
"""
# 获取所有OAuth管理器的子类实例
applications = [p() for p in BaseOauthManager.__subclasses__()] applications = [p() for p in BaseOauthManager.__subclasses__()]
for application in applications: for application in applications:
# 为每个OAuth服务创建配置
c = OAuthConfig() c = OAuthConfig()
c.type = application.ICON_NAME.lower() c.type = application.ICON_NAME.lower() # 服务类型(小写)
c.appkey = 'appkey' c.appkey = 'appkey' # 测试用的AppKey
c.appsecret = 'appsecret' c.appsecret = 'appsecret' # 测试用的AppSecret
c.save() c.save()
return applications return applications
def get_app_by_type(self, type): def get_app_by_type(self, type):
"""
根据类型获取对应的OAuth应用实例
Args:
type: OAuth服务类型
Returns:
对应的OAuth管理器实例
"""
for app in self.apps: for app in self.apps:
if app.ICON_NAME.lower() == type: if app.ICON_NAME.lower() == type:
return app return app
@ -57,73 +95,117 @@ class OauthLoginTest(TestCase):
@patch("oauth.oauthmanager.WBOauthManager.do_post") @patch("oauth.oauthmanager.WBOauthManager.do_post")
@patch("oauth.oauthmanager.WBOauthManager.do_get") @patch("oauth.oauthmanager.WBOauthManager.do_get")
def test_weibo_login(self, mock_do_get, mock_do_post): def test_weibo_login(self, mock_do_get, mock_do_post):
"""
测试微博OAuth登录流程
使用mock模拟API调用
"""
weibo_app = self.get_app_by_type('weibo') weibo_app = self.get_app_by_type('weibo')
assert weibo_app assert weibo_app # 确保获取到微博应用实例
# 获取授权URL
url = weibo_app.get_authorization_url() url = weibo_app.get_authorization_url()
# 设置mock返回值 - 获取access token
mock_do_post.return_value = json.dumps({"access_token": "access_token", mock_do_post.return_value = json.dumps({"access_token": "access_token",
"uid": "uid" "uid": "uid"
}) })
# 设置mock返回值 - 获取用户信息
mock_do_get.return_value = json.dumps({ mock_do_get.return_value = json.dumps({
"avatar_large": "avatar_large", "avatar_large": "avatar_large",
"screen_name": "screen_name", "screen_name": "screen_name",
"id": "id", "id": "id",
"email": "email", "email": "email",
}) })
# 执行获取access token的操作
userinfo = weibo_app.get_access_token_by_code('code') userinfo = weibo_app.get_access_token_by_code('code')
# 验证返回的用户信息
self.assertEqual(userinfo.token, 'access_token') self.assertEqual(userinfo.token, 'access_token')
self.assertEqual(userinfo.openid, 'id') self.assertEqual(userinfo.openid, 'id')
@patch("oauth.oauthmanager.GoogleOauthManager.do_post") @patch("oauth.oauthmanager.GoogleOauthManager.do_post")
@patch("oauth.oauthmanager.GoogleOauthManager.do_get") @patch("oauth.oauthmanager.GoogleOauthManager.do_get")
def test_google_login(self, mock_do_get, mock_do_post): def test_google_login(self, mock_do_get, mock_do_post):
"""
测试Google OAuth登录流程
"""
google_app = self.get_app_by_type('google') google_app = self.get_app_by_type('google')
assert google_app assert google_app
url = google_app.get_authorization_url() url = google_app.get_authorization_url()
# 模拟Google OAuth的token响应
mock_do_post.return_value = json.dumps({ mock_do_post.return_value = json.dumps({
"access_token": "access_token", "access_token": "access_token",
"id_token": "id_token", "id_token": "id_token",
}) })
# 模拟Google用户信息响应
mock_do_get.return_value = json.dumps({ mock_do_get.return_value = json.dumps({
"picture": "picture", "picture": "picture",
"name": "name", "name": "name",
"sub": "sub", "sub": "sub", # Google的用户ID字段
"email": "email", "email": "email",
}) })
token = google_app.get_access_token_by_code('code') token = google_app.get_access_token_by_code('code')
userinfo = google_app.get_oauth_userinfo() userinfo = google_app.get_oauth_userinfo()
# 验证用户信息
self.assertEqual(userinfo.token, 'access_token') self.assertEqual(userinfo.token, 'access_token')
self.assertEqual(userinfo.openid, 'sub') self.assertEqual(userinfo.openid, 'sub') # Google使用sub作为用户ID
@patch("oauth.oauthmanager.GitHubOauthManager.do_post") @patch("oauth.oauthmanager.GitHubOauthManager.do_post")
@patch("oauth.oauthmanager.GitHubOauthManager.do_get") @patch("oauth.oauthmanager.GitHubOauthManager.do_get")
def test_github_login(self, mock_do_get, mock_do_post): def test_github_login(self, mock_do_get, mock_do_post):
"""
测试GitHub OAuth登录流程
"""
github_app = self.get_app_by_type('github') github_app = self.get_app_by_type('github')
assert github_app assert github_app
url = github_app.get_authorization_url() url = github_app.get_authorization_url()
# 验证GitHub授权URL包含必要信息
self.assertTrue("github.com" in url) self.assertTrue("github.com" in url)
self.assertTrue("client_id" in url) self.assertTrue("client_id" in url)
# 模拟GitHub的token响应字符串格式
mock_do_post.return_value = "access_token=gho_16C7e42F292c6912E7710c838347Ae178B4a&scope=repo%2Cgist&token_type=bearer" mock_do_post.return_value = "access_token=gho_16C7e42F292c6912E7710c838347Ae178B4a&scope=repo%2Cgist&token_type=bearer"
# 模拟GitHub用户信息响应
mock_do_get.return_value = json.dumps({ mock_do_get.return_value = json.dumps({
"avatar_url": "avatar_url", "avatar_url": "avatar_url",
"name": "name", "name": "name",
"id": "id", "id": "id",
"email": "email", "email": "email",
}) })
token = github_app.get_access_token_by_code('code') token = github_app.get_access_token_by_code('code')
userinfo = github_app.get_oauth_userinfo() userinfo = github_app.get_oauth_userinfo()
# 验证用户信息
self.assertEqual(userinfo.token, 'gho_16C7e42F292c6912E7710c838347Ae178B4a') self.assertEqual(userinfo.token, 'gho_16C7e42F292c6912E7710c838347Ae178B4a')
self.assertEqual(userinfo.openid, 'id') self.assertEqual(userinfo.openid, 'id')
@patch("oauth.oauthmanager.FaceBookOauthManager.do_post") @patch("oauth.oauthmanager.FaceBookOauthManager.do_post")
@patch("oauth.oauthmanager.FaceBookOauthManager.do_get") @patch("oauth.oauthmanager.FaceBookOauthManager.do_get")
def test_facebook_login(self, mock_do_get, mock_do_post): def test_facebook_login(self, mock_do_get, mock_do_post):
"""
测试Facebook OAuth登录流程
"""
facebook_app = self.get_app_by_type('facebook') facebook_app = self.get_app_by_type('facebook')
assert facebook_app assert facebook_app
url = facebook_app.get_authorization_url() url = facebook_app.get_authorization_url()
self.assertTrue("facebook.com" in url) self.assertTrue("facebook.com" in url) # 验证Facebook授权URL
# 模拟Facebook token响应
mock_do_post.return_value = json.dumps({ mock_do_post.return_value = json.dumps({
"access_token": "access_token", "access_token": "access_token",
}) })
# 模拟Facebook用户信息响应嵌套结构
mock_do_get.return_value = json.dumps({ mock_do_get.return_value = json.dumps({
"name": "name", "name": "name",
"id": "id", "id": "id",
@ -134,14 +216,16 @@ class OauthLoginTest(TestCase):
} }
} }
}) })
token = facebook_app.get_access_token_by_code('code') token = facebook_app.get_access_token_by_code('code')
userinfo = facebook_app.get_oauth_userinfo() userinfo = facebook_app.get_oauth_userinfo()
self.assertEqual(userinfo.token, 'access_token') self.assertEqual(userinfo.token, 'access_token')
@patch("oauth.oauthmanager.QQOauthManager.do_get", side_effect=[ @patch("oauth.oauthmanager.QQOauthManager.do_get", side_effect=[
'access_token=access_token&expires_in=3600', 'access_token=access_token&expires_in=3600', # 第一次调用获取token
'callback({"client_id":"appid","openid":"openid"} );', 'callback({"client_id":"appid","openid":"openid"} );', # 第二次调用获取openid
json.dumps({ json.dumps({ # 第三次调用:获取用户信息
"nickname": "nickname", "nickname": "nickname",
"email": "email", "email": "email",
"figureurl": "figureurl", "figureurl": "figureurl",
@ -149,21 +233,33 @@ class OauthLoginTest(TestCase):
}) })
]) ])
def test_qq_login(self, mock_do_get): def test_qq_login(self, mock_do_get):
"""
测试QQ OAuth登录流程
使用side_effect模拟多次不同的API响应
"""
qq_app = self.get_app_by_type('qq') qq_app = self.get_app_by_type('qq')
assert qq_app assert qq_app
url = qq_app.get_authorization_url() url = qq_app.get_authorization_url()
self.assertTrue("qq.com" in url) self.assertTrue("qq.com" in url) # 验证QQ授权URL
token = qq_app.get_access_token_by_code('code') token = qq_app.get_access_token_by_code('code')
userinfo = qq_app.get_oauth_userinfo() userinfo = qq_app.get_oauth_userinfo()
self.assertEqual(userinfo.token, 'access_token') self.assertEqual(userinfo.token, 'access_token')
@patch("oauth.oauthmanager.WBOauthManager.do_post") @patch("oauth.oauthmanager.WBOauthManager.do_post")
@patch("oauth.oauthmanager.WBOauthManager.do_get") @patch("oauth.oauthmanager.WBOauthManager.do_get")
def test_weibo_authoriz_login_with_email(self, mock_do_get, mock_do_post): def test_weibo_authoriz_login_with_email(self, mock_do_get, mock_do_post):
"""
测试包含邮箱的微博授权登录完整流程
验证用户认证和会话管理
"""
# 模拟获取access token
mock_do_post.return_value = json.dumps({"access_token": "access_token", mock_do_post.return_value = json.dumps({"access_token": "access_token",
"uid": "uid" "uid": "uid"
}) })
# 模拟用户信息(包含邮箱)
mock_user_info = { mock_user_info = {
"avatar_large": "avatar_large", "avatar_large": "avatar_large",
"screen_name": "screen_name1", "screen_name": "screen_name1",
@ -172,25 +268,31 @@ class OauthLoginTest(TestCase):
} }
mock_do_get.return_value = json.dumps(mock_user_info) mock_do_get.return_value = json.dumps(mock_user_info)
# 测试登录跳转
response = self.client.get('/oauth/oauthlogin?type=weibo') response = self.client.get('/oauth/oauthlogin?type=weibo')
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302)
self.assertTrue("api.weibo.com" in response.url) self.assertTrue("api.weibo.com" in response.url)
# 测试授权回调
response = self.client.get('/oauth/authorize?type=weibo&code=code') response = self.client.get('/oauth/authorize?type=weibo&code=code')
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, '/') self.assertEqual(response.url, '/')
# 验证用户认证状态
user = auth.get_user(self.client) user = auth.get_user(self.client)
assert user.is_authenticated assert user.is_authenticated
self.assertTrue(user.is_authenticated) self.assertTrue(user.is_authenticated)
self.assertEqual(user.username, mock_user_info['screen_name']) self.assertEqual(user.username, mock_user_info['screen_name'])
self.assertEqual(user.email, mock_user_info['email']) self.assertEqual(user.email, mock_user_info['email'])
# 登出后再次测试
self.client.logout() self.client.logout()
response = self.client.get('/oauth/authorize?type=weibo&code=code') response = self.client.get('/oauth/authorize?type=weibo&code=code')
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, '/') self.assertEqual(response.url, '/')
# 再次验证用户认证状态
user = auth.get_user(self.client) user = auth.get_user(self.client)
assert user.is_authenticated assert user.is_authenticated
self.assertTrue(user.is_authenticated) self.assertTrue(user.is_authenticated)
@ -200,10 +302,15 @@ class OauthLoginTest(TestCase):
@patch("oauth.oauthmanager.WBOauthManager.do_post") @patch("oauth.oauthmanager.WBOauthManager.do_post")
@patch("oauth.oauthmanager.WBOauthManager.do_get") @patch("oauth.oauthmanager.WBOauthManager.do_get")
def test_weibo_authoriz_login_without_email(self, mock_do_get, mock_do_post): def test_weibo_authoriz_login_without_email(self, mock_do_get, mock_do_post):
"""
测试不包含邮箱的微博授权登录流程
验证邮箱补充流程
"""
# 模拟获取access token
mock_do_post.return_value = json.dumps({"access_token": "access_token", mock_do_post.return_value = json.dumps({"access_token": "access_token",
"uid": "uid" "uid": "uid"
}) })
# 模拟用户信息(不包含邮箱)
mock_user_info = { mock_user_info = {
"avatar_large": "avatar_large", "avatar_large": "avatar_large",
"screen_name": "screen_name1", "screen_name": "screen_name1",
@ -211,28 +318,34 @@ class OauthLoginTest(TestCase):
} }
mock_do_get.return_value = json.dumps(mock_user_info) mock_do_get.return_value = json.dumps(mock_user_info)
# 测试登录跳转
response = self.client.get('/oauth/oauthlogin?type=weibo') response = self.client.get('/oauth/oauthlogin?type=weibo')
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302)
self.assertTrue("api.weibo.com" in response.url) self.assertTrue("api.weibo.com" in response.url)
# 测试授权回调 - 应该重定向到邮箱补充页面
response = self.client.get('/oauth/authorize?type=weibo&code=code') response = self.client.get('/oauth/authorize?type=weibo&code=code')
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302)
# 解析OAuth用户ID
oauth_user_id = int(response.url.split('/')[-1].split('.')[0]) oauth_user_id = int(response.url.split('/')[-1].split('.')[0])
self.assertEqual(response.url, f'/oauth/requireemail/{oauth_user_id}.html') self.assertEqual(response.url, f'/oauth/requireemail/{oauth_user_id}.html')
# 提交邮箱信息
response = self.client.post(response.url, {'email': 'test@gmail.com', 'oauthid': oauth_user_id}) response = self.client.post(response.url, {'email': 'test@gmail.com', 'oauthid': oauth_user_id})
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302)
# 生成邮箱验证签名
sign = get_sha256(settings.SECRET_KEY + sign = get_sha256(settings.SECRET_KEY +
str(oauth_user_id) + settings.SECRET_KEY) str(oauth_user_id) + settings.SECRET_KEY)
# 验证绑定成功URL
url = reverse('oauth:bindsuccess', kwargs={ url = reverse('oauth:bindsuccess', kwargs={
'oauthid': oauth_user_id, 'oauthid': oauth_user_id,
}) })
self.assertEqual(response.url, f'{url}?type=email') self.assertEqual(response.url, f'{url}?type=email')
# 验证邮箱确认流程
path = reverse('oauth:email_confirm', kwargs={ path = reverse('oauth:email_confirm', kwargs={
'id': oauth_user_id, 'id': oauth_user_id,
'sign': sign 'sign': sign
@ -240,10 +353,12 @@ class OauthLoginTest(TestCase):
response = self.client.get(path) response = self.client.get(path)
self.assertEqual(response.status_code, 302) self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, f'/oauth/bindsuccess/{oauth_user_id}.html?type=success') self.assertEqual(response.url, f'/oauth/bindsuccess/{oauth_user_id}.html?type=success')
# 验证最终用户状态
user = auth.get_user(self.client) user = auth.get_user(self.client)
from oauth.models import OAuthUser from oauth.models import OAuthUser
oauth_user = OAuthUser.objects.get(author=user) oauth_user = OAuthUser.objects.get(author=user)
self.assertTrue(user.is_authenticated) self.assertTrue(user.is_authenticated)
self.assertEqual(user.username, mock_user_info['screen_name']) self.assertEqual(user.username, mock_user_info['screen_name'])
self.assertEqual(user.email, 'test@gmail.com') self.assertEqual(user.email, 'test@gmail.com')
self.assertEqual(oauth_user.pk, oauth_user_id) self.assertEqual(oauth_user.pk, oauth_user_id)
Loading…
Cancel
Save