From f6f5a8af7e6c5359e13d7c75172adb0bc6cac555 Mon Sep 17 00:00:00 2001
From: hyb <kk_huangyangbo@163.com>
Date: Fri, 23 Jan 2026 01:53:11 +0000
Subject: [PATCH] feat: 添加用户登录功能、添加注册功能、注册审批功能，参考：注册审批功能测试指南.md、驱动代码模块增加插入其他项目代码功能，驱动代码模块增加插入其他项目代码功能，优化测试报告模块前端布局和美化 - 实现超级用户管理系统用户，可进行增删改查，其他用户只可查看本人的数据，不可进行增删改 - 将Django后台管理中的用户配置相关功能迁移到系统中进行适配、可配置禁用、是否管理员、是否超级用户 - 增加用户管理的分组字段、可直接进行分组操作，实现用户管理即可通过分组配置实现各项目的访问权限 - 实现登录页面注册 - 实现管理员首页点击注册用户审批可实现通过不通过 - 实现未审核用户的登录提示与审核未通过的用户登录提示 - 实现一键插入其他本人可访问项目的脚本代码
---
测试组/Test_platform/Interface_automation/backend/apps/lunaruser/views.py | 236 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 230 insertions(+), 6 deletions(-)
diff --git "a/\346\265\213\350\257\225\347\273\204/Test_platform/Interface_automation/backend/apps/lunaruser/views.py" "b/\346\265\213\350\257\225\347\273\204/Test_platform/Interface_automation/backend/apps/lunaruser/views.py"
index 4d6fa9b..c75e23e 100644
--- "a/\346\265\213\350\257\225\347\273\204/Test_platform/Interface_automation/backend/apps/lunaruser/views.py"
+++ "b/\346\265\213\350\257\225\347\273\204/Test_platform/Interface_automation/backend/apps/lunaruser/views.py"
@@ -4,11 +4,15 @@
from rest_framework_jwt.settings import api_settings
from rest_framework.response import Response
from rest_framework.views import APIView
-from rest_framework.viewsets import GenericViewSet
+from rest_framework.viewsets import GenericViewSet, ModelViewSet
+from rest_framework.decorators import action
from drf_yasg.utils import swagger_auto_schema
from rest_framework import status
from rest_framework.permissions import IsAuthenticated
+from rest_framework.filters import SearchFilter
from backend.utils.auth import MyJWTAuthentication
+from backend.utils.permissions import CustomIsAdminUser
+import logging
from backend.utils.request_util import save_login_log
from lunarlink.models import LoginLog
@@ -43,16 +47,33 @@
except KeyError:
return Response(response.KEY_MISS)
- user = authenticate(username=username, password=password)
-
- if user is None:
+ try:
+ user = User.objects.get(username=username)
+ except User.DoesNotExist:
return Response(response.LOGIN_FAILED)
- # 0后面还需要优化定义明确
+ if not user.check_password(password):
+ return Response(response.LOGIN_FAILED)
+
+ if hasattr(user, 'status') and user.status == 'pending':
+ return Response({
+ "success": False,
+ "msg": "你提交的注册还在审批中，暂时无法登录，可联系管理员确认！"
+ }, status=status.HTTP_403_FORBIDDEN)
+
+ if hasattr(user, 'status') and user.status == 'rejected':
+ return Response({
+ "success": False,
+ "msg": "你的注册申请未通过审核，无法登录！"
+ }, status=status.HTTP_403_FORBIDDEN)
+
if user.is_active == 0:
return Response(response.USER_BLOCKED)
- # JWT token creation
+ from django.utils import timezone
+ user.last_login = timezone.now()
+ user.save(update_fields=['last_login'])
+
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
@@ -65,6 +86,7 @@
"user": user.username,
"name": user.name,
"is_superuser": user.is_superuser,
+ "is_staff": user.is_staff,
"show_hosts": user.show_hosts,
"token": token,
}
@@ -144,3 +166,205 @@
"success": False,
"msg": str(e)
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+
+class UserManagementViewSet(ModelViewSet):
+ """
+ 用户管理视图集
+ 超级用户可以管理所有用户，普通用户只能查看自己的信息
+ """
+ authentication_classes = [MyJWTAuthentication]
+ permission_classes = [IsAuthenticated]
+ pagination_class = None
+ filter_backends = [SearchFilter]
+ search_fields = ['username', 'name']
+
+ def get_queryset(self):
+ if self.request.user.is_superuser:
+ return User.objects.filter(status='approved').prefetch_related('groups')
+ else:
+ return User.objects.filter(id=self.request.user.id).prefetch_related('groups')
+
+ def get_serializer_class(self):
+ if self.action == 'create':
+ return serializers.UserCreateSerializer
+ elif self.action in ['update', 'partial_update']:
+ return serializers.UserUpdateSerializer
+ else:
+ return serializers.UserDetailSerializer
+
+ def get_permissions(self):
+ if self.action in ['create', 'update', 'partial_update', 'destroy']:
+ self.permission_classes = [IsAuthenticated, CustomIsAdminUser]
+ return super().get_permissions()
+
+ def create(self, request, *args, **kwargs):
+ try:
+ return super().create(request, *args, **kwargs)
+ except Exception as e:
+ import logging
+ logger = logging.getLogger(__name__)
+ logger.error(f"创建用户失败: {str(e)}", exc_info=True)
+ return Response({
+ "success": False,
+ "msg": f"创建用户失败: {str(e)}"
+ }, status=status.HTTP_400_BAD_REQUEST)
+
+ def update(self, request, *args, **kwargs):
+ try:
+ return super().update(request, *args, **kwargs)
+ except Exception as e:
+ import logging
+ logger = logging.getLogger(__name__)
+ logger.error(f"更新用户失败: {str(e)}", exc_info=True)
+ return Response({
+ "success": False,
+ "msg": f"更新用户失败: {str(e)}"
+ }, status=status.HTTP_400_BAD_REQUEST)
+
+ def perform_create(self, serializer):
+ serializer.save()
+
+ def perform_update(self, serializer):
+ serializer.save()
+
+ def perform_destroy(self, instance):
+ instance.delete()
+
+
+class RegisterView(APIView):
+ """
+ 用户注册视图
+ """
+
+ authentication_classes = ()
+ permission_classes = ()
+
+ @swagger_auto_schema(request_body=serializers.UserRegisterSerializer)
+ def post(self, request):
+ """
+ 用户注册
+ {
+ username: str
+ password: str
+ confirm_password: str
+ name: str
+ phone: str
+ groups: list (optional)
+ }
+ """
+ serializer = serializers.UserRegisterSerializer(data=request.data)
+ if not serializer.is_valid():
+ return Response({
+ "success": False,
+ "msg": serializer.errors
+ }, status=status.HTTP_400_BAD_REQUEST)
+
+ try:
+ user = serializer.save()
+ return Response({
+ "success": True,
+ "msg": "你的注册申请已提交，待管理员审核中！",
+ "data": {
+ "id": user.id,
+ "username": user.username
+ }
+ }, status=status.HTTP_201_CREATED)
+ except Exception as e:
+ import logging
+ logger = logging.getLogger(__name__)
+ logger.error(f"注册失败: {str(e)}", exc_info=True)
+ return Response({
+ "success": False,
+ "msg": f"注册失败: {str(e)}"
+ }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+
+class UserApprovalViewSet(ModelViewSet):
+ """
+ 用户审批视图集
+ """
+ authentication_classes = [MyJWTAuthentication]
+ permission_classes = [IsAuthenticated]
+ pagination_class = None
+ filter_backends = [SearchFilter]
+ search_fields = ['username', 'name']
+
+ def get_queryset(self):
+ if self.request.user.is_superuser or self.request.user.is_staff:
+ return User.objects.filter(status='pending').prefetch_related('groups')
+ else:
+ return User.objects.none()
+
+ def get_serializer_class(self):
+ return serializers.UserDetailSerializer
+
+ def list(self, request, *args, **kwargs):
+ queryset = self.get_queryset()
+ serializer = self.get_serializer(queryset, many=True)
+ return Response({
+ "success": True,
+ "results": serializer.data
+ })
+
+ @swagger_auto_schema(request_body=serializers.UserApprovalSerializer)
+ @action(detail=True, methods=['post'])
+ def approve(self, request, pk=None):
+ """
+ 审批用户
+ {
+ action: 'approve' | 'reject'
+ reason: str (optional)
+ }
+ """
+ user_id = pk
+ try:
+ user = User.objects.get(id=user_id)
+ except User.DoesNotExist:
+ return Response({
+ "success": False,
+ "msg": "用户不存在"
+ }, status=status.HTTP_404_NOT_FOUND)
+
+ serializer = serializers.UserApprovalSerializer(data=request.data)
+ if not serializer.is_valid():
+ return Response({
+ "success": False,
+ "msg": serializer.errors
+ }, status=status.HTTP_400_BAD_REQUEST)
+
+ action = serializer.validated_data['action']
+
+ if action == 'approve':
+ user.status = 'approved'
+ user.is_active = True
+ user.save()
+ return Response({
+ "success": True,
+ "msg": "用户审批已通过"
+ })
+ elif action == 'reject':
+ user.status = 'rejected'
+ user.save()
+ return Response({
+ "success": True,
+ "msg": "用户审批未通过"
+ })
+
+
+class GroupListView(APIView):
+ """
+ 获取分组列表（无需认证）
+ 用于注册时选择分组
+ """
+ authentication_classes = ()
+ permission_classes = ()
+
+ def get(self, request):
+ """
+ 获取所有分组列表
+ """
+ from django.contrib.auth.models import Group
+ groups = Group.objects.all()
+ group_list = [{"id": group.id, "name": group.name} for group in groups]
+ return Response(group_list)
--
Gitblit v1.9.1