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"]