汤应雪 3 months ago
commit bfa0f5034c

@ -1,294 +0,0 @@
# 导入Django测试客户端、请求工厂、测试用例
from django.test import Client, RequestFactory, TestCase
# 导入URL反向解析
from django.urls import reverse
# 导入时区工具
from django.utils import timezone
# 导入延迟翻译函数
from django.utils.translation import gettext_lazy as _
# 导入账户模型
from accounts.models import BlogUser
# 导入博客模型
from blog.models import Article, Category
# 导入项目工具函数
from djangoblog.utils import *
# 导入当前应用的工具函数
from . import utils
# 在这里创建测试
class AccountTest(TestCase):
"""账户功能测试类"""
def setUp(self):
"""测试前置设置,每个测试方法执行前都会调用"""
# 创建测试客户端
self.client = Client()
# 创建请求工厂
self.factory = RequestFactory()
# 创建测试用户
self.blog_user = BlogUser.objects.create_user(
username="test", # 用户名
email="admin@admin.com", # 邮箱
password="12345678" # 密码
)
# 设置新测试密码
self.new_test = "xxx123--="
def test_validate_account(self):
"""测试账户验证功能"""
# 获取当前站点
site = get_current_site().domain
# 创建超级用户
user = BlogUser.objects.create_superuser(
email="liangliangyy1@gmail.com", # 邮箱
username="liangliangyy1", # 用户名
password="qwer!@#$ggg" # 密码
)
# 从数据库获取刚创建的用户
testuser = BlogUser.objects.get(username='liangliangyy1')
# 尝试登录
loginresult = self.client.login(
username='liangliangyy1', # 用户名
password='qwer!@#$ggg' # 密码
)
# 断言登录成功
self.assertEqual(loginresult, True)
# 访问管理员页面
response = self.client.get('/admin/')
# 断言页面访问成功
self.assertEqual(response.status_code, 200)
# 创建分类
category = Category()
category.name = "categoryaaa" # 分类名称
category.creation_time = timezone.now() # 创建时间
category.last_modify_time = timezone.now() # 最后修改时间
category.save()
# 创建文章
article = Article()
article.title = "nicetitleaaa" # 文章标题
article.body = "nicecontentaaa" # 文章内容
article.author = user # 文章作者
article.category = category # 文章分类
article.type = 'a' # 文章类型
article.status = 'p' # 文章状态:发布
article.save()
# 访问文章管理页面
response = self.client.get(article.get_admin_url())
# 断言页面访问成功
self.assertEqual(response.status_code, 200)
def test_validate_register(self):
"""测试用户注册功能"""
# 断言注册前邮箱不存在
self.assertEquals(
0, len(BlogUser.objects.filter(email='user123@user.com')))
# 发送注册POST请求
response = self.client.post(reverse('account:register'), {
'username': 'user1233', # 用户名
'email': 'user123@user.com', # 邮箱
'password1': 'password123!q@wE#R$T', # 密码
'password2': 'password123!q@wE#R$T', # 确认密码
})
# 断言注册后邮箱存在
self.assertEquals(
1, len(BlogUser.objects.filter(email='user123@user.com')))
# 获取刚注册的用户
user = BlogUser.objects.filter(email='user123@user.com')[0]
# 生成验证签名
sign = get_sha256(get_sha256(settings.SECRET_KEY + str(user.id)))
# 构建验证URL路径
path = reverse('accounts:result')
# 构建完整验证URL
url = '{path}?type=validation&id={id}&sign={sign}'.format(
path=path, # 路径
id=user.id, # 用户ID
sign=sign # 签名
)
# 访问验证URL
response = self.client.get(url)
# 断言页面访问成功
self.assertEqual(response.status_code, 200)
# 登录用户
self.client.login(username='user1233', password='password123!q@wE#R$T')
# 获取用户并设置为管理员
user = BlogUser.objects.filter(email='user123@user.com')[0]
user.is_superuser = True # 设置为超级用户
user.is_staff = True # 设置为工作人员
user.save()
# 删除侧边栏缓存
delete_sidebar_cache()
# 创建分类
category = Category()
category.name = "categoryaaa" # 分类名称
category.creation_time = timezone.now() # 创建时间
category.last_modify_time = timezone.now() # 最后修改时间
category.save()
# 创建文章
article = Article()
article.category = category # 文章分类
article.title = "nicetitle333" # 文章标题
article.body = "nicecontentttt" # 文章内容
article.author = user # 文章作者
article.type = 'a' # 文章类型
article.status = 'p' # 文章状态:发布
article.save()
# 访问文章管理页面
response = self.client.get(article.get_admin_url())
# 断言页面访问成功
self.assertEqual(response.status_code, 200)
# 登出用户
response = self.client.get(reverse('account:logout'))
# 断言登出成功(重定向状态码)
self.assertIn(response.status_code, [301, 302, 200])
# 再次访问文章管理页面(应该被重定向到登录页)
response = self.client.get(article.get_admin_url())
# 断言被重定向
self.assertIn(response.status_code, [301, 302, 200])
# 使用错误密码尝试登录
response = self.client.post(reverse('account:login'), {
'username': 'user1233', # 用户名
'password': 'password123' # 错误密码
})
# 断言登录失败(重定向状态码)
self.assertIn(response.status_code, [301, 302, 200])
# 再次访问文章管理页面(应该仍然被重定向)
response = self.client.get(article.get_admin_url())
# 断言被重定向
self.assertIn(response.status_code, [301, 302, 200])
def test_verify_email_code(self):
"""测试邮箱验证码功能"""
# 测试邮箱
to_email = "admin@admin.com"
# 生成验证码
code = generate_code()
# 设置验证码到缓存
utils.set_code(to_email, code)
# 发送验证邮件
utils.send_verify_email(to_email, code)
# 验证正确验证码
err = utils.verify("admin@admin.com", code)
# 断言验证成功返回None
self.assertEqual(err, None)
# 验证错误验证码
err = utils.verify("admin@123.com", code)
# 断言验证失败(返回错误信息字符串)
self.assertEqual(type(err), str)
def test_forget_password_email_code_success(self):
"""测试成功获取忘记密码验证码"""
# 发送获取验证码的POST请求
resp = self.client.post(
path=reverse("account:forget_password_code"), # URL路径
data=dict(email="admin@admin.com") # 请求数据:邮箱
)
# 断言响应状态码为200
self.assertEqual(resp.status_code, 200)
# 断言响应内容为"ok"
self.assertEqual(resp.content.decode("utf-8"), "ok")
def test_forget_password_email_code_fail(self):
"""测试获取忘记密码验证码失败情况"""
# 发送空数据的POST请求
resp = self.client.post(
path=reverse("account:forget_password_code"), # URL路径
data=dict() # 空数据
)
# 断言返回错误信息
self.assertEqual(resp.content.decode("utf-8"), "错误的邮箱")
# 发送无效邮箱的POST请求
resp = self.client.post(
path=reverse("account:forget_password_code"), # URL路径
data=dict(email="admin@com") # 无效邮箱格式
)
# 断言返回错误信息
self.assertEqual(resp.content.decode("utf-8"), "错误的邮箱")
def test_forget_password_email_success(self):
"""测试成功重置密码"""
# 生成验证码
code = generate_code()
# 设置验证码到缓存
utils.set_code(self.blog_user.email, code)
# 准备请求数据
data = dict(
new_password1=self.new_test, # 新密码
new_password2=self.new_test, # 确认密码
email=self.blog_user.email, # 邮箱
code=code, # 验证码
)
# 发送重置密码的POST请求
resp = self.client.post(
path=reverse("account:forget_password"), # URL路径
data=data # 请求数据
)
# 断言重定向响应状态码302
self.assertEqual(resp.status_code, 302)
# 验证用户密码是否修改成功
blog_user = BlogUser.objects.filter(
email=self.blog_user.email,
).first() # 类型注解BlogUser
# 断言用户存在
self.assertNotEqual(blog_user, None)
# 断言新密码验证通过
self.assertEqual(blog_user.check_password(data["new_password1"]), True)
def test_forget_password_email_not_user(self):
"""测试重置密码时邮箱不存在的情况"""
# 准备请求数据(不存在的邮箱)
data = dict(
new_password1=self.new_test, # 新密码
new_password2=self.new_test, # 确认密码
email="123@123.com", # 不存在的邮箱
code="123456", # 验证码
)
# 发送重置密码的POST请求
resp = self.client.post(
path=reverse("account:forget_password"), # URL路径
data=data # 请求数据
)
# 断言返回表单页面状态码200
self.assertEqual(resp.status_code, 200)
def test_forget_password_email_code_error(self):
"""测试重置密码时验证码错误的情况"""
# 生成验证码
code = generate_code()
# 设置验证码到缓存
utils.set_code(self.blog_user.email, code)
# 准备请求数据(错误的验证码)
data = dict(
new_password1=self.new_test, # 新密码
new_password2=self.new_test, # 确认密码
email=self.blog_user.email, # 邮箱
code="111111", # 错误的验证码
)
# 发送重置密码的POST请求
resp = self.client.post(
path=reverse("account:forget_password"), # URL路径
data=data # 请求数据
)
# 断言返回表单页面状态码200
self.assertEqual(resp.status_code, 200)

