from django.contrib.auth.models import User from django.test import TestCase from rest_framework import status from rest_framework.test import APIClient from rest_framework_simplejwt.tokens import RefreshToken from basic_info.models import Employee, EmployeeStatusEnum, Merchant, MerchantTypeEnum from flower.error_code import AuthErrorCode class AuthLoginAPITestCase(TestCase): def setUp(self): self.url = '/api/auth/login/' self.client = APIClient() self.merchant = Merchant.objects.create(name='登录商户', type=MerchantTypeEnum.FACTORY) def _create_user_with_employee(self, *, username='login_user', password='pass12345', is_active=True, employee_status=EmployeeStatusEnum.ACTIVE): user = User.objects.create_user(username=username, password=password, is_active=is_active) Employee.objects.create( merchant=self.merchant, sys_user=user, name=f'{username}员工', status=employee_status, ) return user def assert_login_error(self, response, *, http_status, error_code, code, message): self.assertEqual(response.status_code, http_status) self.assertEqual(response.data['error_code'], int(error_code)) self.assertEqual(response.data['code'], code) self.assertEqual(response.data['message'], message) self.assertEqual(response.data['detail'], message) def test_login_success(self): self._create_user_with_employee(username='success_user', password='pass12345') response = self.client.post( self.url, {'username': 'success_user', 'password': 'pass12345'}, format='json', ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('access', response.data) self.assertIn('refresh', response.data) def test_login_requires_username_and_password(self): response = self.client.post(self.url, {}, format='json') self.assert_login_error( response, http_status=status.HTTP_400_BAD_REQUEST, error_code=AuthErrorCode.MISSING_CREDENTIALS, code='missing_credentials', message='请输入用户名和密码', ) def test_login_requires_username(self): response = self.client.post(self.url, {'password': 'pass12345'}, format='json') self.assert_login_error( response, http_status=status.HTTP_400_BAD_REQUEST, error_code=AuthErrorCode.MISSING_USERNAME, code='missing_username', message='请输入用户名', ) def test_login_requires_password(self): response = self.client.post(self.url, {'username': 'login_user'}, format='json') self.assert_login_error( response, http_status=status.HTTP_400_BAD_REQUEST, error_code=AuthErrorCode.MISSING_PASSWORD, code='missing_password', message='请输入密码', ) def test_login_rejects_unknown_user(self): response = self.client.post( self.url, {'username': 'missing_user', 'password': 'pass12345'}, format='json', ) self.assert_login_error( response, http_status=status.HTTP_401_UNAUTHORIZED, error_code=AuthErrorCode.USER_NOT_FOUND, code='user_not_found', message='用户不存在', ) def test_login_rejects_invalid_password(self): self._create_user_with_employee(username='wrong_password_user', password='pass12345') response = self.client.post( self.url, {'username': 'wrong_password_user', 'password': 'bad-password'}, format='json', ) self.assert_login_error( response, http_status=status.HTTP_401_UNAUTHORIZED, error_code=AuthErrorCode.INVALID_PASSWORD, code='invalid_password', message='密码错误', ) def test_login_rejects_inactive_user(self): self._create_user_with_employee(username='inactive_user', password='pass12345', is_active=False) response = self.client.post( self.url, {'username': 'inactive_user', 'password': 'pass12345'}, format='json', ) self.assert_login_error( response, http_status=status.HTTP_403_FORBIDDEN, error_code=AuthErrorCode.USER_INACTIVE, code='user_inactive', message='该用户已被禁用', ) def test_login_rejects_user_without_employee(self): User.objects.create_user(username='no_employee_user', password='pass12345') response = self.client.post( self.url, {'username': 'no_employee_user', 'password': 'pass12345'}, format='json', ) self.assert_login_error( response, http_status=status.HTTP_403_FORBIDDEN, error_code=AuthErrorCode.EMPLOYEE_NOT_BOUND, code='employee_not_bound', message='该用户未绑定员工身份', ) def test_login_rejects_inactive_employee(self): self._create_user_with_employee( username='inactive_employee_user', password='pass12345', employee_status=EmployeeStatusEnum.INACTIVE, ) response = self.client.post( self.url, {'username': 'inactive_employee_user', 'password': 'pass12345'}, format='json', ) self.assert_login_error( response, http_status=status.HTTP_403_FORBIDDEN, error_code=AuthErrorCode.EMPLOYEE_INACTIVE, code='employee_inactive', message='该员工已离职或停用', ) class JwtAuthenticationFailureAPITestCase(TestCase): def setUp(self): self.url = '/api/v2/me/visible-pages/' self.client = APIClient() def assert_auth_error(self, response, *, http_status, error_code, code, message): self.assertEqual(response.status_code, http_status) self.assertEqual(response.data['error_code'], int(error_code)) self.assertEqual(response.data['code'], code) self.assertEqual(response.data['message'], message) self.assertEqual(response.data['detail'], message) def test_protected_api_without_token_returns_structured_error(self): response = self.client.get(self.url) self.assert_auth_error( response, http_status=status.HTTP_401_UNAUTHORIZED, error_code=AuthErrorCode.NOT_AUTHENTICATED, code='not_authenticated', message='未提供认证凭据', ) def test_protected_api_bad_authorization_header_returns_structured_error(self): response = self.client.get(self.url, HTTP_AUTHORIZATION='Bearer token extra') self.assert_auth_error( response, http_status=status.HTTP_401_UNAUTHORIZED, error_code=AuthErrorCode.BAD_AUTHORIZATION_HEADER, code='bad_authorization_header', message='Authorization 请求头格式错误', ) def test_protected_api_invalid_token_returns_structured_error(self): response = self.client.get(self.url, HTTP_AUTHORIZATION='Bearer invalid-token') self.assert_auth_error( response, http_status=status.HTTP_401_UNAUTHORIZED, error_code=AuthErrorCode.TOKEN_NOT_VALID, code='token_not_valid', message='Token 无效或已过期', ) self.assertIn('messages', response.data) def test_protected_api_deleted_token_user_returns_structured_error(self): user = User.objects.create_user(username='deleted_token_user', password='pass12345') token = str(RefreshToken.for_user(user).access_token) user.delete() response = self.client.get(self.url, HTTP_AUTHORIZATION=f'Bearer {token}') self.assert_auth_error( response, http_status=status.HTTP_401_UNAUTHORIZED, error_code=AuthErrorCode.TOKEN_USER_NOT_FOUND, code='token_user_not_found', message='Token 对应用户不存在', ) def test_protected_api_inactive_token_user_returns_structured_error(self): user = User.objects.create_user(username='inactive_token_user', password='pass12345', is_active=False) token = str(RefreshToken.for_user(user).access_token) response = self.client.get(self.url, HTTP_AUTHORIZATION=f'Bearer {token}') self.assert_auth_error( response, http_status=status.HTTP_401_UNAUTHORIZED, error_code=AuthErrorCode.TOKEN_USER_INACTIVE, code='token_user_inactive', message='Token 对应用户已被禁用', ) class ErrorCodeListAPITestCase(TestCase): def setUp(self): self.url = '/api/error-codes/' self.client = APIClient() def test_error_code_list_is_public_and_includes_auth_codes(self): response = self.client.get(self.url) self.assertEqual(response.status_code, status.HTTP_200_OK) auth_group = next(group for group in response.data['modules'] if group['module'] == 'auth') self.assertEqual(auth_group['title'], '认证') codes = {item['error_code']: item for item in auth_group['codes']} self.assertEqual(codes[10001]['name'], 'MISSING_CREDENTIALS') self.assertEqual(codes[10001]['code'], 'missing_credentials') self.assertEqual(codes[10001]['message'], '请输入用户名和密码') self.assertEqual(codes[10014]['name'], 'PERMISSION_DENIED') self.assertEqual(codes[10014]['code'], 'permission_denied') self.assertEqual(codes[10014]['message'], '无权限访问该资源') self.assertEqual(set(codes.keys()), {int(item) for item in AuthErrorCode})