浅念
129 lines
5.6 KiB
Python
129 lines
5.6 KiB
Python
import pytest
|
||
from django.contrib.auth import get_user_model
|
||
from django.core.cache import cache
|
||
from rest_framework.test import APIClient
|
||
|
||
pytestmark = pytest.mark.django_db
|
||
User = get_user_model()
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clear_cache():
|
||
cache.clear() # locmem 限流历史跨测试残留，会让 429 用例假红（同 test_collect 模式）
|
||
|
||
|
||
@pytest.fixture
|
||
def client(): # 覆盖 pytest-django 的 Django Client——登录用例需要 credentials()/format="json"
|
||
return APIClient()
|
||
|
||
|
||
@pytest.fixture
|
||
def admin_user():
|
||
return User.objects.create_user(username="admin", password="adminpass123", is_staff=True)
|
||
|
||
|
||
def _auth(client, user):
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
token = RefreshToken.for_user(user)
|
||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token.access_token}")
|
||
|
||
|
||
def test_login_ok_returns_tokens(client, admin_user):
|
||
r = client.post("/api/admin/auth/token",
|
||
{"username": "admin", "password": "adminpass123"}, format="json")
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["code"] == 200 and body["data"]["access"] and body["data"]["refresh"]
|
||
|
||
|
||
def test_login_wrong_password_401_envelope(client, admin_user):
|
||
"""simplejwt 失败默认 {detail} 无信封——同 Task7 经验，finalize_response 统一转
|
||
{code,data,message}，管理台前端（Task17）按信封对接。"""
|
||
r = client.post("/api/admin/auth/token",
|
||
{"username": "admin", "password": "wrongpass"}, format="json")
|
||
assert r.status_code == 401
|
||
body = r.json()
|
||
assert body["code"] == 401 and body["data"] is None and body["message"]
|
||
|
||
|
||
def test_refresh_returns_new_access(client, admin_user):
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
refresh = RefreshToken.for_user(admin_user)
|
||
r = client.post("/api/admin/auth/refresh", {"refresh": str(refresh)}, format="json")
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["code"] == 200 and body["data"]["access"]
|
||
|
||
|
||
def test_me_requires_auth_envelope(client):
|
||
r = client.get("/api/admin/auth/me")
|
||
assert r.status_code == 401
|
||
assert r.json()["code"] == 401 # 无 token 的 401 同样走信封
|
||
|
||
|
||
def test_me_returns_username_and_flag(client, admin_user):
|
||
_auth(client, admin_user)
|
||
r = client.get("/api/admin/auth/me")
|
||
assert r.json()["data"] == {"username": "admin", "is_admin": True}
|
||
|
||
|
||
def test_user_create_admin_only(client, admin_user):
|
||
member = User.objects.create_user(username="member", password="memberpass123")
|
||
_auth(client, member)
|
||
r = client.post("/api/admin/users",
|
||
{"username": "newbie", "password": "longenough1"}, format="json")
|
||
assert r.status_code == 403
|
||
_auth(client, admin_user)
|
||
r = client.post("/api/admin/users",
|
||
{"username": "newbie", "password": "longenough1"}, format="json")
|
||
assert r.status_code == 200 and User.objects.filter(username="newbie", is_staff=False).exists()
|
||
|
||
|
||
def test_user_create_validates_input(client, admin_user):
|
||
_auth(client, admin_user)
|
||
assert client.post("/api/admin/users",
|
||
{"username": "x", "password": "short"}, format="json").status_code == 400
|
||
assert client.post("/api/admin/users",
|
||
{"username": "admin", "password": "longenough1"}, format="json").status_code == 409
|
||
|
||
|
||
# ---- 审查修复锁定测试 ----
|
||
|
||
def test_user_create_malformed_body_400_not_500(client, admin_user):
|
||
"""探针证实：JSON 数组 body / 非字符串 username/password 直接 500——入口统一 400 invalid_input。"""
|
||
_auth(client, admin_user)
|
||
assert client.post("/api/admin/users", ["arr"], format="json").status_code == 400
|
||
assert client.post("/api/admin/users",
|
||
{"username": 123, "password": "longenough1"}, format="json").status_code == 400
|
||
assert client.post("/api/admin/users",
|
||
{"username": "okname", "password": 12345678}, format="json").status_code == 400
|
||
|
||
def test_login_missing_password_field_envelope(client, admin_user):
|
||
"""simplejwt 缺 password 的 400 是 {"password":[...]} 字段校验形状——信封统一，前端单条解析路径。"""
|
||
r = client.post("/api/admin/auth/token", {"username": "admin"}, format="json")
|
||
assert r.status_code == 400
|
||
body = r.json()
|
||
assert body["code"] == 400 and body["message"] == "invalid_input" and body["data"]
|
||
|
||
def test_user_create_username_charset_and_length(client, admin_user):
|
||
"""字符集对齐 UnicodeUsernameValidator（create_user 不跑校验器）；>150 防 MySQL DataError（sqlite 不报错须入口拦）。"""
|
||
_auth(client, admin_user)
|
||
assert client.post("/api/admin/users",
|
||
{"username": "bad name", "password": "longenough1"}, format="json").status_code == 400
|
||
assert client.post("/api/admin/users",
|
||
{"username": "x" * 151, "password": "longenough1"}, format="json").status_code == 400
|
||
assert client.post("/api/admin/users",
|
||
{"username": "ok.name+1", "password": "longenough1"}, format="json").status_code == 200
|
||
|
||
def test_login_throttle_429_envelope(client, admin_user):
|
||
"""直接打默认 10/min（THROTTLE_RATES import 时固化为类属性，override_settings 提速无效，
|
||
同 collect.py CollectThrottle.get_rate 注释）——第 11 次 429 且带信封。"""
|
||
for _ in range(10):
|
||
assert client.post("/api/admin/auth/token",
|
||
{"username": "admin", "password": "wrongpass"}, format="json").status_code == 401
|
||
r = client.post("/api/admin/auth/token",
|
||
{"username": "admin", "password": "wrongpass"}, format="json")
|
||
assert r.status_code == 429
|
||
body = r.json()
|
||
assert body["code"] == 429 and body["data"] is None and body["message"]
|