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