245 lines
7.5 KiB
Python
245 lines
7.5 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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()["profile"]["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()["profile"]["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
|