@ -1,53 +0,0 @@
# 导入URL路径函数
from django.urls import path
# 导入正则URL路径函数兼容老版本
from django.urls import re_path
# 导入当前应用的视图
from . import views
# 导入登录表单
from .forms import LoginForm
# 应用命名空间用于URL反向解析
app_name = "accounts"
# URL模式列表
urlpatterns = [
# 登录URL - 使用正则表达式匹配 /login/ 路径
re_path(r'^login/$',
# 使用LoginView视图类登录成功后重定向到首页
views.LoginView.as_view(success_url='/'),
name='login', # URL名称login
# 传入额外参数:指定认证表单类
kwargs={'authentication_form': LoginForm}),
# 注册URL - 使用正则表达式匹配 /register/ 路径
re_path(r'^register/$',
# 使用RegisterView视图类注册成功后重定向到首页
views.RegisterView.as_view(success_url="/"),
name='register'), # URL名称register
# 登出URL - 使用正则表达式匹配 /logout/ 路径
re_path(r'^logout/$',
# 使用LogoutView视图类
views.LogoutView.as_view(),
name='logout'), # URL名称logout
# 账户结果页面URL - 使用path匹配固定路径
path(r'account/result.html',
# 使用account_result函数视图
views.account_result,
name='result'), # URL名称result
# 忘记密码URL - 使用正则表达式匹配 /forget_password/ 路径
re_path(r'^forget_password/$',
# 使用ForgetPasswordView视图类
views.ForgetPasswordView.as_view(),
name='forget_password'), # URL名称forget_password
# 获取忘记密码验证码URL - 使用正则表达式匹配 /forget_password_code/ 路径
re_path(r'^forget_password_code/$',
# 使用ForgetPasswordEmailCode视图类
views.ForgetPasswordEmailCode.as_view(),
name='forget_password_code'), # URL名称forget_password_code
]

