@@ -0,0 +1,87 @@
|
||||
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
|
||||
Reference in New Issue
Block a user