forked from erp-dev/erp
133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
"""
|
||
企业微信快捷登录 API 测试
|
||
"""
|
||
|
||
from django.contrib.auth.models import User
|
||
from django.test import TestCase, override_settings
|
||
from rest_framework.test import APIClient
|
||
|
||
from basic_info.models import Employee, Merchant, MerchantTypeEnum
|
||
|
||
|
||
@override_settings(AGENT_ACCESS_KEY='test-key-123')
|
||
class WecomLoginAPITestCase(TestCase):
|
||
"""测试企业微信快捷登录接口"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
self.url = '/api/v2/wecom/login/'
|
||
self.auth_header = 'test-key-123'
|
||
|
||
# 创建商户
|
||
self.merchant = Merchant.objects.create(
|
||
name='测试商户',
|
||
type=MerchantTypeEnum.STORE,
|
||
)
|
||
|
||
# 创建用户
|
||
self.user = User.objects.create_user(
|
||
username='testuser',
|
||
password='testpass123',
|
||
)
|
||
|
||
# 创建员工并关联用户，设置 wecom_user_id
|
||
self.employee = Employee.objects.create(
|
||
merchant=self.merchant,
|
||
name='测试员工',
|
||
sys_user=self.user,
|
||
wecom_user_id='wx_user_001',
|
||
)
|
||
|
||
def _post(self, data, auth=None):
|
||
"""辅助方法：发送 POST 请求"""
|
||
headers = {'HTTP_AUTHORIZATION': auth or self.auth_header}
|
||
return self.client.post(self.url, data, format='json', **headers)
|
||
|
||
# ==================== 认证测试 ====================
|
||
|
||
def test_missing_authorization_header(self):
|
||
"""缺少 Authorization 头应返回 401"""
|
||
resp = self.client.post(self.url, {'wecom_user_id': 'wx_user_001'}, format='json')
|
||
self.assertEqual(resp.status_code, 401)
|
||
|
||
def test_invalid_authorization_key(self):
|
||
"""错误的 AGENT_ACCESS_KEY 应返回 401"""
|
||
resp = self._post({'wecom_user_id': 'wx_user_001'}, auth='wrong-key')
|
||
self.assertEqual(resp.status_code, 401)
|
||
|
||
# ==================== 参数验证测试 ====================
|
||
|
||
def test_missing_wecom_user_id(self):
|
||
"""缺少 wecom_user_id 应返回 400"""
|
||
resp = self._post({})
|
||
self.assertEqual(resp.status_code, 400)
|
||
self.assertIn('wecom_user_id', resp.json())
|
||
|
||
# ==================== 登录成功测试 ====================
|
||
|
||
def test_login_success(self):
|
||
"""正常登录应返回 200 和 JWT token"""
|
||
resp = self._post({'wecom_user_id': 'wx_user_001'})
|
||
self.assertEqual(resp.status_code, 200)
|
||
data = resp.json()
|
||
self.assertIn('access', data)
|
||
self.assertIn('refresh', data)
|
||
self.assertEqual(data['employee_id'], self.employee.id)
|
||
self.assertEqual(data['employee_name'], '测试员工')
|
||
self.assertEqual(data['username'], 'testuser')
|
||
# access token 应该是非空字符串
|
||
self.assertTrue(len(data['access']) > 0)
|
||
self.assertTrue(len(data['refresh']) > 0)
|
||
|
||
def test_login_token_is_valid_jwt(self):
|
||
"""返回的 token 应该可以用于认证后续请求"""
|
||
resp = self._post({'wecom_user_id': 'wx_user_001'})
|
||
self.assertEqual(resp.status_code, 200)
|
||
access_token = resp.json()['access']
|
||
|
||
# 使用 token 访问需要认证的接口
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {access_token}')
|
||
# 访问 health check 不需要特殊权限，但需要认证
|
||
health_resp = self.client.get('/api/v2/health/')
|
||
# health check 应该返回 200（不管认证方式）
|
||
self.assertEqual(health_resp.status_code, 200)
|
||
|
||
# ==================== 错误场景测试 ====================
|
||
|
||
def test_unbound_wecom_user_id(self):
|
||
"""未绑定的 wecom_user_id 应返回 404"""
|
||
resp = self._post({'wecom_user_id': 'wx_unknown_user'})
|
||
self.assertEqual(resp.status_code, 404)
|
||
self.assertEqual(resp.json()['error'], '该企业微信用户未绑定本系统员工')
|
||
|
||
def test_employee_without_sys_user(self):
|
||
"""员工未关联系统用户应返回 404"""
|
||
# 创建一个没有 sys_user 的员工
|
||
employee_no_user = Employee.objects.create(
|
||
merchant=self.merchant,
|
||
name='无用户员工',
|
||
wecom_user_id='wx_no_user',
|
||
)
|
||
resp = self._post({'wecom_user_id': 'wx_no_user'})
|
||
self.assertEqual(resp.status_code, 404)
|
||
self.assertEqual(resp.json()['error'], '该员工未关联系统用户')
|
||
|
||
def test_inactive_user(self):
|
||
"""被禁用的用户应返回 403"""
|
||
self.user.is_active = False
|
||
self.user.save()
|
||
resp = self._post({'wecom_user_id': 'wx_user_001'})
|
||
self.assertEqual(resp.status_code, 403)
|
||
self.assertEqual(resp.json()['error'], '该用户已被禁用')
|
||
|
||
# ==================== 边界测试 ====================
|
||
|
||
def test_multiple_logins_return_different_tokens(self):
|
||
"""多次登录应返回不同的 token（每次生成新 token）"""
|
||
resp1 = self._post({'wecom_user_id': 'wx_user_001'})
|
||
resp2 = self._post({'wecom_user_id': 'wx_user_001'})
|
||
self.assertEqual(resp1.status_code, 200)
|
||
self.assertEqual(resp2.status_code, 200)
|
||
# refresh token 每次应该不同
|
||
self.assertNotEqual(resp1.json()['refresh'], resp2.json()['refresh'])
|