@ -1,42 +0,0 @@
# 导入获取用户模型的函数
from django.contrib.auth import get_user_model
# 导入模型后端认证基类
from django.contrib.auth.backends import ModelBackend
class EmailOrUsernameModelBackend(ModelBackend):
"""
自定义认证后端允许使用用户名或邮箱登录
继承自Django的ModelBackend
"""
def authenticate(self, request, username=None, password=None, **kwargs):
"""用户认证方法"""
# 检查用户名中是否包含@符号(判断是否为邮箱)
if '@' in username:
# 如果是邮箱,设置查询参数为邮箱
kwargs = {'email': username}
else:
# 如果是用户名,设置查询参数为用户名
kwargs = {'username': username}
try:
# 根据查询参数获取用户对象
user = get_user_model().objects.get(**kwargs)
# 检查密码是否正确
if user.check_password(password):
# 密码正确,返回用户对象
return user
# 捕获用户不存在的异常
except get_user_model().DoesNotExist:
# 用户不存在返回None
return None
def get_user(self, username):
"""根据用户ID获取用户对象"""
try:
# 根据主键用户ID获取用户对象
return get_user_model().objects.get(pk=username)
# 捕获用户不存在的异常
except get_user_model().DoesNotExist:
# 用户不存在返回None
return None

