Mplan
e75833ee4b
- models: 删除 Profile 模型，字段合并到 User；User 新增云朵/公开/收藏计数器；Cloud 新增 favorite_count - schemas/serializers: 展平 UserOut/AuthOut/MeOut/AdminUserOut，移除嵌套 profile - deps: user.profile.role→user.role，移除 selectinload profile - auth/register: 注册时直接设置 user 字段，不再创建 Profile - clouds: 创建/删除云朵时同步 user.cloud_count 计数器，状态变化时同步 public_cloud_count - favorites: 点赞/取消时同步 cloud.favorite_count 计数器 - admin: 审批/隐藏/批量删时同步 public_cloud_count 计数器 - profiles/stats: 用计数器替代实时 COUNT 查询 - alembic: 新增 migration 合并 profiles 到 users，初始化 counters
317 lines
9.7 KiB
Python
317 lines
9.7 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from tests.conftest import (
|
|
admin_token,
|
|
bearer,
|
|
png_bytes,
|
|
register_and_confirm,
|
|
upload_cloud,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# stats
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_stats(client: TestClient) -> None:
|
|
r = client.get("/api/v1/admin/stats", headers=bearer(admin_token(client)))
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
for key in ("users", "images", "today_uploads", "pending", "approved", "rejected", "hidden"):
|
|
assert key in body
|
|
|
|
|
|
def test_stats_non_admin(client: TestClient, monkeypatch) -> None:
|
|
info = register_and_confirm(
|
|
client, monkeypatch,
|
|
f"notadmin-{uuid.uuid4().hex[:8]}@example.com",
|
|
"notadmin12", "非管理员",
|
|
)
|
|
r = client.get("/api/v1/admin/stats", headers=bearer(info["token"]))
|
|
assert r.status_code == 403
|
|
assert "管理员" in r.json()["detail"]
|
|
|
|
|
|
def test_stats_unauthenticated(client: TestClient) -> None:
|
|
r = client.get("/api/v1/admin/stats")
|
|
assert r.status_code == 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list users
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_users(client: TestClient) -> None:
|
|
r = client.get("/api/v1/admin/users", headers=bearer(admin_token(client)))
|
|
assert r.status_code == 200
|
|
assert r.json()["total"] >= 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create user
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_user(client: TestClient) -> None:
|
|
adm = admin_token(client)
|
|
r = client.post(
|
|
"/api/v1/admin/users",
|
|
headers=bearer(adm),
|
|
json={
|
|
"email": f"admin-created-{uuid.uuid4().hex[:8]}@example.com",
|
|
"password": "createdpass12",
|
|
"username": "管理员创建的用户",
|
|
},
|
|
)
|
|
assert r.status_code == 201
|
|
body = r.json()
|
|
assert body["user"]["email_verified"] is True
|
|
assert body["user"]["role"] == "user"
|
|
|
|
|
|
def test_create_user_with_role(client: TestClient) -> None:
|
|
adm = admin_token(client)
|
|
r = client.post(
|
|
"/api/v1/admin/users",
|
|
headers=bearer(adm),
|
|
json={
|
|
"email": f"newadmin-{uuid.uuid4().hex[:8]}@example.com",
|
|
"password": "newadmin12",
|
|
"username": "新建管理员",
|
|
"role": "admin",
|
|
},
|
|
)
|
|
assert r.status_code == 201
|
|
assert r.json()["user"]["role"] == "admin"
|
|
|
|
|
|
def test_create_user_duplicate_email(client: TestClient) -> None:
|
|
adm = admin_token(client)
|
|
email = f"dup-admin-{uuid.uuid4().hex[:8]}@example.com"
|
|
client.post(
|
|
"/api/v1/admin/users",
|
|
headers=bearer(adm),
|
|
json={"email": email, "password": "dup1pass12", "username": "第一人"},
|
|
)
|
|
r = client.post(
|
|
"/api/v1/admin/users",
|
|
headers=bearer(adm),
|
|
json={"email": email, "password": "dup2pass12", "username": "第二人"},
|
|
)
|
|
assert r.status_code == 409
|
|
|
|
|
|
def test_create_user_non_admin(client: TestClient, monkeypatch) -> None:
|
|
info = register_and_confirm(
|
|
client, monkeypatch,
|
|
f"na-create-{uuid.uuid4().hex[:8]}@example.com",
|
|
"nonadmin1", "非管理员创建",
|
|
)
|
|
r = client.post(
|
|
"/api/v1/admin/users",
|
|
headers=bearer(info["token"]),
|
|
json={
|
|
"email": f"should-fail-{uuid.uuid4().hex[:8]}@example.com",
|
|
"password": "failpass12",
|
|
"username": "失败的用户",
|
|
},
|
|
)
|
|
assert r.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# update user
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_update_user_role(client: TestClient, monkeypatch) -> None:
|
|
info = register_and_confirm(
|
|
client, monkeypatch,
|
|
f"promote-{uuid.uuid4().hex[:8]}@example.com",
|
|
"promote12", "升管理员",
|
|
)
|
|
r = client.patch(
|
|
f"/api/v1/admin/users/{info['user_id']}",
|
|
headers=bearer(admin_token(client)),
|
|
json={"role": "admin"},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["user"]["role"] == "admin"
|
|
|
|
|
|
def test_update_user_disable(client: TestClient, monkeypatch) -> None:
|
|
info = register_and_confirm(
|
|
client, monkeypatch,
|
|
f"disableme-{uuid.uuid4().hex[:8]}@example.com",
|
|
"disableme", "被禁用",
|
|
)
|
|
adm = admin_token(client)
|
|
r = client.patch(
|
|
f"/api/v1/admin/users/{info['user_id']}",
|
|
headers=bearer(adm),
|
|
json={"is_disabled": True},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["user"]["is_disabled"] is True
|
|
|
|
client.patch(
|
|
f"/api/v1/admin/users/{info['user_id']}",
|
|
headers=bearer(adm),
|
|
json={"is_disabled": False},
|
|
)
|
|
|
|
|
|
def test_update_self_disable(client: TestClient) -> None:
|
|
adm = admin_token(client)
|
|
me = client.get("/api/v1/auth/me", headers=bearer(adm))
|
|
admin_id = me.json()["user"]["id"]
|
|
r = client.patch(
|
|
f"/api/v1/admin/users/{admin_id}",
|
|
headers=bearer(adm),
|
|
json={"is_disabled": True},
|
|
)
|
|
assert r.status_code == 400
|
|
assert "不能禁用自己" in r.json()["detail"]
|
|
|
|
|
|
def test_update_self_demote(client: TestClient) -> None:
|
|
adm = admin_token(client)
|
|
me = client.get("/api/v1/auth/me", headers=bearer(adm))
|
|
admin_id = me.json()["user"]["id"]
|
|
r = client.patch(
|
|
f"/api/v1/admin/users/{admin_id}",
|
|
headers=bearer(adm),
|
|
json={"role": "user"},
|
|
)
|
|
assert r.status_code == 400
|
|
assert "移除自己的管理员权限" in r.json()["detail"]
|
|
|
|
|
|
def test_update_nonexistent_user(client: TestClient) -> None:
|
|
r = client.patch(
|
|
f"/api/v1/admin/users/{uuid.uuid4()}",
|
|
headers=bearer(admin_token(client)),
|
|
json={"role": "admin"},
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list clouds
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_admin_clouds(client: TestClient) -> None:
|
|
r = client.get("/api/v1/admin/clouds", headers=bearer(admin_token(client)))
|
|
assert r.status_code == 200
|
|
assert "items" in r.json()
|
|
|
|
|
|
def test_list_admin_clouds_filter_status(client: TestClient) -> None:
|
|
r = client.get(
|
|
"/api/v1/admin/clouds",
|
|
headers=bearer(admin_token(client)),
|
|
params={"status": "pending"},
|
|
)
|
|
assert r.status_code == 200
|
|
for item in r.json()["items"]:
|
|
assert item["status"] == "pending"
|
|
|
|
|
|
def test_list_admin_clouds_filter_hidden(client: TestClient) -> None:
|
|
r = client.get(
|
|
"/api/v1/admin/clouds",
|
|
headers=bearer(admin_token(client)),
|
|
params={"is_hidden": "true"},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# update cloud status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_update_cloud_status(client: TestClient, monkeypatch) -> None:
|
|
info = register_and_confirm(
|
|
client, monkeypatch,
|
|
f"admin-cld-{uuid.uuid4().hex[:8]}@example.com",
|
|
"admincld", "管理员云审核",
|
|
)
|
|
ur = upload_cloud(client, info["token"], cloud_type_id="1")
|
|
cloud_id = ur.json()["cloud"]["id"]
|
|
r = client.patch(
|
|
"/api/v1/admin/clouds/status",
|
|
headers=bearer(admin_token(client)),
|
|
json={"ids": [cloud_id], "status": "rejected"},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["updated"] == 1
|
|
|
|
|
|
def test_update_cloud_status_nonexistent(client: TestClient) -> None:
|
|
r = client.patch(
|
|
"/api/v1/admin/clouds/status",
|
|
headers=bearer(admin_token(client)),
|
|
json={"ids": [str(uuid.uuid4())], "status": "approved"},
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# update cloud visibility
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_update_cloud_visibility(client: TestClient, monkeypatch) -> None:
|
|
info = register_and_confirm(
|
|
client, monkeypatch,
|
|
f"vis-{uuid.uuid4().hex[:8]}@example.com",
|
|
"vispass12", "可见性测试",
|
|
)
|
|
ur = upload_cloud(client, info["token"], cloud_type_id="1")
|
|
cloud_id = ur.json()["cloud"]["id"]
|
|
r = client.patch(
|
|
"/api/v1/admin/clouds/visibility",
|
|
headers=bearer(admin_token(client)),
|
|
json={"ids": [cloud_id], "is_hidden": True},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["updated"] == 1
|
|
|
|
g = client.get(f"/api/v1/clouds/{cloud_id}", headers=bearer(info["token"]))
|
|
assert g.json()["is_hidden"] is True
|
|
|
|
client.patch(
|
|
"/api/v1/admin/clouds/visibility",
|
|
headers=bearer(admin_token(client)),
|
|
json={"ids": [cloud_id], "is_hidden": False},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# batch delete clouds (admin)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_admin_batch_delete_clouds(client: TestClient, monkeypatch) -> None:
|
|
info = register_and_confirm(
|
|
client, monkeypatch,
|
|
f"adm-del-{uuid.uuid4().hex[:8]}@example.com",
|
|
"admdel12", "管理员删图",
|
|
)
|
|
ur = upload_cloud(client, info["token"], cloud_type_id="1")
|
|
cloud_id = ur.json()["cloud"]["id"]
|
|
r = client.post(
|
|
"/api/v1/admin/clouds/batch-delete",
|
|
headers=bearer(admin_token(client)),
|
|
json={"ids": [cloud_id]},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["deleted"] == 1
|