diff --git a/tests/conftest.py b/tests/conftest.py index 91647c7..36013e7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,5 @@ import asyncio +import io import os import shutil import tempfile @@ -8,12 +9,23 @@ from pathlib import Path import pytest from fastapi.testclient import TestClient +from PIL import Image TEST_ID = uuid.uuid4().hex TEST_DB = Path(tempfile.gettempdir()) / f"opencloud-{TEST_ID}.sqlite3" TEST_UPLOADS = Path(tempfile.gettempdir()) / f"opencloud-{TEST_ID}-uploads" + +def bearer(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +def png_bytes() -> bytes: + output = io.BytesIO() + Image.new("RGB", (200, 200), color=(125, 196, 240)).save(output, format="PNG") + return output.getvalue() + os.environ["ENVIRONMENT"] = "test" os.environ["DATABASE_URL"] = f"sqlite+aiosqlite:///{TEST_DB}" os.environ["SECRET_KEY"] = "test-secret-key-that-is-long-enough-for-opencloud" diff --git a/tests/test_admin.py b/tests/test_admin.py new file mode 100644 index 0000000..cbda5ae --- /dev/null +++ b/tests/test_admin.py @@ -0,0 +1,285 @@ +import uuid + +import pytest +from fastapi.testclient import TestClient + +from app.routers import auth as auth_router +from tests.conftest import bearer, png_bytes + +ADMIN_EMAIL = "admin@example.com" +ADMIN_PASSWORD = "admin-password-123" + + +def _capture_email(monkeypatch) -> dict[str, str]: + captured: dict[str, str] = {} + + async def capture(recipient: str, token: str) -> None: + captured["token"] = token + + monkeypatch.setattr(auth_router, "send_confirmation_email", capture) + return captured + + +def _register_and_confirm(client, monkeypatch, email, password, username) -> dict[str, str]: + captured = _capture_email(monkeypatch) + r = client.post( + "/api/v1/auth/register", + json={"email": email, "password": password, "username": username}, + ) + assert r.status_code == 201 + client.post("/api/v1/auth/confirm-email", json={"token": captured["token"]}) + lr = client.post("/api/v1/auth/login", json={"email": email, "password": password}) + body = lr.json() + return {"token": body["access_token"], "user_id": body["user"]["id"]} + + +def _admin_token(client: TestClient) -> str: + r = client.post( + "/api/v1/auth/login", + json={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD}, + ) + return r.json()["access_token"] + + +def _upload_cloud(client: TestClient, token: str, **overrides): + data = {"cloud_type_id": "1", "is_hidden": "false"} + data.update(overrides) + return client.post( + "/api/v1/clouds", + headers=bearer(token), + files={"image": ("cloud.png", png_bytes(), "image/png")}, + data=data, + ) + + +# --------------------------------------------------------------------------- +# 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", "升管理员", + ) + admin = _admin_token(client) + r = client.patch( + f"/api/v1/admin/users/{info['user_id']}", + headers=bearer(admin), + 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", "被禁用", + ) + admin = _admin_token(client) + r = client.patch( + f"/api/v1/admin/users/{info['user_id']}", + headers=bearer(admin), + 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(admin), + json={"is_disabled": False}, + ) + + +def test_update_self_disable(client: TestClient) -> None: + admin_token = _admin_token(client) + me = client.get("/api/v1/auth/me", headers=bearer(admin_token)) + admin_id = me.json()["user"]["id"] + r = client.patch( + f"/api/v1/admin/users/{admin_id}", + headers=bearer(admin_token), + json={"is_disabled": True}, + ) + assert r.status_code == 400 + assert "不能禁用自己" in r.json()["detail"] + + +def test_update_self_demote(client: TestClient) -> None: + admin_token = _admin_token(client) + me = client.get("/api/v1/auth/me", headers=bearer(admin_token)) + admin_id = me.json()["user"]["id"] + r = client.patch( + f"/api/v1/admin/users/{admin_id}", + headers=bearer(admin_token), + 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 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..c4f95af --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,324 @@ +import uuid + +import pytest +from fastapi.testclient import TestClient + +from app.routers import auth as auth_router +from tests.conftest import bearer + +ADMIN_EMAIL = "admin@example.com" +ADMIN_PASSWORD = "admin-password-123" + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def _capture_email(monkeypatch) -> dict[str, str]: + captured: dict[str, str] = {} + + async def capture(recipient: str, token: str) -> None: + captured["token"] = token + + monkeypatch.setattr(auth_router, "send_confirmation_email", capture) + return captured + + +def _register(client: TestClient, email: str, password: str, username: str): + return client.post( + "/api/v1/auth/register", + json={"email": email, "password": password, "username": username}, + ) + + +def _confirm(client: TestClient, token: str): + return client.post("/api/v1/auth/confirm-email", json={"token": token}) + + +def _login(client: TestClient, email: str, password: str): + return client.post( + "/api/v1/auth/login", + json={"email": email, "password": password}, + ) + + +def _register_and_confirm( + client: TestClient, + monkeypatch, + email: str, + password: str, + username: str, +) -> dict[str, str]: + captured = _capture_email(monkeypatch) + r = _register(client, email, password, username) + assert r.status_code == 201 + token = captured.get("token") + assert token + cr = _confirm(client, token) + assert cr.status_code == 200 + lr = _login(client, email, password) + assert lr.status_code == 200 + body = lr.json() + return {"token": body["access_token"], "user_id": body["user"]["id"]} + + +def _login_as_user(client: TestClient, email: str, password: str) -> str: + r = _login(client, email, password) + return r.json()["access_token"] + + +# --------------------------------------------------------------------------- +# register +# --------------------------------------------------------------------------- + + +def test_register_success(client: TestClient, monkeypatch) -> None: + email = f"reg-{uuid.uuid4().hex[:8]}@example.com" + captured = _capture_email(monkeypatch) + r = _register(client, email, "regpassword", "注册测试") + assert r.status_code == 201 + assert r.json()["message"] == "注册成功,请查收确认邮件" + assert "token" in captured + + +def test_register_duplicate_email(client: TestClient, monkeypatch) -> None: + email = f"dup-{uuid.uuid4().hex[:8]}@example.com" + _register(client, email, "firstpass", "第一个用户") + r = _register(client, email, "secondpass", "第二个用户") + assert r.status_code == 409 + assert "该邮箱已被注册" == r.json()["detail"] + + +def test_register_duplicate_username(client: TestClient, monkeypatch) -> None: + email_a = f"dupname-a-{uuid.uuid4().hex[:8]}@example.com" + email_b = f"dupname-b-{uuid.uuid4().hex[:8]}@example.com" + _register(client, email_a, "passA123", "同名用户") + r = _register(client, email_b, "passB123", "同名用户") + assert r.status_code == 409 + assert "这个昵称已经被使用" == r.json()["detail"] + + +def test_register_short_password(client: TestClient) -> None: + r = _register(client, f"s-{uuid.uuid4().hex[:8]}@e.com", "1234567", "短密码") + assert r.status_code == 422 + + +def test_register_short_username(client: TestClient) -> None: + r = _register(client, f"s-{uuid.uuid4().hex[:8]}@e.com", "pass12345", "A") + assert r.status_code == 422 + + +def test_register_invalid_email(client: TestClient) -> None: + r = _register(client, "not-an-email", "pass12345", "坏邮箱") + assert r.status_code == 422 + + +# --------------------------------------------------------------------------- +# confirm-email +# --------------------------------------------------------------------------- + + +def test_confirm_email_valid(client: TestClient, monkeypatch) -> None: + email = f"confirm-{uuid.uuid4().hex[:8]}@example.com" + captured = _capture_email(monkeypatch) + r = _register(client, email, "confirmpass", "确认测试") + assert r.status_code == 201 + token = captured.get("token") + assert token + cr = _confirm(client, token) + assert cr.status_code == 200 + assert cr.json()["message"] == "邮箱确认成功" + + +def test_confirm_email_invalid_token(client: TestClient) -> None: + r = _confirm(client, "not-a-valid-token-at-all-never") + assert r.status_code == 400 + assert "无效" in r.json()["detail"] + + +# --------------------------------------------------------------------------- +# login +# --------------------------------------------------------------------------- + + +def test_login_success(client: TestClient, monkeypatch) -> None: + email = f"login-{uuid.uuid4().hex[:8]}@example.com" + info = _register_and_confirm(client, monkeypatch, email, "loginpass", "登录测试") + assert info["token"] + assert info["user_id"] + + +def test_login_wrong_password(client: TestClient, monkeypatch) -> None: + email = f"loginfail-{uuid.uuid4().hex[:8]}@example.com" + _register_and_confirm(client, monkeypatch, email, "rightpass", "登录失败") + r = _login(client, email, "wrongpass") + assert r.status_code == 401 + assert "邮箱或密码错误" == r.json()["detail"] + + +def test_login_nonexistent_email(client: TestClient) -> None: + r = _login(client, "nobody@example.com", "whatever") + assert r.status_code == 401 + + +def test_login_unverified_email(client: TestClient, monkeypatch) -> None: + email = f"unverified-{uuid.uuid4().hex[:8]}@example.com" + _capture_email(monkeypatch) + _register(client, email, "unverified", "未确认") + r = _login(client, email, "unverified") + assert r.status_code == 403 + assert "邮箱尚未确认" == r.json()["detail"] + + +def test_login_disabled_user(client: TestClient, monkeypatch) -> None: + email = f"disabled-{uuid.uuid4().hex[:8]}@example.com" + info = _register_and_confirm(client, monkeypatch, email, "disabled", "禁用测试") + admin_token = _login_as_user(client, ADMIN_EMAIL, ADMIN_PASSWORD) + client.patch( + f"/api/v1/admin/users/{info['user_id']}", + headers=bearer(admin_token), + json={"is_disabled": True}, + ) + r = _login(client, email, "disabled") + assert r.status_code == 403 + assert "账号已被禁用" == r.json()["detail"] + + client.patch( + f"/api/v1/admin/users/{info['user_id']}", + headers=bearer(admin_token), + json={"is_disabled": False}, + ) + + +# --------------------------------------------------------------------------- +# refresh +# --------------------------------------------------------------------------- + + +def test_refresh_success(client: TestClient, monkeypatch) -> None: + email = f"refresh-{uuid.uuid4().hex[:8]}@example.com" + _register_and_confirm(client, monkeypatch, email, "refreshpass", "刷新测试") + r = client.post("/api/v1/auth/refresh") + assert r.status_code == 200 + assert "access_token" in r.json() + + +def test_refresh_no_cookie(client: TestClient) -> None: + session_client = TestClient(client.app) + r = session_client.post("/api/v1/auth/refresh") + assert r.status_code == 401 + assert "缺少刷新令牌" == r.json()["detail"] + + +def test_refresh_invalid_cookie(client: TestClient) -> None: + r = client.post( + "/api/v1/auth/refresh", + cookies={"opencloud_refresh": "not-a-valid-refresh-token-at-all"}, + ) + assert r.status_code == 401 + + +# --------------------------------------------------------------------------- +# logout +# --------------------------------------------------------------------------- + + +def test_logout(client: TestClient) -> None: + r = client.post("/api/v1/auth/logout") + assert r.status_code == 200 + assert r.json()["message"] == "已退出登录" + + +# --------------------------------------------------------------------------- +# forgot-password / reset-password +# --------------------------------------------------------------------------- + + +def test_forgot_password_existing_email(client: TestClient, monkeypatch) -> None: + email = f"forgot-{uuid.uuid4().hex[:8]}@example.com" + _register_and_confirm(client, monkeypatch, email, "forgotpass", "忘记密码") + captured = {} + monkeypatch.setattr(auth_router, "send_password_reset_email", captured.__setitem__) + r = client.post("/api/v1/auth/forgot-password", json={"email": email}) + assert r.status_code == 202 + assert "邮件" in r.json()["message"] + + +def test_forgot_password_nonexistent_email(client: TestClient) -> None: + r = client.post("/api/v1/auth/forgot-password", json={"email": "ghost@example.com"}) + assert r.status_code == 202 + + +def test_reset_password_valid(client: TestClient, monkeypatch) -> None: + email = f"reset-{uuid.uuid4().hex[:8]}@example.com" + _register_and_confirm(client, monkeypatch, email, "oldpass12", "重置密码") + captured = {} + monkeypatch.setattr(auth_router, "send_password_reset_email", lambda r, t: captured.update(token=t)) + + client.post("/api/v1/auth/forgot-password", json={"email": email}) + token = captured.get("token") + assert token + + r = client.post( + "/api/v1/auth/reset-password", + json={"token": token, "password": "newpass12"}, + ) + assert r.status_code == 200 + assert r.json()["message"] == "密码已重置,请重新登录" + + r = _login(client, email, "oldpass12") + assert r.status_code == 401 + r = _login(client, email, "newpass12") + assert r.status_code == 200 + + +def test_reset_password_invalid_token(client: TestClient) -> None: + r = client.post( + "/api/v1/auth/reset-password", + json={"token": "not-a-real-reset-token-ever", "password": "newpass12"}, + ) + assert r.status_code == 400 + assert "无效" in r.json()["detail"] + + +# --------------------------------------------------------------------------- +# change-password +# --------------------------------------------------------------------------- + + +def test_change_password(client: TestClient, monkeypatch) -> None: + email = f"chpw-{uuid.uuid4().hex[:8]}@example.com" + info = _register_and_confirm(client, monkeypatch, email, "oldpass12", "改密") + r = client.patch( + "/api/v1/auth/password", + headers=bearer(info["token"]), + json={"password": "updatedpass12"}, + ) + assert r.status_code == 200 + assert "access_token" in r.json() + + r = _login(client, email, "oldpass12") + assert r.status_code == 401 + r = _login(client, email, "updatedpass12") + assert r.status_code == 200 + + +# --------------------------------------------------------------------------- +# me +# --------------------------------------------------------------------------- + + +def test_me(client: TestClient, monkeypatch) -> None: + email = f"me-{uuid.uuid4().hex[:8]}@example.com" + info = _register_and_confirm(client, monkeypatch, email, "mepass12!", "我的信息") + r = client.get("/api/v1/auth/me", headers=bearer(info["token"])) + assert r.status_code == 200 + body = r.json() + assert body["user"]["email"] == email + assert body["profile"]["username"] == "我的信息" + + +def test_me_unauthenticated(client: TestClient) -> None: + r = client.get("/api/v1/auth/me") + assert r.status_code == 401 diff --git a/tests/test_cloud_types.py b/tests/test_cloud_types.py new file mode 100644 index 0000000..63e9ef8 --- /dev/null +++ b/tests/test_cloud_types.py @@ -0,0 +1,41 @@ +"""Tests for read‑only cloud‑type endpoints.""" + +from fastapi.testclient import TestClient + + +def test_list_cloud_types(client: TestClient) -> None: + r = client.get("/api/v1/cloud-types") + assert r.status_code == 200 + items = r.json() + assert len(items) >= 10 + for item in items: + for key in ("id", "name", "name_en", "genus", "rarity", "description"): + assert key in item + + +def test_get_cloud_type(client: TestClient) -> None: + r = client.get("/api/v1/cloud-types/1") + assert r.status_code == 200 + ct = r.json() + assert ct["id"] == 1 + assert ct["name"] == "积云" + assert ct["name_en"] == "Cumulus" + assert ct["rarity"] == "common" + + +def test_get_nonexistent_cloud_type(client: TestClient) -> None: + r = client.get("/api/v1/cloud-types/9999") + assert r.status_code == 404 + assert "不存在" in r.json()["detail"] + + +def test_get_cloud_type_clouds(client: TestClient) -> None: + r = client.get("/api/v1/cloud-types/1/clouds") + assert r.status_code == 200 + assert "items" in r.json() + assert "total" in r.json() + + +def test_get_cloud_type_clouds_nonexistent(client: TestClient) -> None: + r = client.get("/api/v1/cloud-types/9999/clouds") + assert r.status_code == 404 diff --git a/tests/test_clouds.py b/tests/test_clouds.py new file mode 100644 index 0000000..1d6205c --- /dev/null +++ b/tests/test_clouds.py @@ -0,0 +1,393 @@ +import uuid + +import pytest +from fastapi.testclient import TestClient + +from app.routers import auth as auth_router +from tests.conftest import bearer, png_bytes + +ADMIN_EMAIL = "admin@example.com" +ADMIN_PASSWORD = "admin-password-123" + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def _capture_email(monkeypatch) -> dict[str, str]: + captured: dict[str, str] = {} + + async def capture(recipient: str, token: str) -> None: + captured["token"] = token + + monkeypatch.setattr(auth_router, "send_confirmation_email", capture) + return captured + + +def _register_and_confirm(client, monkeypatch, email, password, username) -> dict[str, str]: + captured = _capture_email(monkeypatch) + r = client.post( + "/api/v1/auth/register", + json={"email": email, "password": password, "username": username}, + ) + assert r.status_code == 201 + client.post("/api/v1/auth/confirm-email", json={"token": captured["token"]}) + lr = client.post("/api/v1/auth/login", json={"email": email, "password": password}) + body = lr.json() + return {"token": body["access_token"], "user_id": body["user"]["id"]} + + +def _approve_cloud(client: TestClient, admin_token: str, cloud_id: str) -> None: + r = client.patch( + "/api/v1/admin/clouds/status", + headers=bearer(admin_token), + json={"ids": [cloud_id], "status": "approved"}, + ) + assert r.status_code == 200 + + +def _admin_token(client: TestClient) -> str: + r = client.post( + "/api/v1/auth/login", + json={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD}, + ) + return r.json()["access_token"] + + +def _upload_cloud(client: TestClient, token: str, **overrides): + data = { + "cloud_type_id": "1", + "is_hidden": "false", + } + data.update(overrides) + return client.post( + "/api/v1/clouds", + headers=bearer(token), + files={"image": ("cloud.png", png_bytes(), "image/png")}, + data=data, + ) + + +# --------------------------------------------------------------------------- +# create cloud +# --------------------------------------------------------------------------- + + +def test_create_cloud_with_type(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"cloud-create-{uuid.uuid4().hex[:8]}@example.com", + "cloudpass12", "云图创建者", + ) + r = _upload_cloud(client, info["token"], cloud_type_id="1") + assert r.status_code == 201 + cloud = r.json()["cloud"] + assert cloud["status"] == "pending" + assert cloud["cloud_type_id"] == 1 + assert cloud["cloud_type_name"] == "积云" + badge = r.json()["unlocked_badge"] + assert badge["cloud_type_id"] == 1 + + +def test_create_cloud_custom_type(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"custom-{uuid.uuid4().hex[:8]}@example.com", + "custompass", "自定义类型", + ) + r = _upload_cloud(client, info["token"], cloud_type_id=None, custom_cloud_type="火烧云") + assert r.status_code == 201 + cloud = r.json()["cloud"] + assert cloud["cloud_type_id"] is None + assert cloud["custom_cloud_type"] == "火烧云" + assert r.json()["unlocked_badge"] is None + + +def test_create_cloud_missing_type(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"notype-{uuid.uuid4().hex[:8]}@example.com", + "notypepass", "缺类型", + ) + r = _upload_cloud(client, info["token"], cloud_type_id=None, custom_cloud_type=None) + assert r.status_code == 422 + assert "必须选择" in r.json()["detail"] + + +def test_create_cloud_unauthenticated(client: TestClient) -> None: + r = _upload_cloud(client, "bad-token", cloud_type_id="1") + assert r.status_code == 401 + + +def test_create_cloud_lat_lon_mismatch(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"latlon-{uuid.uuid4().hex[:8]}@example.com", + "latlonpass", "坐标测试", + ) + r = _upload_cloud(client, info["token"], cloud_type_id="1", latitude="31.23", longitude=None) + assert r.status_code == 422 + assert "经纬度" in r.json()["detail"] + + +def test_create_duplicate_collection_no_badge(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"dupbadge-{uuid.uuid4().hex[:8]}@example.com", + "dupbadge", "重复徽章", + ) + r1 = _upload_cloud(client, info["token"], cloud_type_id="1") + assert r1.status_code == 201 + assert r1.json()["unlocked_badge"] is not None + + r2 = _upload_cloud(client, info["token"], cloud_type_id="1") + assert r2.status_code == 201 + assert r2.json()["unlocked_badge"] is None + + +# --------------------------------------------------------------------------- +# list gallery +# --------------------------------------------------------------------------- + + +def test_list_gallery_empty_while_no_approved(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"gallery-empty-{uuid.uuid4().hex[:8]}@example.com", + "gallerypass", "画廊空", + ) + _upload_cloud(client, info["token"], cloud_type_id="1") + r = client.get("/api/v1/clouds") + assert r.status_code == 200 + assert r.json()["total"] == 0 + + +def test_list_gallery_with_approved(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"gallery-{uuid.uuid4().hex[:8]}@example.com", + "gallerypass", "画廊有", + ) + ur = _upload_cloud(client, info["token"], cloud_type_id="1") + cloud_id = ur.json()["cloud"]["id"] + _approve_cloud(client, _admin_token(client), cloud_id) + r = client.get("/api/v1/clouds") + assert r.status_code == 200 + assert r.json()["total"] >= 1 + + +def test_list_gallery_search_by_cloud_name(client: TestClient) -> None: + r = client.get("/api/v1/clouds", params={"search": "积云"}) + assert r.status_code == 200 + for item in r.json()["items"]: + assert "积云" in item["cloud_type_name"] or "积云" in (item.get("custom_cloud_type") or "") + + +def test_list_gallery_search_by_username(client: TestClient, monkeypatch) -> None: + raw = uuid.uuid4().hex[:8] + info = _register_and_confirm( + client, monkeypatch, + f"searchuser-{raw}@example.com", "searchuser", raw, + ) + ur = _upload_cloud(client, info["token"], cloud_type_id="1") + _approve_cloud(client, _admin_token(client), ur.json()["cloud"]["id"]) + r = client.get("/api/v1/clouds", params={"search": f"@{raw}"}) + assert r.status_code == 200 + assert r.json()["total"] >= 1 + + +# --------------------------------------------------------------------------- +# get single cloud +# --------------------------------------------------------------------------- + + +def test_get_own_cloud(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"own-{uuid.uuid4().hex[:8]}@example.com", + "ownpass12", "拥有者", + ) + ur = _upload_cloud(client, info["token"], cloud_type_id="1") + cloud_id = ur.json()["cloud"]["id"] + r = client.get(f"/api/v1/clouds/{cloud_id}", headers=bearer(info["token"])) + assert r.status_code == 200 + assert r.json()["status"] == "pending" + + +def test_get_others_pending_cloud(client: TestClient, monkeypatch) -> None: + owner_info = _register_and_confirm( + client, monkeypatch, + f"owner-{uuid.uuid4().hex[:8]}@example.com", + "ownerpass1", "上传者", + ) + ur = _upload_cloud(client, owner_info["token"], cloud_type_id="1") + cloud_id = ur.json()["cloud"]["id"] + + viewer_info = _register_and_confirm( + client, monkeypatch, + f"viewer-{uuid.uuid4().hex[:8]}@example.com", + "viewerpass", "路人", + ) + r = client.get(f"/api/v1/clouds/{cloud_id}", headers=bearer(viewer_info["token"])) + assert r.status_code == 404 + + +def test_get_nonexistent_cloud(client: TestClient) -> None: + r = client.get(f"/api/v1/clouds/{uuid.uuid4()}") + assert r.status_code == 404 + + +# --------------------------------------------------------------------------- +# update cloud +# --------------------------------------------------------------------------- + + +def test_update_own_cloud(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"upd-{uuid.uuid4().hex[:8]}@example.com", + "updpass12", "更新者", + ) + ur = _upload_cloud(client, info["token"], cloud_type_id="1") + cloud_id = ur.json()["cloud"]["id"] + r = client.patch( + f"/api/v1/clouds/{cloud_id}", + headers=bearer(info["token"]), + json={"description": "更新后的描述"}, + ) + assert r.status_code == 200 + assert r.json()["description"] == "更新后的描述" + + +def test_update_others_cloud(client: TestClient, monkeypatch) -> None: + owner_info = _register_and_confirm( + client, monkeypatch, + f"upd-owner-{uuid.uuid4().hex[:8]}@example.com", + "ownerpass", "原主", + ) + ur = _upload_cloud(client, owner_info["token"], cloud_type_id="1") + cloud_id = ur.json()["cloud"]["id"] + + other_info = _register_and_confirm( + client, monkeypatch, + f"upd-other-{uuid.uuid4().hex[:8]}@example.com", + "otherpass", "别人", + ) + r = client.patch( + f"/api/v1/clouds/{cloud_id}", + headers=bearer(other_info["token"]), + json={"description": "被改了"}, + ) + assert r.status_code == 403 + + +# --------------------------------------------------------------------------- +# delete cloud +# --------------------------------------------------------------------------- + + +def test_delete_own_cloud(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"del-{uuid.uuid4().hex[:8]}@example.com", + "delpass12", "删除者", + ) + ur = _upload_cloud(client, info["token"], cloud_type_id="1") + cloud_id = ur.json()["cloud"]["id"] + r = client.delete(f"/api/v1/clouds/{cloud_id}", headers=bearer(info["token"])) + assert r.status_code == 200 + assert r.json()["deleted"] == 1 + r = client.get(f"/api/v1/clouds/{cloud_id}", headers=bearer(info["token"])) + assert r.status_code == 404 + + +def test_delete_others_cloud(client: TestClient, monkeypatch) -> None: + owner_info = _register_and_confirm( + client, monkeypatch, + f"del-own-{uuid.uuid4().hex[:8]}@example.com", + "ownerpass", "物主", + ) + ur = _upload_cloud(client, owner_info["token"], cloud_type_id="1") + cloud_id = ur.json()["cloud"]["id"] + + other_info = _register_and_confirm( + client, monkeypatch, + f"del-oth-{uuid.uuid4().hex[:8]}@example.com", + "otherpass", "外人", + ) + r = client.delete(f"/api/v1/clouds/{cloud_id}", headers=bearer(other_info["token"])) + assert r.status_code == 403 + + +# --------------------------------------------------------------------------- +# batch delete +# --------------------------------------------------------------------------- + + +def test_batch_delete(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"batch-{uuid.uuid4().hex[:8]}@example.com", + "batchpass", "批量删除", + ) + u1 = _upload_cloud(client, info["token"], cloud_type_id="1") + u2 = _upload_cloud(client, info["token"], cloud_type_id="2") + ids = [u1.json()["cloud"]["id"], u2.json()["cloud"]["id"]] + r = client.post( + "/api/v1/clouds/batch-delete", + headers=bearer(info["token"]), + json={"ids": ids}, + ) + assert r.status_code == 200 + assert r.json()["deleted"] == 2 + + +def test_batch_delete_partial_invalid(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"batch2-{uuid.uuid4().hex[:8]}@example.com", + "batch2pass", "批量部分", + ) + u1 = _upload_cloud(client, info["token"], cloud_type_id="1") + ids = [u1.json()["cloud"]["id"], str(uuid.uuid4())] + r = client.post( + "/api/v1/clouds/batch-delete", + headers=bearer(info["token"]), + json={"ids": ids}, + ) + assert r.status_code == 404 + + +# --------------------------------------------------------------------------- +# map +# --------------------------------------------------------------------------- + + +def test_map_clouds(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"map-{uuid.uuid4().hex[:8]}@example.com", + "mappass12", "地图用户", + ) + ur = _upload_cloud( + client, info["token"], + cloud_type_id="1", latitude="31.23", longitude="121.47", + captured_at="2025-01-01T00:00:00", + ) + _approve_cloud(client, _admin_token(client), ur.json()["cloud"]["id"]) + r = client.get("/api/v1/clouds/map", params={ + "start": "2024-12-31T00:00:00", + "end": "2026-01-01T00:00:00", + }) + assert r.status_code == 200 + assert len(r.json()) >= 1 + + +def test_map_invalid_time_range(client: TestClient) -> None: + r = client.get("/api/v1/clouds/map", params={ + "start": "2025-01-02T00:00:00", + "end": "2025-01-01T00:00:00", + }) + assert r.status_code == 422 + assert "晚于开始" in r.json()["detail"] diff --git a/tests/test_collections.py b/tests/test_collections.py new file mode 100644 index 0000000..45b104d --- /dev/null +++ b/tests/test_collections.py @@ -0,0 +1,62 @@ +import uuid + +import pytest +from fastapi.testclient import TestClient + +from app.routers import auth as auth_router +from tests.conftest import bearer, png_bytes + + +def _capture_email(monkeypatch) -> dict[str, str]: + captured: dict[str, str] = {} + + async def capture(recipient: str, token: str) -> None: + captured["token"] = token + + monkeypatch.setattr(auth_router, "send_confirmation_email", capture) + return captured + + +def _register_and_confirm(client, monkeypatch, email, password, username) -> dict[str, str]: + captured = _capture_email(monkeypatch) + r = client.post( + "/api/v1/auth/register", + json={"email": email, "password": password, "username": username}, + ) + assert r.status_code == 201 + client.post("/api/v1/auth/confirm-email", json={"token": captured["token"]}) + lr = client.post("/api/v1/auth/login", json={"email": email, "password": password}) + body = lr.json() + return {"token": body["access_token"], "user_id": body["user"]["id"]} + + +# --------------------------------------------------------------------------- +# my collections +# --------------------------------------------------------------------------- + + +def test_my_collections_after_upload(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"coll-{uuid.uuid4().hex[:8]}@example.com", + "collpass1", "收藏用户", + ) + r = client.get("/api/v1/collections/me", headers=bearer(info["token"])) + assert r.status_code == 200 + assert len(r.json()) == 0 + + client.post( + "/api/v1/clouds", + headers=bearer(info["token"]), + files={"image": ("cloud.png", png_bytes(), "image/png")}, + data={"cloud_type_id": "1", "is_hidden": "false"}, + ) + r = client.get("/api/v1/collections/me", headers=bearer(info["token"])) + assert r.status_code == 200 + assert len(r.json()) == 1 + assert r.json()[0]["cloud_type_id"] == 1 + + +def test_my_collections_unauthenticated(client: TestClient) -> None: + r = client.get("/api/v1/collections/me") + assert r.status_code == 401 diff --git a/tests/test_profiles.py b/tests/test_profiles.py new file mode 100644 index 0000000..e4d03b2 --- /dev/null +++ b/tests/test_profiles.py @@ -0,0 +1,171 @@ +import uuid + +import pytest +from fastapi.testclient import TestClient + +from app.routers import auth as auth_router +from tests.conftest import bearer, png_bytes + +ADMIN_EMAIL = "admin@example.com" +ADMIN_PASSWORD = "admin-password-123" + + +def _capture_email(monkeypatch) -> dict[str, str]: + captured: dict[str, str] = {} + + async def capture(recipient: str, token: str) -> None: + captured["token"] = token + + monkeypatch.setattr(auth_router, "send_confirmation_email", capture) + return captured + + +def _register_and_confirm(client, monkeypatch, email, password, username) -> dict[str, str]: + captured = _capture_email(monkeypatch) + r = client.post( + "/api/v1/auth/register", + json={"email": email, "password": password, "username": username}, + ) + assert r.status_code == 201 + client.post("/api/v1/auth/confirm-email", json={"token": captured["token"]}) + lr = client.post("/api/v1/auth/login", json={"email": email, "password": password}) + body = lr.json() + return {"token": body["access_token"], "user_id": body["user"]["id"]} + + +def _admin_token(client: TestClient) -> str: + r = client.post( + "/api/v1/auth/login", + json={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD}, + ) + return r.json()["access_token"] + + +def _upload_and_approve(client: TestClient, token: str) -> str: + ur = client.post( + "/api/v1/clouds", + headers=bearer(token), + files={"image": ("cloud.png", png_bytes(), "image/png")}, + data={"cloud_type_id": "1", "is_hidden": "false"}, + ) + cloud_id = ur.json()["cloud"]["id"] + client.patch( + "/api/v1/admin/clouds/status", + headers=bearer(_admin_token(client)), + json={"ids": [cloud_id], "status": "approved"}, + ) + return cloud_id + + +# --------------------------------------------------------------------------- +# get profile +# --------------------------------------------------------------------------- + + +def test_get_profile(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"prof-{uuid.uuid4().hex[:8]}@example.com", + "profpass1", "资料展示", + ) + r = client.get(f"/api/v1/profiles/{info['user_id']}") + assert r.status_code == 200 + assert r.json()["username"] == "资料展示" + + +def test_get_nonexistent_profile(client: TestClient) -> None: + r = client.get(f"/api/v1/profiles/{uuid.uuid4()}") + assert r.status_code == 404 + + +# --------------------------------------------------------------------------- +# update my profile +# --------------------------------------------------------------------------- + + +def test_update_me(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"updme-{uuid.uuid4().hex[:8]}@example.com", + "updmepass", "原昵称", + ) + r = client.patch( + "/api/v1/profiles/me", + headers=bearer(info["token"]), + json={"username": "新昵称"}, + ) + assert r.status_code == 200 + assert r.json()["username"] == "新昵称" + + +def test_update_me_short_username(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"shortname-{uuid.uuid4().hex[:8]}@example.com", + "shortname", "短名字测试", + ) + r = client.patch( + "/api/v1/profiles/me", + headers=bearer(info["token"]), + json={"username": "A"}, + ) + assert r.status_code == 422 + + +def test_update_me_duplicate_username(client: TestClient, monkeypatch) -> None: + _register_and_confirm( + client, monkeypatch, + f"dupname-u1-{uuid.uuid4().hex[:8]}@example.com", + "pass11112", "已占用的名字", + ) + info2 = _register_and_confirm( + client, monkeypatch, + f"dupname-u2-{uuid.uuid4().hex[:8]}@example.com", + "pass22223", "第二人", + ) + r = client.patch( + "/api/v1/profiles/me", + headers=bearer(info2["token"]), + json={"username": "已占用的名字"}, + ) + assert r.status_code == 409 + assert "昵称" in r.json()["detail"] + + +# --------------------------------------------------------------------------- +# get user clouds +# --------------------------------------------------------------------------- + + +def test_get_user_clouds_public(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"pubcloud-{uuid.uuid4().hex[:8]}@example.com", + "pubcloud1", "公开云主", + ) + _upload_and_approve(client, info["token"]) + r = client.get(f"/api/v1/profiles/{info['user_id']}/clouds") + assert r.status_code == 200 + assert r.json()["total"] >= 1 + for item in r.json()["items"]: + assert item["status"] == "approved" + + +def test_get_user_clouds_as_owner(client: TestClient, monkeypatch) -> None: + info = _register_and_confirm( + client, monkeypatch, + f"owncloud-{uuid.uuid4().hex[:8]}@example.com", + "owncloud1", "私有云主", + ) + cloud_id = _upload_and_approve(client, info["token"]) + r = client.get( + f"/api/v1/profiles/{info['user_id']}/clouds", + headers=bearer(info["token"]), + ) + assert r.status_code == 200 + assert r.json()["total"] >= 1 + + +def test_get_nonexistent_user_clouds(client: TestClient) -> None: + r = client.get(f"/api/v1/profiles/{uuid.uuid4()}/clouds") + assert r.status_code == 404