@ -1,70 +0,0 @@
# 导入类型提示模块
import typing
# 导入时间间隔类
from datetime import timedelta
# 导入Django缓存框架
from django.core.cache import cache
# 导入国际化翻译函数
from django.utils.translation import gettext
# 导入延迟翻译函数
from django.utils.translation import gettext_lazy as _
# 导入发送邮件的工具函数
from djangoblog.utils import send_email
# 验证码有效期5分钟
_code_ttl = timedelta(minutes=5)
def send_verify_email(to_mail: str, code: str, subject: str = _("Verify Email")):
"""发送验证邮件,用于密码重置等场景
Args:
to_mail: 接收邮件的邮箱地址
subject: 邮件主题默认为"验证邮箱"
code: 验证码内容
"""
# 构建邮件HTML内容包含验证码信息
html_content = _(
# 翻译文本:您正在重置密码,验证码是:{code}5分钟内有效请妥善保管
"You are resetting the password, the verification code is%(code)s, valid within 5 minutes, please keep it "
"properly"
) % {'code': code} # 将code插入到格式化字符串中
# 调用发送邮件函数
send_email([to_mail], subject, html_content)
def verify(email: str, code: str) -> typing.Optional[str]:
"""验证验证码是否有效
Args:
email: 请求验证的邮箱地址
code: 用户输入的验证码
Return:
如果验证失败返回错误信息字符串验证成功返回None
Note:
这里的错误处理不太合理应该采用raise抛出异常
否则调用方也需要对error进行处理
"""
# 从缓存中获取该邮箱对应的验证码
cache_code = get_code(email)
# 比较缓存中的验证码和用户输入的验证码
if cache_code != code:
# 如果不匹配,返回错误信息
return gettext("Verification code error")
# 验证成功返回None
def set_code(email: str, code: str):
"""将验证码设置到缓存中"""
# 使用cache.set方法key为邮箱value为验证码设置过期时间
cache.set(email, code, _code_ttl.seconds)
def get_code(email: str) -> typing.Optional[str]:
"""从缓存中获取验证码"""
# 使用cache.get方法根据邮箱获取验证码
return cache.get(email)

