From 12d15878da75a104e085a604c5617f928abc06f9 Mon Sep 17 00:00:00 2001 From: Mplan Date: Sun, 26 Jul 2026 01:23:12 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=9F=E4=B8=80=E6=B5=8B=E8=AF=95=E8=BE=85?= =?UTF-8?q?=E5=8A=A9=E5=87=BD=E6=95=B0=E5=88=B0=20conftest.py=EF=BC=8C?= =?UTF-8?q?=E6=B6=88=E9=99=A4=E4=BB=A3=E7=A0=81=E9=87=8D=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/conftest.py | 117 ++++++++++++++++++++++++---- tests/test_admin.py | 115 +++++++++------------------ tests/test_api.py | 43 +++-------- tests/test_auth.py | 140 +++++++++++---------------------- tests/test_clouds.py | 159 ++++++++++++-------------------------- tests/test_collections.py | 34 +------- tests/test_profiles.py | 89 +++++++-------------- 7 files changed, 276 insertions(+), 421 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 36013e7..76f8f6f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,6 +16,31 @@ 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" +ADMIN_EMAIL = "admin@example.com" +ADMIN_PASSWORD = "admin-password-123" + +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" +os.environ["PUBLIC_BASE_URL"] = "http://testserver" +os.environ["FRONTEND_URL"] = "http://testserver" +os.environ["CORS_ORIGINS"] = "http://testserver" +os.environ["UPLOAD_DIR"] = str(TEST_UPLOADS) +os.environ["EMAIL_DELIVERY_MODE"] = "console" +os.environ["ADMIN_EMAIL"] = ADMIN_EMAIL +os.environ["ADMIN_PASSWORD"] = ADMIN_PASSWORD +os.environ["ADMIN_USERNAME"] = "测试管理员" + +from app.database import Base, engine # noqa: E402 +from app.main import app # noqa: E402 +from app.routers import auth as auth_router # noqa: E402 +from app.seed import seed # noqa: E402 + + +# --------------------------------------------------------------------------- +# shared helpers — usable from any test file +# --------------------------------------------------------------------------- + def bearer(token: str) -> dict[str, str]: return {"Authorization": f"Bearer {token}"} @@ -26,21 +51,85 @@ def png_bytes() -> bytes: 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" -os.environ["PUBLIC_BASE_URL"] = "http://testserver" -os.environ["FRONTEND_URL"] = "http://testserver" -os.environ["CORS_ORIGINS"] = "http://testserver" -os.environ["UPLOAD_DIR"] = str(TEST_UPLOADS) -os.environ["EMAIL_DELIVERY_MODE"] = "console" -os.environ["ADMIN_EMAIL"] = "admin@example.com" -os.environ["ADMIN_PASSWORD"] = "admin-password-123" -os.environ["ADMIN_USERNAME"] = "测试管理员" -from app.database import Base, engine # noqa: E402 -from app.main import app # noqa: E402 -from app.seed import seed # noqa: E402 +def capture_email(monkeypatch) -> dict[str, str]: + """Patch send_confirmation_email and return a dict that will hold the token.""" + 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_user(client: TestClient, email: str, password: str, username: str): + return client.post( + "/api/v1/auth/register", + json={"email": email, "password": password, "username": username}, + ) + + +def confirm_email(client: TestClient, token: str): + return client.post("/api/v1/auth/confirm-email", json={"token": token}) + + +def login_user(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]: + """Create a fully registered + confirmed user. Returns token & user_id.""" + captured = capture_email(monkeypatch) + r = register_user(client, email, password, username) + assert r.status_code == 201 + token = captured.get("token") + assert token + cr = confirm_email(client, token) + assert cr.status_code == 200 + lr = login_user(client, email, password) + assert lr.status_code == 200 + body = lr.json() + return {"token": body["access_token"], "user_id": body["user"]["id"]} + + +def admin_token(client: TestClient) -> str: + r = login_user(client, ADMIN_EMAIL, 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, + ) + + +def approve_cloud(client: TestClient, admin_tok: str, cloud_id: str) -> None: + r = client.patch( + "/api/v1/admin/clouds/status", + headers=bearer(admin_tok), + json={"ids": [cloud_id], "status": "approved"}, + ) + assert r.status_code == 200 + + +# --------------------------------------------------------------------------- +# fixtures +# --------------------------------------------------------------------------- async def _prepare_database() -> None: diff --git a/tests/test_admin.py b/tests/test_admin.py index cbda5ae..fa74ded 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -3,53 +3,13 @@ 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, - ) +from tests.conftest import ( + admin_token, + bearer, + png_bytes, + register_and_confirm, + upload_cloud, +) # --------------------------------------------------------------------------- @@ -58,7 +18,7 @@ def _upload_cloud(client: TestClient, token: str, **overrides): def test_stats(client: TestClient) -> None: - r = client.get("/api/v1/admin/stats", headers=bearer(_admin_token(client))) + 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"): @@ -66,7 +26,7 @@ def test_stats(client: TestClient) -> None: def test_stats_non_admin(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"notadmin-{uuid.uuid4().hex[:8]}@example.com", "notadmin12", "非管理员", @@ -87,7 +47,7 @@ def test_stats_unauthenticated(client: TestClient) -> None: def test_list_users(client: TestClient) -> None: - r = client.get("/api/v1/admin/users", headers=bearer(_admin_token(client))) + r = client.get("/api/v1/admin/users", headers=bearer(admin_token(client))) assert r.status_code == 200 assert r.json()["total"] >= 1 @@ -98,15 +58,14 @@ def test_list_users(client: TestClient) -> None: def test_update_user_role(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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), + headers=bearer(admin_token(client)), json={"role": "admin"}, ) assert r.status_code == 200 @@ -114,15 +73,15 @@ def test_update_user_role(client: TestClient, monkeypatch) -> None: def test_update_user_disable(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"disableme-{uuid.uuid4().hex[:8]}@example.com", "disableme", "被禁用", ) - admin = _admin_token(client) + adm = admin_token(client) r = client.patch( f"/api/v1/admin/users/{info['user_id']}", - headers=bearer(admin), + headers=bearer(adm), json={"is_disabled": True}, ) assert r.status_code == 200 @@ -130,18 +89,18 @@ def test_update_user_disable(client: TestClient, monkeypatch) -> None: client.patch( f"/api/v1/admin/users/{info['user_id']}", - headers=bearer(admin), + headers=bearer(adm), 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)) + 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(admin_token), + headers=bearer(adm), json={"is_disabled": True}, ) assert r.status_code == 400 @@ -149,12 +108,12 @@ def test_update_self_disable(client: TestClient) -> None: def test_update_self_demote(client: TestClient) -> None: - admin_token = _admin_token(client) - me = client.get("/api/v1/auth/me", headers=bearer(admin_token)) + 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(admin_token), + headers=bearer(adm), json={"role": "user"}, ) assert r.status_code == 400 @@ -164,7 +123,7 @@ def test_update_self_demote(client: TestClient) -> None: def test_update_nonexistent_user(client: TestClient) -> None: r = client.patch( f"/api/v1/admin/users/{uuid.uuid4()}", - headers=bearer(_admin_token(client)), + headers=bearer(admin_token(client)), json={"role": "admin"}, ) assert r.status_code == 404 @@ -176,7 +135,7 @@ def test_update_nonexistent_user(client: TestClient) -> None: def test_list_admin_clouds(client: TestClient) -> None: - r = client.get("/api/v1/admin/clouds", headers=bearer(_admin_token(client))) + r = client.get("/api/v1/admin/clouds", headers=bearer(admin_token(client))) assert r.status_code == 200 assert "items" in r.json() @@ -184,7 +143,7 @@ def test_list_admin_clouds(client: TestClient) -> None: def test_list_admin_clouds_filter_status(client: TestClient) -> None: r = client.get( "/api/v1/admin/clouds", - headers=bearer(_admin_token(client)), + headers=bearer(admin_token(client)), params={"status": "pending"}, ) assert r.status_code == 200 @@ -195,7 +154,7 @@ def test_list_admin_clouds_filter_status(client: TestClient) -> None: def test_list_admin_clouds_filter_hidden(client: TestClient) -> None: r = client.get( "/api/v1/admin/clouds", - headers=bearer(_admin_token(client)), + headers=bearer(admin_token(client)), params={"is_hidden": "true"}, ) assert r.status_code == 200 @@ -207,16 +166,16 @@ def test_list_admin_clouds_filter_hidden(client: TestClient) -> None: def test_update_cloud_status(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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)), + headers=bearer(admin_token(client)), json={"ids": [cloud_id], "status": "rejected"}, ) assert r.status_code == 200 @@ -226,7 +185,7 @@ def test_update_cloud_status(client: TestClient, monkeypatch) -> None: def test_update_cloud_status_nonexistent(client: TestClient) -> None: r = client.patch( "/api/v1/admin/clouds/status", - headers=bearer(_admin_token(client)), + headers=bearer(admin_token(client)), json={"ids": [str(uuid.uuid4())], "status": "approved"}, ) assert r.status_code == 404 @@ -238,16 +197,16 @@ def test_update_cloud_status_nonexistent(client: TestClient) -> None: def test_update_cloud_visibility(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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)), + headers=bearer(admin_token(client)), json={"ids": [cloud_id], "is_hidden": True}, ) assert r.status_code == 200 @@ -258,7 +217,7 @@ def test_update_cloud_visibility(client: TestClient, monkeypatch) -> None: client.patch( "/api/v1/admin/clouds/visibility", - headers=bearer(_admin_token(client)), + headers=bearer(admin_token(client)), json={"ids": [cloud_id], "is_hidden": False}, ) @@ -269,16 +228,16 @@ def test_update_cloud_visibility(client: TestClient, monkeypatch) -> None: def test_admin_batch_delete_clouds(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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)), + headers=bearer(admin_token(client)), json={"ids": [cloud_id]}, ) assert r.status_code == 200 diff --git a/tests/test_api.py b/tests/test_api.py index 1a12e5e..f662cd4 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,19 +1,6 @@ -import io - from fastapi.testclient import TestClient -from PIL import Image -from app.routers import auth as auth_router - - -def _png_bytes() -> bytes: - output = io.BytesIO() - Image.new("RGB", (1200, 800), color=(125, 196, 240)).save(output, format="PNG") - return output.getvalue() - - -def _bearer(token: str) -> dict[str, str]: - return {"Authorization": f"Bearer {token}"} +from tests.conftest import bearer, capture_email, confirm_email, png_bytes def test_openapi_contains_all_feature_groups(client: TestClient) -> None: @@ -34,12 +21,7 @@ def test_openapi_contains_all_feature_groups(client: TestClient) -> None: def test_complete_user_cloud_and_admin_flow(client: TestClient, monkeypatch) -> None: - captured_tokens: dict[str, str] = {} - - async def capture_confirmation(_: str, token: str) -> None: - captured_tokens["confirmation"] = token - - monkeypatch.setattr(auth_router, "send_confirmation_email", capture_confirmation) + captured_tokens = capture_email(monkeypatch) health = client.get("/api/v1/health") assert health.status_code == 200 @@ -53,12 +35,9 @@ def test_complete_user_cloud_and_admin_flow(client: TestClient, monkeypatch) -> }, ) assert registered.status_code == 201 - assert "confirmation" in captured_tokens + assert "token" in captured_tokens - confirmed = client.post( - "/api/v1/auth/confirm-email", - json={"token": captured_tokens["confirmation"]}, - ) + confirmed = confirm_email(client, captured_tokens["token"]) assert confirmed.status_code == 200 login = client.post( @@ -69,21 +48,21 @@ def test_complete_user_cloud_and_admin_flow(client: TestClient, monkeypatch) -> user_token = login.json()["access_token"] user_id = login.json()["user"]["id"] - me = client.get("/api/v1/auth/me", headers=_bearer(user_token)) + me = client.get("/api/v1/auth/me", headers=bearer(user_token)) assert me.status_code == 200 assert me.json()["profile"]["username"] == "观云者" renamed = client.patch( "/api/v1/profiles/me", - headers=_bearer(user_token), + headers=bearer(user_token), json={"username": "天空观察员"}, ) assert renamed.status_code == 200 upload = client.post( "/api/v1/clouds", - headers=_bearer(user_token), - files={"image": ("cloud.png", _png_bytes(), "image/png")}, + headers=bearer(user_token), + files={"image": ("cloud.png", png_bytes(), "image/png")}, data={ "cloud_type_id": "1", "latitude": "31.2345", @@ -101,7 +80,7 @@ def test_complete_user_cloud_and_admin_flow(client: TestClient, monkeypatch) -> assert cloud["latitude"] == 31.23 assert upload.json()["unlocked_badge"]["cloud_type_id"] == 1 - collection = client.get("/api/v1/collections/me", headers=_bearer(user_token)) + collection = client.get("/api/v1/collections/me", headers=bearer(user_token)) assert collection.status_code == 200 assert len(collection.json()) == 1 @@ -116,7 +95,7 @@ def test_complete_user_cloud_and_admin_flow(client: TestClient, monkeypatch) -> admin_token = admin_login.json()["access_token"] approved = client.patch( "/api/v1/admin/clouds/status", - headers=_bearer(admin_token), + headers=bearer(admin_token), json={"ids": [cloud_id], "status": "approved"}, ) assert approved.status_code == 200 @@ -140,7 +119,7 @@ def test_complete_user_cloud_and_admin_flow(client: TestClient, monkeypatch) -> assert public_profile_clouds.status_code == 200 assert public_profile_clouds.json()["total"] == 1 - deleted = client.delete(f"/api/v1/clouds/{cloud_id}", headers=_bearer(user_token)) + deleted = client.delete(f"/api/v1/clouds/{cloud_id}", headers=bearer(user_token)) assert deleted.status_code == 200 assert deleted.json()["deleted"] == 1 diff --git a/tests/test_auth.py b/tests/test_auth.py index c4f95af..fcbdff2 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -4,68 +4,14 @@ 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"] +from tests.conftest import ( + bearer, + capture_email, + confirm_email, + login_user, + register_and_confirm, + register_user, +) # --------------------------------------------------------------------------- @@ -75,8 +21,8 @@ def _login_as_user(client: TestClient, email: str, password: str) -> str: 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", "注册测试") + captured = capture_email(monkeypatch) + r = register_user(client, email, "regpassword", "注册测试") assert r.status_code == 201 assert r.json()["message"] == "注册成功，请查收确认邮件" assert "token" in captured @@ -84,8 +30,8 @@ def test_register_success(client: TestClient, monkeypatch) -> None: 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", "第二个用户") + register_user(client, email, "firstpass", "第一个用户") + r = register_user(client, email, "secondpass", "第二个用户") assert r.status_code == 409 assert "该邮箱已被注册" == r.json()["detail"] @@ -93,24 +39,24 @@ def test_register_duplicate_email(client: TestClient, monkeypatch) -> None: 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", "同名用户") + register_user(client, email_a, "passA123", "同名用户") + r = register_user(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", "短密码") + r = register_user(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") + r = register_user(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", "坏邮箱") + r = register_user(client, "not-an-email", "pass12345", "坏邮箱") assert r.status_code == 422 @@ -121,18 +67,18 @@ def test_register_invalid_email(client: TestClient) -> None: 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", "确认测试") + captured = capture_email(monkeypatch) + r = register_user(client, email, "confirmpass", "确认测试") assert r.status_code == 201 token = captured.get("token") assert token - cr = _confirm(client, token) + cr = confirm_email(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") + r = confirm_email(client, "not-a-valid-token-at-all-never") assert r.status_code == 400 assert "无效" in r.json()["detail"] @@ -144,49 +90,51 @@ def test_confirm_email_invalid_token(client: TestClient) -> None: 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", "登录测试") + 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") + register_and_confirm(client, monkeypatch, email, "rightpass", "登录失败") + r = login_user(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") + r = login_user(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") + capture_email(monkeypatch) + register_user(client, email, "unverified", "未确认") + r = login_user(client, email, "unverified") assert r.status_code == 403 assert "邮箱尚未确认" == r.json()["detail"] def test_login_disabled_user(client: TestClient, monkeypatch) -> None: + from tests.conftest import ADMIN_EMAIL, ADMIN_PASSWORD + 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) + info = register_and_confirm(client, monkeypatch, email, "disabled", "禁用测试") + admin_tok = login_user(client, ADMIN_EMAIL, ADMIN_PASSWORD).json()["access_token"] client.patch( f"/api/v1/admin/users/{info['user_id']}", - headers=bearer(admin_token), + headers=bearer(admin_tok), json={"is_disabled": True}, ) - r = _login(client, email, "disabled") + r = login_user(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), + headers=bearer(admin_tok), json={"is_disabled": False}, ) @@ -198,7 +146,7 @@ def test_login_disabled_user(client: TestClient, monkeypatch) -> None: def test_refresh_success(client: TestClient, monkeypatch) -> None: email = f"refresh-{uuid.uuid4().hex[:8]}@example.com" - _register_and_confirm(client, monkeypatch, email, "refreshpass", "刷新测试") + 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() @@ -237,7 +185,7 @@ def test_logout(client: TestClient) -> None: 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", "忘记密码") + 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}) @@ -252,7 +200,7 @@ def test_forgot_password_nonexistent_email(client: TestClient) -> None: 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", "重置密码") + register_and_confirm(client, monkeypatch, email, "oldpass12", "重置密码") captured = {} monkeypatch.setattr(auth_router, "send_password_reset_email", lambda r, t: captured.update(token=t)) @@ -267,9 +215,9 @@ def test_reset_password_valid(client: TestClient, monkeypatch) -> None: assert r.status_code == 200 assert r.json()["message"] == "密码已重置，请重新登录" - r = _login(client, email, "oldpass12") + r = login_user(client, email, "oldpass12") assert r.status_code == 401 - r = _login(client, email, "newpass12") + r = login_user(client, email, "newpass12") assert r.status_code == 200 @@ -289,7 +237,7 @@ def test_reset_password_invalid_token(client: TestClient) -> None: 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", "改密") + info = register_and_confirm(client, monkeypatch, email, "oldpass12", "改密") r = client.patch( "/api/v1/auth/password", headers=bearer(info["token"]), @@ -298,9 +246,9 @@ def test_change_password(client: TestClient, monkeypatch) -> None: assert r.status_code == 200 assert "access_token" in r.json() - r = _login(client, email, "oldpass12") + r = login_user(client, email, "oldpass12") assert r.status_code == 401 - r = _login(client, email, "updatedpass12") + r = login_user(client, email, "updatedpass12") assert r.status_code == 200 @@ -311,7 +259,7 @@ def test_change_password(client: TestClient, monkeypatch) -> None: def test_me(client: TestClient, monkeypatch) -> None: email = f"me-{uuid.uuid4().hex[:8]}@example.com" - info = _register_and_confirm(client, monkeypatch, email, "mepass12!", "我的信息") + 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() diff --git a/tests/test_clouds.py b/tests/test_clouds.py index 1d6205c..ede2746 100644 --- a/tests/test_clouds.py +++ b/tests/test_clouds.py @@ -3,70 +3,14 @@ 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, - ) +from tests.conftest import ( + admin_token, + approve_cloud, + bearer, + png_bytes, + register_and_confirm, + upload_cloud, +) # --------------------------------------------------------------------------- @@ -75,12 +19,12 @@ def _upload_cloud(client: TestClient, token: str, **overrides): def test_create_cloud_with_type(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + r = upload_cloud(client, info["token"], cloud_type_id="1") assert r.status_code == 201 cloud = r.json()["cloud"] assert cloud["status"] == "pending" @@ -91,12 +35,12 @@ def test_create_cloud_with_type(client: TestClient, monkeypatch) -> None: def test_create_cloud_custom_type(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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="火烧云") + 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 @@ -105,43 +49,43 @@ def test_create_cloud_custom_type(client: TestClient, monkeypatch) -> None: def test_create_cloud_missing_type(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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) + 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") + 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( + 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) + 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( + 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") + 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") + r2 = upload_cloud(client, info["token"], cloud_type_id="1") assert r2.status_code == 201 assert r2.json()["unlocked_badge"] is None @@ -152,26 +96,25 @@ def test_create_duplicate_collection_no_badge(client: TestClient, monkeypatch) - def test_list_gallery_empty_while_no_approved(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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( + 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) + 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") assert r.status_code == 200 assert r.json()["total"] >= 1 @@ -186,12 +129,12 @@ def test_list_gallery_search_by_cloud_name(client: TestClient) -> None: def test_list_gallery_search_by_username(client: TestClient, monkeypatch) -> None: raw = uuid.uuid4().hex[:8] - info = _register_and_confirm( + 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"]) + 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 @@ -203,12 +146,12 @@ def test_list_gallery_search_by_username(client: TestClient, monkeypatch) -> Non def test_get_own_cloud(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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 @@ -216,15 +159,15 @@ def test_get_own_cloud(client: TestClient, monkeypatch) -> None: def test_get_others_pending_cloud(client: TestClient, monkeypatch) -> None: - owner_info = _register_and_confirm( + 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") + ur = upload_cloud(client, owner_info["token"], cloud_type_id="1") cloud_id = ur.json()["cloud"]["id"] - viewer_info = _register_and_confirm( + viewer_info = register_and_confirm( client, monkeypatch, f"viewer-{uuid.uuid4().hex[:8]}@example.com", "viewerpass", "路人", @@ -244,12 +187,12 @@ def test_get_nonexistent_cloud(client: TestClient) -> None: def test_update_own_cloud(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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}", @@ -261,15 +204,15 @@ def test_update_own_cloud(client: TestClient, monkeypatch) -> None: def test_update_others_cloud(client: TestClient, monkeypatch) -> None: - owner_info = _register_and_confirm( + 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") + ur = upload_cloud(client, owner_info["token"], cloud_type_id="1") cloud_id = ur.json()["cloud"]["id"] - other_info = _register_and_confirm( + other_info = register_and_confirm( client, monkeypatch, f"upd-other-{uuid.uuid4().hex[:8]}@example.com", "otherpass", "别人", @@ -288,12 +231,12 @@ def test_update_others_cloud(client: TestClient, monkeypatch) -> None: def test_delete_own_cloud(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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 @@ -303,15 +246,15 @@ def test_delete_own_cloud(client: TestClient, monkeypatch) -> None: def test_delete_others_cloud(client: TestClient, monkeypatch) -> None: - owner_info = _register_and_confirm( + 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") + ur = upload_cloud(client, owner_info["token"], cloud_type_id="1") cloud_id = ur.json()["cloud"]["id"] - other_info = _register_and_confirm( + other_info = register_and_confirm( client, monkeypatch, f"del-oth-{uuid.uuid4().hex[:8]}@example.com", "otherpass", "外人", @@ -326,13 +269,13 @@ def test_delete_others_cloud(client: TestClient, monkeypatch) -> None: def test_batch_delete(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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", @@ -344,12 +287,12 @@ def test_batch_delete(client: TestClient, monkeypatch) -> None: def test_batch_delete_partial_invalid(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + 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") + 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", @@ -365,17 +308,17 @@ def test_batch_delete_partial_invalid(client: TestClient, monkeypatch) -> None: def test_map_clouds(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"map-{uuid.uuid4().hex[:8]}@example.com", "mappass12", "地图用户", ) - ur = _upload_cloud( + 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"]) + 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", diff --git a/tests/test_collections.py b/tests/test_collections.py index 45b104d..d6c7756 100644 --- a/tests/test_collections.py +++ b/tests/test_collections.py @@ -1,42 +1,12 @@ 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 -# --------------------------------------------------------------------------- +from tests.conftest import bearer, png_bytes, register_and_confirm def test_my_collections_after_upload(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"coll-{uuid.uuid4().hex[:8]}@example.com", "collpass1", "收藏用户", diff --git a/tests/test_profiles.py b/tests/test_profiles.py index e4d03b2..6ef1db2 100644 --- a/tests/test_profiles.py +++ b/tests/test_profiles.py @@ -3,58 +3,13 @@ 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 +from tests.conftest import ( + admin_token, + approve_cloud, + bearer, + png_bytes, + register_and_confirm, +) # --------------------------------------------------------------------------- @@ -63,7 +18,7 @@ def _upload_and_approve(client: TestClient, token: str) -> str: def test_get_profile(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"prof-{uuid.uuid4().hex[:8]}@example.com", "profpass1", "资料展示", @@ -84,7 +39,7 @@ def test_get_nonexistent_profile(client: TestClient) -> None: def test_update_me(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"updme-{uuid.uuid4().hex[:8]}@example.com", "updmepass", "原昵称", @@ -99,7 +54,7 @@ def test_update_me(client: TestClient, monkeypatch) -> None: def test_update_me_short_username(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"shortname-{uuid.uuid4().hex[:8]}@example.com", "shortname", "短名字测试", @@ -113,12 +68,12 @@ def test_update_me_short_username(client: TestClient, monkeypatch) -> None: def test_update_me_duplicate_username(client: TestClient, monkeypatch) -> None: - _register_and_confirm( + register_and_confirm( client, monkeypatch, f"dupname-u1-{uuid.uuid4().hex[:8]}@example.com", "pass11112", "已占用的名字", ) - info2 = _register_and_confirm( + info2 = register_and_confirm( client, monkeypatch, f"dupname-u2-{uuid.uuid4().hex[:8]}@example.com", "pass22223", "第二人", @@ -138,12 +93,18 @@ def test_update_me_duplicate_username(client: TestClient, monkeypatch) -> None: def test_get_user_clouds_public(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"pubcloud-{uuid.uuid4().hex[:8]}@example.com", "pubcloud1", "公开云主", ) - _upload_and_approve(client, info["token"]) + ur = 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"}, + ) + approve_cloud(client, admin_token(client), ur.json()["cloud"]["id"]) r = client.get(f"/api/v1/profiles/{info['user_id']}/clouds") assert r.status_code == 200 assert r.json()["total"] >= 1 @@ -152,12 +113,18 @@ def test_get_user_clouds_public(client: TestClient, monkeypatch) -> None: def test_get_user_clouds_as_owner(client: TestClient, monkeypatch) -> None: - info = _register_and_confirm( + info = register_and_confirm( client, monkeypatch, f"owncloud-{uuid.uuid4().hex[:8]}@example.com", "owncloud1", "私有云主", ) - cloud_id = _upload_and_approve(client, info["token"]) + ur = 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"}, + ) + approve_cloud(client, admin_token(client), ur.json()["cloud"]["id"]) r = client.get( f"/api/v1/profiles/{info['user_id']}/clouds", headers=bearer(info["token"]),