wehub-resource-sync
255 lines
11 KiB
Python
255 lines
11 KiB
Python
"""
|
|
Integration tests for authentication-related API routes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
|
|
|
|
|
|
async def _require_superadmin(test_client, headers):
|
|
response = await test_client.get("/api/auth/me", headers=headers)
|
|
assert response.status_code == 200, response.text
|
|
if response.json()["role"] != "superadmin":
|
|
pytest.fail("This test requires TEST_USERNAME to be a superadmin account.")
|
|
|
|
|
|
async def _create_department_with_admin(test_client, headers, label: str) -> dict:
|
|
suffix = uuid.uuid4().hex[:8]
|
|
admin_uid = f"adm{label}_{suffix}"
|
|
admin_password = f"Pw!{suffix}"
|
|
response = await test_client.post(
|
|
"/api/departments",
|
|
json={
|
|
"name": f"pytest_{label}_{suffix}",
|
|
"description": "pytest managed department",
|
|
"admin_uid": admin_uid,
|
|
"admin_password": admin_password,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
|
|
login_response = await test_client.post(
|
|
"/api/auth/token",
|
|
data={"username": admin_uid, "password": admin_password},
|
|
)
|
|
assert login_response.status_code == 200, login_response.text
|
|
|
|
login_payload = login_response.json()
|
|
return {
|
|
"department": response.json(),
|
|
"admin_id": login_payload["user_id"],
|
|
"admin_headers": {"Authorization": f"Bearer {login_payload['access_token']}"},
|
|
}
|
|
|
|
|
|
async def _create_user(test_client, headers, label: str, role: str = "user", department_id: int | None = None) -> dict:
|
|
suffix = uuid.uuid4().hex[:8]
|
|
payload = {
|
|
"username": f"u{label}_{suffix}",
|
|
"password": f"Pw!{suffix}",
|
|
"role": role,
|
|
}
|
|
if department_id is not None:
|
|
payload["department_id"] = department_id
|
|
|
|
response = await test_client.post("/api/auth/users", json=payload, headers=headers)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
async def _cleanup_user(test_client, headers, user_id: int) -> None:
|
|
response = await test_client.delete(f"/api/auth/users/{user_id}", headers=headers)
|
|
assert response.status_code in {200, 404}, response.text
|
|
|
|
|
|
async def _cleanup_department(test_client, headers, department_id: int) -> None:
|
|
response = await test_client.delete(f"/api/departments/{department_id}", headers=headers)
|
|
assert response.status_code in {200, 404}, response.text
|
|
|
|
|
|
async def test_login_with_invalid_credentials(test_client):
|
|
response = await test_client.post("/api/auth/token", data={"username": "invalid", "password": "invalid"})
|
|
assert response.status_code == 401
|
|
assert "detail" in response.json()
|
|
|
|
|
|
async def test_user_is_locked_after_repeated_failed_logins(test_client, standard_user):
|
|
uid = standard_user["user"]["uid"]
|
|
|
|
for attempt in range(1, 5):
|
|
response = await test_client.post("/api/auth/token", data={"username": uid, "password": "wrong-password"})
|
|
assert response.status_code == 401, response.text
|
|
assert response.json()["detail"] == "用户名或密码错误"
|
|
|
|
locked_response = await test_client.post("/api/auth/token", data={"username": uid, "password": "wrong-password"})
|
|
assert locked_response.status_code == 423, locked_response.text
|
|
assert "X-Lock-Remaining" in locked_response.headers
|
|
assert "账户已被锁定" in locked_response.json()["detail"]
|
|
|
|
still_locked_response = await test_client.post(
|
|
"/api/auth/token",
|
|
data={"username": uid, "password": standard_user["password"]},
|
|
)
|
|
assert still_locked_response.status_code == 423, still_locked_response.text
|
|
assert "X-Lock-Remaining" in still_locked_response.headers
|
|
assert "登录被锁定" in still_locked_response.json()["detail"]
|
|
|
|
|
|
async def test_admin_can_login_and_fetch_profile(test_client, admin_headers):
|
|
profile_response = await test_client.get("/api/auth/me", headers=admin_headers)
|
|
assert profile_response.status_code == 200
|
|
data = profile_response.json()
|
|
assert data["role"] in {"admin", "superadmin"}
|
|
assert data["username"]
|
|
assert data["id"]
|
|
|
|
|
|
async def test_profile_requires_authentication(test_client):
|
|
response = await test_client.get("/api/auth/me")
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "请登录后再访问"
|
|
|
|
|
|
async def test_admin_can_create_and_delete_user(test_client, admin_headers):
|
|
suffix = uuid.uuid4().hex[:8]
|
|
payload = {
|
|
"username": f"rtu_{suffix}",
|
|
"password": "routerTest123!",
|
|
"role": "user",
|
|
}
|
|
create_response = await test_client.post("/api/auth/users", json=payload, headers=admin_headers)
|
|
assert create_response.status_code == 200, create_response.text
|
|
|
|
created_user = create_response.json()
|
|
assert created_user["username"] == payload["username"]
|
|
assert created_user["role"] == payload["role"]
|
|
|
|
delete_response = await test_client.delete(f"/api/auth/users/{created_user['id']}", headers=admin_headers)
|
|
assert delete_response.status_code == 200, delete_response.text
|
|
delete_payload = delete_response.json()
|
|
assert delete_payload["success"] is True
|
|
assert delete_payload["message"] == "用户已删除"
|
|
|
|
|
|
async def test_department_admin_is_limited_to_own_department_users(test_client, admin_headers):
|
|
await _require_superadmin(test_client, admin_headers)
|
|
|
|
user_ids: list[int] = []
|
|
admin_ids: list[int] = []
|
|
department_ids: list[int] = []
|
|
|
|
try:
|
|
dept_a = await _create_department_with_admin(test_client, admin_headers, "a")
|
|
dept_b = await _create_department_with_admin(test_client, admin_headers, "b")
|
|
department_a = dept_a["department"]
|
|
department_b = dept_b["department"]
|
|
department_ids.extend([department_a["id"], department_b["id"]])
|
|
admin_ids.extend([dept_a["admin_id"], dept_b["admin_id"]])
|
|
|
|
user_a = await _create_user(test_client, dept_a["admin_headers"], "a")
|
|
user_b = await _create_user(test_client, dept_b["admin_headers"], "b")
|
|
superadmin_created_user = await _create_user(test_client, admin_headers, "s", department_id=department_b["id"])
|
|
user_ids.extend([user_a["id"], user_b["id"], superadmin_created_user["id"]])
|
|
|
|
assert user_a["department_id"] == department_a["id"]
|
|
assert superadmin_created_user["department_id"] == department_b["id"]
|
|
|
|
forbidden_create = await test_client.post(
|
|
"/api/auth/users",
|
|
json={
|
|
"username": f"ux_{uuid.uuid4().hex[:8]}",
|
|
"password": "routerTest123!",
|
|
"role": "user",
|
|
"department_id": department_b["id"],
|
|
},
|
|
headers=dept_a["admin_headers"],
|
|
)
|
|
assert forbidden_create.status_code == 403, forbidden_create.text
|
|
|
|
list_response = await test_client.get("/api/auth/users", headers=dept_a["admin_headers"])
|
|
assert list_response.status_code == 200, list_response.text
|
|
listed_users = list_response.json()
|
|
listed_user_ids = {user["id"] for user in listed_users}
|
|
assert user_a["id"] in listed_user_ids
|
|
assert user_b["id"] not in listed_user_ids
|
|
assert all(user["department_id"] == department_a["id"] for user in listed_users)
|
|
|
|
options_response = await test_client.get("/api/auth/users/access-options", headers=dept_a["admin_headers"])
|
|
assert options_response.status_code == 200, options_response.text
|
|
access_options = options_response.json()
|
|
option_uids = {user["uid"] for user in access_options}
|
|
assert user_a["uid"] in option_uids
|
|
assert user_b["uid"] not in option_uids
|
|
assert all(user["department_id"] == department_a["id"] for user in access_options)
|
|
|
|
superadmin_list_response = await test_client.get("/api/auth/users?limit=1000", headers=admin_headers)
|
|
assert superadmin_list_response.status_code == 200, superadmin_list_response.text
|
|
superadmin_user_ids = {user["id"] for user in superadmin_list_response.json()}
|
|
assert user_a["id"] in superadmin_user_ids
|
|
assert user_b["id"] in superadmin_user_ids
|
|
|
|
own_read = await test_client.get(f"/api/auth/users/{user_a['id']}", headers=dept_a["admin_headers"])
|
|
assert own_read.status_code == 200, own_read.text
|
|
|
|
cross_read = await test_client.get(f"/api/auth/users/{user_b['id']}", headers=dept_a["admin_headers"])
|
|
assert cross_read.status_code == 403, cross_read.text
|
|
|
|
cross_update = await test_client.put(
|
|
f"/api/auth/users/{user_b['id']}",
|
|
json={"username": f"ub_{uuid.uuid4().hex[:8]}"},
|
|
headers=dept_a["admin_headers"],
|
|
)
|
|
assert cross_update.status_code == 403, cross_update.text
|
|
|
|
role_escalation = await test_client.put(
|
|
f"/api/auth/users/{user_a['id']}", json={"role": "admin"}, headers=dept_a["admin_headers"]
|
|
)
|
|
assert role_escalation.status_code == 403, role_escalation.text
|
|
|
|
cross_delete = await test_client.delete(f"/api/auth/users/{user_b['id']}", headers=dept_a["admin_headers"])
|
|
assert cross_delete.status_code == 403, cross_delete.text
|
|
|
|
own_delete = await test_client.delete(f"/api/auth/users/{user_a['id']}", headers=dept_a["admin_headers"])
|
|
assert own_delete.status_code == 200, own_delete.text
|
|
user_ids.remove(user_a["id"])
|
|
finally:
|
|
for user_id in user_ids:
|
|
await _cleanup_user(test_client, admin_headers, user_id)
|
|
for admin_id in admin_ids:
|
|
await _cleanup_user(test_client, admin_headers, admin_id)
|
|
for department_id in department_ids:
|
|
await _cleanup_department(test_client, admin_headers, department_id)
|
|
|
|
|
|
async def test_invalid_token_is_rejected(test_client):
|
|
headers = {"Authorization": "Bearer not-a-real-token"}
|
|
response = await test_client.get("/api/auth/me", headers=headers)
|
|
assert response.status_code == 401
|
|
|
|
|
|
async def test_deleted_user_token_is_rejected(test_client, admin_headers, standard_user):
|
|
user_id = standard_user["user"]["id"]
|
|
|
|
delete_response = await test_client.delete(f"/api/auth/users/{user_id}", headers=admin_headers)
|
|
assert delete_response.status_code == 200, delete_response.text
|
|
|
|
profile_response = await test_client.get("/api/auth/me", headers=standard_user["headers"])
|
|
assert profile_response.status_code == 401
|
|
|
|
|
|
async def test_locked_user_token_is_rejected(test_client, standard_user):
|
|
uid = standard_user["user"]["uid"]
|
|
|
|
for _ in range(5):
|
|
await test_client.post("/api/auth/token", data={"username": uid, "password": "wrong-password"})
|
|
|
|
profile_response = await test_client.get("/api/auth/me", headers=standard_user["headers"])
|
|
assert profile_response.status_code == 423
|
|
assert "X-Lock-Remaining" in profile_response.headers
|