@ -1,327 +0,0 @@
# 导入日志模块
import logging
# 导入延迟翻译函数
from django.utils.translation import gettext_lazy as _
# 导入Django设置
from django.conf import settings
# 导入Django认证框架
from django.contrib import auth
# 导入重定向字段名常量
from django.contrib.auth import REDIRECT_FIELD_NAME
# 导入获取用户模型的函数
from django.contrib.auth import get_user_model
# 导入登出函数
from django.contrib.auth import logout
# 导入Django内置认证表单
from django.contrib.auth.forms import AuthenticationForm
# 导入密码哈希函数
from django.contrib.auth.hashers import make_password
# 导入HTTP响应重定向和禁止访问响应
from django.http import HttpResponseRedirect, HttpResponseForbidden
# 导入HTTP请求类型
from django.http.request import HttpRequest
# 导入HTTP响应类型
from django.http.response import HttpResponse
# 导入快捷函数获取对象或404错误
from django.shortcuts import get_object_or_404
# 导入快捷函数:渲染模板
from django.shortcuts import render
# 导入URL反向解析
from django.urls import reverse
# 导入方法装饰器
from django.utils.decorators import method_decorator
# 导入URL安全验证函数
from django.utils.http import url_has_allowed_host_and_scheme
# 导入基于类的视图基类
from django.views import View
# 导入禁止缓存装饰器
from django.views.decorators.cache import never_cache
# 导入CSRF保护装饰器
from django.views.decorators.csrf import csrf_protect
# 导入敏感参数保护装饰器
from django.views.decorators.debug import sensitive_post_parameters
# 导入通用视图类
from django.views.generic import FormView, RedirectView
# 导入项目工具函数
from djangoblog.utils import send_email, get_sha256, get_current_site, generate_code, delete_sidebar_cache
# 导入当前应用的工具函数
from . import utils
# 导入自定义表单
from .forms import RegisterForm, LoginForm, ForgetPasswordForm, ForgetPasswordCodeForm
# 导入用户模型
from .models import BlogUser
# 获取当前模块的日志器
logger = logging.getLogger(__name__)
# 在这里创建视图
class RegisterView(FormView):
"""用户注册视图"""
# 指定使用的表单类
form_class = RegisterForm
# 指定使用的模板
template_name = 'account/registration_form.html'
@method_decorator(csrf_protect) # CSRF保护装饰器
def dispatch(self, *args, **kwargs):
"""处理请求分发"""
# 调用父类的dispatch方法
return super(RegisterView, self).dispatch(*args, **kwargs)
def form_valid(self, form):
"""处理表单验证通过的情况"""
# 检查表单是否有效
if form.is_valid():
# 保存表单数据但不提交到数据库commit=False
user = form.save(False)
# 设置用户为未激活状态(需要邮箱验证)
user.is_active = False
# 设置用户来源为注册页面
user.source = 'Register'
# 保存用户到数据库
user.save(True)
# 获取当前站点域名
site = get_current_site().domain
# 生成验证签名:对密钥+用户ID进行双重SHA256哈希
sign = get_sha256(get_sha256(settings.SECRET_KEY + str(user.id)))
# 如果是调试模式,使用本地地址
if settings.DEBUG:
site = '127.0.0.1:8000'
# 获取结果页面的URL路径
path = reverse('account:result')
# 构建完整的验证URL
url = "http://{site}{path}?type=validation&id={id}&sign={sign}".format(
site=site, # 站点地址
path=path, # 结果页面路径
id=user.id, # 用户ID
sign=sign # 验证签名
)
# 构建邮件内容
content = """
<p>请点击下面链接验证您的邮箱</p>
<a href="{url}" rel="bookmark">{url}</a>
再次感谢您
<br />
如果上面链接无法打开请将此链接复制至浏览器
{url}
""".format(url=url)
# 发送验证邮件
send_email(
emailto=[user.email], # 收件人邮箱
title='验证您的电子邮箱', # 邮件标题
content=content # 邮件内容
)
# 构建注册结果页面URL
url = reverse('accounts:result') + '?type=register&id=' + str(user.id)
# 重定向到结果页面
return HttpResponseRedirect(url)
else:
# 表单无效,重新渲染表单并显示错误
return self.render_to_response({'form': form})
class LogoutView(RedirectView):
"""用户登出视图"""
# 登出后重定向的URL
url = '/login/'
@method_decorator(never_cache) # 禁止缓存装饰器
def dispatch(self, request, *args, **kwargs):
"""处理请求分发"""
# 调用父类的dispatch方法
return super(LogoutView, self).dispatch(request, *args, **kwargs)
def get(self, request, *args, **kwargs):
"""处理GET请求登出操作"""
# 调用Django登出函数
logout(request)
# 删除侧边栏缓存
delete_sidebar_cache()
# 调用父类的get方法进行重定向
return super(LogoutView, self).get(request, *args, **kwargs)
class LoginView(FormView):
"""用户登录视图"""
# 指定使用的表单类
form_class = LoginForm
# 指定使用的模板
template_name = 'account/login.html'
# 登录成功后的重定向URL
success_url = '/'
# 重定向字段名
redirect_field_name = REDIRECT_FIELD_NAME
# 登录会话保持时间:一个月(秒数)
login_ttl = 2626560
# 方法装饰器保护敏感参数、CSRF保护、禁止缓存
@method_decorator(sensitive_post_parameters('password'))
@method_decorator(csrf_protect)
@method_decorator(never_cache)
def dispatch(self, request, *args, **kwargs):
"""处理请求分发"""
# 调用父类的dispatch方法
return super(LoginView, self).dispatch(request, *args, **kwargs)
def get_context_data(self, **kwargs):
"""获取模板上下文数据"""
# 从GET参数中获取重定向URL
redirect_to = self.request.GET.get(self.redirect_field_name)
# 如果重定向URL为空设置为首页
if redirect_to is None:
redirect_to = '/'
# 将重定向URL添加到上下文数据中
kwargs['redirect_to'] = redirect_to
# 调用父类方法获取基础上下文数据
return super(LoginView, self).get_context_data(**kwargs)
def form_valid(self, form):
"""处理表单验证通过的情况"""
# 创建认证表单实例使用POST数据
form = AuthenticationForm(data=self.request.POST, request=self.request)
# 检查表单是否有效
if form.is_valid():
# 删除侧边栏缓存
delete_sidebar_cache()
# 记录日志
logger.info(self.redirect_field_name)
# 登录用户
auth.login(self.request, form.get_user())
# 如果用户选择了"记住我"
if self.request.POST.get("remember"):
# 设置会话过期时间为一个月
self.request.session.set_expiry(self.login_ttl)
# 调用父类的form_valid方法会处理重定向
return super(LoginView, self).form_valid(form)
else:
# 表单无效,重新渲染表单并显示错误
return self.render_to_response({'form': form})
def get_success_url(self):
"""获取登录成功后的重定向URL"""
# 从POST数据中获取重定向URL
redirect_to = self.request.POST.get(self.redirect_field_name)
# 验证重定向URL是否安全同源策略
if not url_has_allowed_host_and_scheme(
url=redirect_to,
allowed_hosts=[self.request.get_host()]):
# 如果不安全使用默认的成功URL
redirect_to = self.success_url
# 返回重定向URL
return redirect_to
def account_result(request):
"""账户操作结果页面视图函数"""
# 从GET参数获取操作类型
type = request.GET.get('type')
# 从GET参数获取用户ID
id = request.GET.get('id')
# 根据ID获取用户对象如果不存在返回404
user = get_object_or_404(get_user_model(), id=id)
# 记录日志
logger.info(type)
# 如果用户已激活,重定向到首页
if user.is_active:
return HttpResponseRedirect('/')
# 检查操作类型是否为注册或验证
if type and type in ['register', 'validation']:
# 如果是注册操作
if type == 'register':
# 设置注册成功的内容和标题
content = '''
恭喜您注册成功一封验证邮件已经发送到您的邮箱请验证您的邮箱后登录本站
'''
title = '注册成功'
else:
# 生成验证签名
c_sign = get_sha256(get_sha256(settings.SECRET_KEY + str(user.id)))
# 从GET参数获取签名
sign = request.GET.get('sign')
# 验证签名是否正确
if sign != c_sign:
# 签名错误,返回禁止访问
return HttpResponseForbidden()
# 激活用户账户
user.is_active = True
# 保存用户
user.save()
# 设置验证成功的内容和标题
content = '''
恭喜您已经成功的完成邮箱验证您现在可以使用您的账号来登录本站
'''
title = '验证成功'
# 渲染结果页面
return render(request, 'account/result.html', {
'title': title, # 页面标题
'content': content # 页面内容
})
else:
# 无效的操作类型,重定向到首页
return HttpResponseRedirect('/')
class ForgetPasswordView(FormView):
"""忘记密码重置视图"""
# 指定使用的表单类
form_class = ForgetPasswordForm
# 指定使用的模板
template_name = 'account/forget_password.html'
def form_valid(self, form):
"""处理表单验证通过的情况"""
# 检查表单是否有效
if form.is_valid():
# 根据邮箱获取用户对象
blog_user = BlogUser.objects.filter(
email=form.cleaned_data.get("email")
).get()
# 使用新密码的哈希值更新用户密码
blog_user.password = make_password(form.cleaned_data["new_password2"])
# 保存用户
blog_user.save()
# 重定向到登录页面
return HttpResponseRedirect('/login/')
else:
# 表单无效,重新渲染表单并显示错误
return self.render_to_response({'form': form})
class ForgetPasswordEmailCode(View):
"""获取忘记密码验证码的API视图"""
def post(self, request: HttpRequest):
"""处理POST请求"""
# 创建表单实例并验证数据
form = ForgetPasswordCodeForm(request.POST)
# 检查表单是否有效
if not form.is_valid():
# 表单无效,返回错误响应
return HttpResponse("错误的邮箱")
# 从已验证数据中获取邮箱
to_email = form.cleaned_data["email"]
# 生成验证码
code = generate_code()
# 发送验证邮件
utils.send_verify_email(to_email, code)
# 将验证码保存到缓存
utils.set_code(to_email, code)
# 返回成功响应
return HttpResponse("ok")
Loading…
Cancel
Save