import os import shutil from pathlib import Path from tempfile import mkdtemp from uuid import uuid4 test_db_dir = Path(mkdtemp(prefix="dating-app-tests-")) test_db_path = test_db_dir / "test.db" os.environ["DATABASE_URL"] = f"sqlite:///{test_db_path.as_posix()}" from fastapi.testclient import TestClient import pytest from app.database import SessionLocal, engine from app.main import app from app.models import Match, Swipe, User, UserStatus from app.storage import StorageService @pytest.fixture(scope="session", autouse=True) def cleanup_test_database(): yield engine.dispose() shutil.rmtree(test_db_dir, ignore_errors=True) def register_user( client, *, full_name=None, bio=None, city=None, country=None, interests=None, relationship_goal="serious_relationship", ): email = f"tester-{uuid4().hex[:10]}@example.com" password = "password123" response = client.post("/api/v1/auth/register", json={ "email_or_phone": email, "password": password, "confirm_password": password, "relationship_goal": relationship_goal, "full_name": full_name, "bio": bio, "city": city, "country": country, "interests": interests, }) assert response.status_code == 201 token = response.json()["access_token"] me_response = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"}) assert me_response.status_code == 200 return me_response.json()["id"], token, email, password def test_private_oss_media_uses_browser_ready_signed_url(): calls = [] class FakeBucket: def sign_url(self, method, key, expires): calls.append((method, key, expires)) return f"https://oss.example.test/{key}?signed=1" service = StorageService.__new__(StorageService) service.enabled = True service.bucket = FakeBucket() assert service.access_url("oss://melove/users/2/avatar.jpg", "/api/v1/media/20") == ( "https://oss.example.test/melove/users/2/avatar.jpg?signed=1" ) assert calls == [("GET", "melove/users/2/avatar.jpg", 3600)] def test_register_login_and_profile_without_otp(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) email = f"tester-{uuid4().hex[:8]}@example.com" with TestClient(app) as client: response = client.post("/api/v1/auth/register", json={ "email_or_phone": email, "password": "password123", "confirm_password": "password123", "relationship_goal": "serious_relationship", "full_name": "测试用户", }) assert response.status_code == 201 registration_response = response.json() assert set(registration_response) == {"access_token", "refresh_token", "token_type"} assert registration_response["refresh_token"] token = registration_response["access_token"] login = client.post("/api/v1/auth/login", json={"email_or_phone": email, "password": "password123"}) assert login.status_code == 200 assert login.json()["user"]["email_or_phone"] == email updated = client.patch( "/api/v1/users/me", headers={"Authorization": f"Bearer {token}"}, json={"bio": "喜欢徒步", "relationship_goal": "casual_dates"}, ) assert updated.status_code == 200 assert updated.json()["profile"]["bio"] == "喜欢徒步" assert updated.json()["profile"]["relationship_goal"] == "casual_dates" assert client.patch( "/api/v1/users/me", headers={"Authorization": f"Bearer {token}"}, json={"relationship_goal": None}, ).status_code == 422 def test_update_profile_details_and_return_in_get_me(): with TestClient(app) as client: _, token, _, _ = register_user(client, full_name="Detail User", city="Shanghai") headers = {"Authorization": f"Bearer {token}"} details = { "occupation": "Creative Director at Studio Drift", "education": "NYU Gallatin School", "height_cm": 175, "beliefs": "spiritual_agnostic", "smoking": "non_smoker", "drinking": "socially", "pets": "golden_retriever", "exercise": "active_4x_week", } updated = client.patch("/api/v1/users/me", headers=headers, json=details) assert updated.status_code == 200 profile = updated.json()["profile"] for key, value in details.items(): assert profile[key] == value me = client.get("/api/v1/auth/me", headers=headers) assert me.status_code == 200 me_profile = me.json()["profile"] for key, value in details.items(): assert me_profile[key] == value assert me_profile["age"] is None assert client.patch( "/api/v1/users/me", headers=headers, json={"birth_date": "2000-01-01"}, ).status_code == 200 me_profile = client.get("/api/v1/auth/me", headers=headers).json()["profile"] assert me_profile["age"] == 26 assert client.patch( "/api/v1/users/me", headers=headers, json={"height_cm": 300}, ).status_code == 422 def test_registration_supports_bio_or_followup_video(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: _, bio_token, _, _ = register_user(client, bio=" I prefer a written introduction. ") bio_headers = {"Authorization": f"Bearer {bio_token}"} bio_profile = client.get("/api/v1/auth/me", headers=bio_headers).json()["profile"] assert bio_profile["bio"] == "I prefer a written introduction." assert bio_profile["video_url"] is None bio_to_video = client.post( "/api/v1/users/me/video", headers=bio_headers, files={"video": ("bio-replacement.mp4", b"bio-replacement-video", "video/mp4")}, ) assert bio_to_video.status_code == 200 replaced_bio_profile = client.get("/api/v1/auth/me", headers=bio_headers).json()["profile"] assert replaced_bio_profile["bio"] is None assert replaced_bio_profile["video_url"] == f"/api/v1/media/{bio_to_video.json()['id']}" _, video_token, _, _ = register_user(client) video_headers = {"Authorization": f"Bearer {video_token}"} before_upload = client.get("/api/v1/auth/me", headers=video_headers).json()["profile"] assert before_upload["bio"] is None assert before_upload["video_url"] is None uploaded = client.post( "/api/v1/users/me/video", headers=video_headers, files={"video": ("introduction.mp4", b"intro-video", "video/mp4")}, ) assert uploaded.status_code == 200 after_upload = client.get("/api/v1/auth/me", headers=video_headers).json()["profile"] assert after_upload["bio"] is None assert after_upload["video_url"] == f"/api/v1/media/{uploaded.json()['id']}" video_to_bio = client.patch( "/api/v1/users/me", headers=video_headers, json={"bio": "I switched to a written introduction."}, ) assert video_to_bio.status_code == 200 assert video_to_bio.json()["profile"]["bio"] == "I switched to a written introduction." assert video_to_bio.json()["profile"]["video_url"] is None assert client.get(f"/api/v1/media/{uploaded.json()['id']}", headers=video_headers).status_code == 404 blank_bio = client.post("/api/v1/auth/register", json={ "email_or_phone": f"blank-bio-{uuid4().hex[:8]}@example.com", "password": "password123", "confirm_password": "password123", "relationship_goal": "serious_relationship", "bio": " ", }) assert blank_bio.status_code == 422 def test_profile_image_and_video_uploads_replace_existing(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: _, token, _, _ = register_user(client) headers = {"Authorization": f"Bearer {token}"} first_image = client.post( "/api/v1/users/me/image", headers=headers, files={"image": ("first.png", b"first-image", "image/png")}, ) second_image = client.post( "/api/v1/users/me/image", headers=headers, files={"image": ("second.webp", b"second-image", "image/webp")}, ) assert first_image.status_code == second_image.status_code == 200 assert first_image.json()["id"] != second_image.json()["id"] assert client.get(f"/api/v1/media/{first_image.json()['id']}", headers=headers).status_code == 404 current_image = client.get(f"/api/v1/media/{second_image.json()['id']}", headers=headers) assert current_image.status_code == 200 assert current_image.content == b"second-image" first_video = client.post( "/api/v1/users/me/video", headers=headers, files={"video": ("first.mp4", b"first-video", "video/mp4")}, ) second_video = client.post( "/api/v1/users/me/video", headers=headers, files={"video": ("second.webm", b"second-video", "video/webm")}, ) assert first_video.status_code == second_video.status_code == 200 assert first_video.json()["id"] != second_video.json()["id"] assert client.get(f"/api/v1/media/{first_video.json()['id']}", headers=headers).status_code == 404 current_video = client.get(f"/api/v1/media/{second_video.json()['id']}", headers=headers) assert current_video.status_code == 200 assert current_video.content == b"second-video" profile = client.get("/api/v1/auth/me", headers=headers).json()["profile"] assert profile["profile_image_url"] == f"/api/v1/media/{second_image.json()['id']}" assert profile["video_url"] == f"/api/v1/media/{second_video.json()['id']}" stored_files = [path for path in tmp_path.rglob("*") if path.is_file()] assert sorted(path.read_bytes() for path in stored_files) == [b"second-image", b"second-video"] def test_profile_media_replacement_keeps_existing_file_when_storage_fails(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: _, token, _, _ = register_user(client) headers = {"Authorization": f"Bearer {token}"} existing = client.post( "/api/v1/users/me/image", headers=headers, files={"image": ("existing.png", b"existing-image", "image/png")}, ) assert existing.status_code == 200 def fail_storage(_local_path, _relative_key): raise RuntimeError("storage unavailable") monkeypatch.setattr("app.main.storage.finalize", fail_storage) replacement = client.post( "/api/v1/users/me/image", headers=headers, files={"image": ("replacement.png", b"replacement-image", "image/png")}, ) assert replacement.status_code == 503 current = client.get(f"/api/v1/media/{existing.json()['id']}", headers=headers) assert current.status_code == 200 assert current.content == b"existing-image" stored_files = [path for path in tmp_path.rglob("*") if path.is_file()] assert len(stored_files) == 1 assert stored_files[0].read_bytes() == b"existing-image" def test_duplicate_registration(): payload = { "email_or_phone": f"138{uuid4().int % 10**8:08d}", "password": "password123", "confirm_password": "password123", "relationship_goal": "serious_relationship", } with TestClient(app) as client: assert client.post("/api/v1/auth/register", json=payload).status_code == 201 assert client.post("/api/v1/auth/register", json=payload).status_code == 409 def test_email_is_case_insensitive_and_new_user_is_active(): email = f"ACTIVE-{uuid4().hex[:10]}@EXAMPLE.COM" payload = { "email_or_phone": email, "password": "password123", "confirm_password": "password123", "relationship_goal": "serious_relationship", } with TestClient(app) as client: registered = client.post("/api/v1/auth/register", json=payload) assert registered.status_code == 201 response = client.post("/api/v1/auth/login", json={ "email_or_phone": email.lower(), "password": payload["password"], }) assert response.status_code == 200 active_headers = {"Authorization": f"Bearer {registered.json()['access_token']}"} assert client.get("/api/v1/meet", headers=active_headers).status_code == 200 assert client.post("/api/v1/auth/register", json={ "email_or_phone": email.lower(), "password": "password123", "confirm_password": "password123", "relationship_goal": "serious_relationship", }).status_code == 409 def test_phone_login_and_invalid_password(): phone = f"139{uuid4().int % 10**8:08d}" with TestClient(app) as client: assert client.post("/api/v1/auth/register", json={ "email_or_phone": phone, "password": "password123", "confirm_password": "password123", "relationship_goal": "serious_relationship", }).status_code == 201 assert client.post("/api/v1/auth/login", json={ "email_or_phone": phone, "password": "password123", }).status_code == 200 assert client.post("/api/v1/auth/login", json={ "email_or_phone": phone, "password": "wrong-password", }).status_code == 401 def test_auth_validation_and_otp_endpoints_are_removed(): with TestClient(app) as client: assert client.post("/api/v1/auth/register", json={"password": "password123"}).status_code == 422 assert client.post("/api/v1/auth/register", json={ "email_or_phone": "missing-confirm@example.com", "password": "password123", }).status_code == 422 assert client.post("/api/v1/auth/register", json={ "email_or_phone": "mismatch-confirm@example.com", "password": "password123", "confirm_password": "password124", "relationship_goal": "serious_relationship", }).status_code == 422 assert client.post("/api/v1/auth/register", json={ "email_or_phone": "missing-goal@example.com", "password": "password123", "confirm_password": "password123", }).status_code == 422 assert client.post("/api/v1/auth/register", json={ "email_or_phone": "invalid-goal@example.com", "password": "password123", "confirm_password": "password123", "relationship_goal": "long_term", }).status_code == 422 assert client.post("/api/v1/auth/register", json={ "email_or_phone": " ", "password": "password123", }).status_code == 422 assert client.post("/api/v1/auth/login", json={ "email_or_phone": "user@example.com", }).status_code == 422 assert client.post("/api/v1/auth/request-code", json={}).status_code == 404 assert client.post("/api/v1/auth/verify", json={}).status_code == 404 def test_interests_support_registration_custom_values_and_unlimited_count(): email = f"interests-{uuid4().hex[:10]}@example.com" interests = ["Photography", "Dogs", "Travel", "Music", "Gaming", "Fitness", "Cooking", "我的自定义兴趣"] with TestClient(app) as client: registered = client.post("/api/v1/auth/register", json={ "email_or_phone": email, "password": "password123", "confirm_password": "password123", "relationship_goal": "new_friends", "gender": "Man", "interests": interests, }) assert registered.status_code == 201 headers = {"Authorization": f"Bearer {registered.json()['access_token']}"} me = client.get("/api/v1/users/me", headers=headers) assert me.status_code == 200 assert me.json()["profile"]["relationship_goal"] == "new_friends" assert me.json()["profile"]["interests"] == interests updated_interests = interests + ["Board Games", "Hiking", "Coffee", "Art", "Reading"] updated = client.put("/api/v1/users/me/interests", headers=headers, json={"interests": updated_interests}) assert updated.status_code == 200 assert updated.json()["interests"] == updated_interests listed = client.get("/api/v1/users/me/interests", headers=headers) assert listed.json()["interests"] == updated_interests def test_interests_require_two_non_empty_unique_values(): email = f"interest-validation-{uuid4().hex[:10]}@example.com" with TestClient(app) as client: registered = client.post("/api/v1/auth/register", json={ "email_or_phone": email, "password": "password123", "confirm_password": "password123", "relationship_goal": "casual_dates", }) headers = {"Authorization": f"Bearer {registered.json()['access_token']}"} assert client.put("/api/v1/users/me/interests", headers=headers, json={"interests": ["Only one"]}).status_code == 422 assert client.put("/api/v1/users/me/interests", headers=headers, json={"interests": ["Travel", " "]}).status_code == 422 duplicate = client.put("/api/v1/users/me/interests", headers=headers, json={"interests": ["Travel", "travel", "Music"]}) assert duplicate.status_code == 200 assert duplicate.json()["interests"] == ["Travel", "Music"] def test_refresh_me_and_logout_session(): email = f"session-{uuid4().hex[:10]}@example.com" with TestClient(app) as client: registered = client.post( "/api/v1/auth/register", json={ "email_or_phone": email, "password": "password123", "confirm_password": "password123", "relationship_goal": "marriage", "full_name": "Session User", }, ) assert registered.status_code == 201 initial = registered.json() assert initial["refresh_token"] refreshed = client.post( "/api/v1/auth/refresh", json={"refresh_token": initial["refresh_token"]}, ) assert refreshed.status_code == 200 rotated = refreshed.json() assert rotated["refresh_token"] != initial["refresh_token"] assert client.post( "/api/v1/auth/refresh", json={"refresh_token": initial["refresh_token"]}, ).status_code == 401 me_headers = {"Authorization": f"Bearer {rotated['access_token']}"} me = client.get("/api/v1/auth/me", headers=me_headers) assert me.status_code == 200 assert me.json()["email_or_phone"] == email assert client.get("/api/v1/users/me", headers=me_headers).status_code == 200 assert client.post( "/api/v1/auth/logout", json={"refresh_token": rotated["refresh_token"]}, ).status_code == 200 assert client.post( "/api/v1/auth/refresh", json={"refresh_token": rotated["refresh_token"]}, ).status_code == 401 def test_meet_match_cancel_and_search(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: first_id, first_token, _, _ = register_user(client, full_name="Alice", city="上海") second_id, second_token, _, _ = register_user( client, full_name="Bob Zhang", bio="喜欢徒步和安静的周末", city="上海", ) first_headers = {"Authorization": f"Bearer {first_token}"} second_headers = {"Authorization": f"Bearer {second_token}"} assert client.put( "/api/v1/users/me/location", headers=first_headers, json={"latitude": 31.2304, "longitude": 121.4737, "city": "上海", "county": "黄浦区"}, ).status_code == 200 assert client.put( "/api/v1/users/me/location", headers=second_headers, json={"latitude": 31.2310, "longitude": 121.4740, "city": "上海", "county": "静安区"}, ).status_code == 200 upload = client.post( "/api/v1/users/me/image", headers=second_headers, files={"image": ("avatar.png", b"fake-image", "image/png")}, ) assert upload.status_code == 200 image_id = upload.json()["id"] identity = client.post( "/api/v1/users/me/identity", headers=second_headers, files={"document": ("passport.png", b"private", "image/png")}, ) assert identity.status_code == 200 identity_id = identity.json()["id"] assert client.get(f"/api/v1/media/{image_id}", headers=first_headers).status_code == 200 assert client.get(f"/api/v1/media/{identity_id}", headers=first_headers).status_code == 403 monkeypatch.setattr( "app.main.storage.access_url", lambda _storage_path, fallback_url: f"https://media.example.test{fallback_url}?signed=1", ) search = client.get( "/api/v1/search", headers=first_headers, params={"q": "Bob 徒步", "max_distance_km": 5}, ) assert search.status_code == 200 bob = next(item for item in search.json() if item["id"] == second_id) assert bob["distance_km"] < 1 assert [item["id"] for item in bob["media"]] == [image_id] expected_image_url = f"https://media.example.test/api/v1/media/{image_id}?signed=1" assert bob["profile_image_url"] == expected_image_url assert bob["media"][0]["url"] == expected_image_url candidates = client.get("/api/v1/meet", headers=first_headers) assert any(item["id"] == second_id for item in candidates.json()) first_swipe = client.post( "/api/v1/meet/swipes", headers=first_headers, json={"target_user_id": second_id, "direction": "like"}, ) assert first_swipe.status_code == 200 assert first_swipe.json()["matched"] is False second_swipe = client.post( "/api/v1/meet/swipes", headers=second_headers, json={"target_user_id": first_id, "direction": "like"}, ) assert second_swipe.status_code == 200 assert second_swipe.json()["matched"] is True assert len(client.get("/api/v1/matches", headers=first_headers).json()) == 1 cancel = client.delete(f"/api/v1/meet/likes/{second_id}", headers=first_headers) assert cancel.status_code == 200 assert client.get("/api/v1/matches", headers=first_headers).json() == [] def test_profile_activity_posts_like_lists_matches_and_swipe_undo(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: alice_id, alice_token, _, _ = register_user(client, full_name="Activity Alice", city="Shanghai") bob_id, bob_token, _, _ = register_user(client, full_name="Activity Bob", city="Shanghai") carol_id, _, _, _ = register_user(client, full_name="Activity Carol", city="Beijing") alice_headers = {"Authorization": f"Bearer {alice_token}"} bob_headers = {"Authorization": f"Bearer {bob_token}"} text_post = client.post( "/api/v1/explore/posts", headers=alice_headers, json={"content": "My text post", "post_type": "text"}, ) assert text_post.status_code == 201 image_post = client.post( "/api/v1/explore/posts", headers=alice_headers, json={"content": "My image post", "post_type": "image"}, ) assert image_post.status_code == 201 uploaded = client.post( f"/api/v1/explore/posts/{image_post.json()['id']}/media", headers=alice_headers, files={"media": ("post.png", b"post-image", "image/png")}, ) assert uploaded.status_code == 201 monkeypatch.setattr( "app.main.storage.access_url", lambda _storage_path, fallback_url: f"https://media.example.test{fallback_url}?signed=1", ) assert client.post( "/api/v1/meet/swipes", headers=bob_headers, json={"target_user_id": alice_id, "direction": "like"}, ).status_code == 200 matched = client.post( "/api/v1/meet/swipes", headers=alice_headers, json={"target_user_id": bob_id, "direction": "like"}, ) assert matched.status_code == 200 assert matched.json()["matched"] is True assert client.post( "/api/v1/meet/swipes", headers=alice_headers, json={"target_user_id": carol_id, "direction": "like"}, ).status_code == 200 me = client.get("/api/v1/auth/me", headers=alice_headers) assert me.status_code == 200 me_body = me.json() assert [post["id"] for post in me_body["posts"]] == [image_post.json()["id"], text_post.json()["id"]] assert me_body["posts"][0]["media"][0]["url"] == ( f"https://media.example.test/api/v1/explore/media/{uploaded.json()['id']}?signed=1" ) assert [item["id"] for item in me_body["liked_users"]] == [carol_id] assert me_body["liked_by_users"] == [] assert [item["user"]["id"] for item in me_body["matches"]] == [bob_id] assert me_body["liked_users_count"] == 1 assert me_body["liked_by_users_count"] == 0 assert me_body["matches_count"] == 1 my_posts = client.get("/api/v1/users/me/posts?limit=1", headers=alice_headers) assert my_posts.status_code == 200 assert [item["id"] for item in my_posts.json()] == [image_post.json()["id"]] public_posts = client.get(f"/api/v1/users/{alice_id}/posts", headers=bob_headers) assert public_posts.status_code == 200 assert [item["id"] for item in public_posts.json()] == [image_post.json()["id"], text_post.json()["id"]] public_profile = client.get(f"/api/v1/users/{alice_id}", headers=bob_headers) assert public_profile.status_code == 200 assert public_profile.json()["full_name"] == "Activity Alice" assert public_profile.json()["relationship_goal"] == "serious_relationship" assert [item["id"] for item in client.get( "/api/v1/users/me/liked", headers=alice_headers ).json()] == [carol_id] assert client.get("/api/v1/users/me/likes", headers=alice_headers).json() == [] assert [item["user"]["id"] for item in client.get( "/api/v1/users/me/matches", headers=alice_headers ).json()] == [bob_id] assert [item["user"]["id"] for item in client.get( "/api/v1/users/me/matches?limit=1&offset=20", headers=alice_headers ).json()] == [bob_id] assert [item["user"]["id"] for item in client.get( "/api/v1/matches?limit=1&offset=20", headers=alice_headers ).json()] == [bob_id] undo_like = client.delete(f"/api/v1/meet/swipes/{carol_id}", headers=alice_headers) assert undo_like.status_code == 200 assert undo_like.json()["previous_direction"] == "like" assert carol_id in [item["id"] for item in client.get( "/api/v1/meet?limit=50", headers=alice_headers ).json()] passed = client.post( "/api/v1/meet/swipes", headers=alice_headers, json={"target_user_id": carol_id, "direction": "pass"}, ) assert passed.status_code == 200 assert carol_id not in [item["id"] for item in client.get( "/api/v1/meet?limit=50", headers=alice_headers ).json()] undo_pass = client.delete(f"/api/v1/meet/swipes/{carol_id}", headers=alice_headers) assert undo_pass.status_code == 200 assert undo_pass.json()["previous_direction"] == "pass" undo_match_like = client.delete(f"/api/v1/meet/swipes/{bob_id}", headers=alice_headers) assert undo_match_like.status_code == 200 assert client.get("/api/v1/matches", headers=alice_headers).json() == [] after_undo = client.get("/api/v1/auth/me", headers=alice_headers).json() assert after_undo["liked_users_count"] == 0 assert after_undo["liked_by_users_count"] == 1 assert after_undo["matches_count"] == 0 assert [item["id"] for item in client.get( "/api/v1/users/me/likes", headers=alice_headers ).json()] == [bob_id] def test_profile_activity_counts_exceed_list_limit_and_exclude_suspended_users(): with TestClient(app) as client: owner_id, token, _, _ = register_user(client) headers = {"Authorization": f"Bearer {token}"} with SessionLocal() as db: targets = [User(status=UserStatus.active) for _ in range(22)] targets[-1].status = UserStatus.suspended db.add_all(targets) db.flush() db.add_all([ Swipe(actor_id=owner_id, target_id=target.id, direction="like") for target in targets ]) db.add_all([ Swipe(actor_id=target.id, target_id=owner_id, direction="super_like") for target in targets[:-1] ]) db.add(Match(user1_id=min(owner_id, targets[0].id), user2_id=max(owner_id, targets[0].id))) db.commit() for path in ("/api/v1/auth/me", "/api/v1/users/me"): response = client.get(path, headers=headers) assert response.status_code == 200 body = response.json() assert len(body["liked_users"]) == 20 assert body["liked_users_count"] == 20 assert body["liked_by_users_count"] == 20 assert body["matches_count"] == 1 for path in ("/api/v1/users/me/liked", "/api/v1/users/me/likes"): response = client.get(path, headers=headers) assert response.status_code == 200 assert len(response.json()) == 20 legacy_params = client.get(f"{path}?limit=1&offset=20", headers=headers) assert legacy_params.status_code == 200 assert [item["id"] for item in legacy_params.json()] == [ item["id"] for item in response.json() ] def test_location_updates_city_county_and_discovery_order(): suffix = uuid4().hex[:8] old_city = f"Old-{suffix}" new_city = f"New-{suffix}" with TestClient(app) as client: viewer_id, viewer_token, _, _ = register_user(client, city=old_city, country="China") old_id, _, _, _ = register_user(client, city=old_city, country="China") new_id, _, _, _ = register_user(client, city=new_city, country="China") headers = {"Authorization": f"Bearer {viewer_token}"} assert client.get("/api/v1/meet?limit=1", headers=headers).json()[0]["id"] == old_id assert client.put("/api/v1/users/me/location", headers=headers, json={ "latitude": 31.2304, "longitude": 121.4737, "county": "Huangpu", }).status_code == 422 assert client.put("/api/v1/users/me/location", headers=headers, json={ "latitude": 31.2304, "longitude": 121.4737, "city": new_city, "county": " ", }).status_code == 422 updated = client.put("/api/v1/users/me/location", headers=headers, json={ "latitude": 31.2304, "longitude": 121.4737, "city": f" {new_city} ", "county": " Huangpu ", }) assert updated.status_code == 200 assert updated.json()["city"] == new_city assert updated.json()["county"] == "Huangpu" assert updated.json()["latitude"] == 31.2304 me = client.get("/api/v1/auth/me", headers=headers).json() assert me["profile"]["city"] == new_city assert me["profile"]["county"] == "Huangpu" assert me["profile"]["country"] == "China" assert client.get(f"/api/v1/users/{viewer_id}", headers=headers).json()["county"] == "Huangpu" assert client.get("/api/v1/meet?limit=1", headers=headers).json()[0]["id"] == new_id changed_city = client.patch("/api/v1/users/me", headers=headers, json={"city": old_city}) assert changed_city.status_code == 200 assert changed_city.json()["profile"]["county"] is None def test_super_like_priority_matching_and_cancel(): city = f"Super-like-city-{uuid4().hex[:8]}" with TestClient(app) as client: viewer_id, viewer_token, _, _ = register_user(client, city=city, country="China") local_id, _, _, _ = register_user(client, city=city, country="China") distant_id, distant_token, _, _ = register_user(client, city="Remote", country="Other") viewer_headers = {"Authorization": f"Bearer {viewer_token}"} distant_headers = {"Authorization": f"Bearer {distant_token}"} viewer_meet = lambda: client.get("/api/v1/meet?limit=1", headers=viewer_headers).json()[0]["id"] assert viewer_meet() == local_id assert client.post("/api/v1/meet/swipes", headers=distant_headers, json={ "target_user_id": viewer_id, "direction": "super_liked", }).status_code == 422 super_like = client.post("/api/v1/meet/swipes", headers=distant_headers, json={ "target_user_id": viewer_id, "direction": "super_like", }) assert super_like.status_code == 200 assert super_like.json() == {"direction": "super_like", "matched": False, "match_id": None} assert viewer_meet() == distant_id assert [item["id"] for item in client.get( "/api/v1/users/me/likes", headers=viewer_headers ).json()] == [distant_id] assert [item["id"] for item in client.get( "/api/v1/users/me/liked", headers=distant_headers ).json()] == [viewer_id] matched = client.post("/api/v1/meet/swipes", headers=viewer_headers, json={ "target_user_id": distant_id, "direction": "like", }) assert matched.status_code == 200 assert matched.json()["matched"] is True assert len(client.get("/api/v1/matches", headers=viewer_headers).json()) == 1 canceled = client.delete(f"/api/v1/meet/swipes/{viewer_id}", headers=distant_headers) assert canceled.status_code == 200 assert canceled.json()["previous_direction"] == "super_like" assert client.get("/api/v1/matches", headers=viewer_headers).json() == [] assert client.delete(f"/api/v1/meet/swipes/{distant_id}", headers=viewer_headers).status_code == 200 assert viewer_meet() == local_id assert client.post("/api/v1/meet/swipes", headers=distant_headers, json={ "target_user_id": viewer_id, "direction": "super_like", }).status_code == 200 assert viewer_meet() == distant_id assert client.post("/api/v1/meet/swipes", headers=distant_headers, json={ "target_user_id": viewer_id, "direction": "pass", }).status_code == 200 assert viewer_meet() == local_id assert client.delete(f"/api/v1/meet/swipes/{viewer_id}", headers=distant_headers).json()[ "previous_direction" ] == "pass" assert client.post("/api/v1/meet/swipes", headers=distant_headers, json={ "target_user_id": viewer_id, "direction": "super_like", }).status_code == 200 assert viewer_meet() == distant_id assert client.post("/api/v1/meet/swipes", headers=distant_headers, json={ "target_user_id": viewer_id, "direction": "like", }).status_code == 200 assert viewer_meet() == local_id assert client.post("/api/v1/meet/swipes", headers=distant_headers, json={ "target_user_id": viewer_id, "direction": "super_like", }).status_code == 200 assert client.delete(f"/api/v1/meet/likes/{viewer_id}", headers=distant_headers).status_code == 200 assert viewer_meet() == local_id def test_discovery_country_compatibility_search_and_explore_order(): suffix = uuid4().hex[:8] country = f"Sortland-{suffix}" home_city = f"Home-{suffix}" other_city = f"Other-{suffix}" foreign_country = f"Elsewhere-{suffix}" with TestClient(app) as client: _, viewer_token, _, _ = register_user( client, full_name="Viewer", city=home_city, country=country, interests=["Travel", "Music"], ) high_id, high_token, _, _ = register_user( client, full_name="Same City High", city=home_city, country=country, interests=["Travel", "Music"], ) low_id, low_token, _, _ = register_user( client, full_name="Same City Low", city=home_city, country=country, interests=["Cooking", "Art"], ) country_id, country_token, _, _ = register_user( client, full_name="Same Country", city=other_city, country=country, interests=["Travel", "Music"], ) foreign_id, foreign_token, _, _ = register_user( client, full_name="Foreign", city=other_city, country=foreign_country, interests=["Travel", "Music"], ) viewer_headers = {"Authorization": f"Bearer {viewer_token}"} meet = client.get("/api/v1/meet", headers=viewer_headers, params={"limit": 50}) assert meet.status_code == 200 meet_items = {item["id"]: item for item in meet.json()} assert meet_items[high_id]["country"] == country assert meet_items[high_id]["compatibility_percentage"] == 100 assert meet_items[low_id]["compatibility_percentage"] == 0 ordered_ids = [item["id"] for item in meet.json()] relevant_ids = [item for item in ordered_ids if item in {high_id, low_id, country_id, foreign_id}] assert relevant_ids == [high_id, low_id, country_id, foreign_id] city_and_country = client.get( "/api/v1/search", headers=viewer_headers, params={"city": f" {home_city.lower()} ", "country": country.lower(), "limit": 50}, ) assert city_and_country.status_code == 200 assert [item["id"] for item in city_and_country.json()] == [high_id, low_id] country_only = client.get( "/api/v1/search", headers=viewer_headers, params={"country": country, "limit": 50}, ) assert country_only.status_code == 200 assert [item["id"] for item in country_only.json()] == [high_id, low_id, country_id] post_ids = {} for name, token in ( ("high", high_token), ("low", low_token), ("country", country_token), ("foreign", foreign_token), ): response = client.post( "/api/v1/explore/posts", headers={"Authorization": f"Bearer {token}"}, json={"content": f"Ordering post {name}", "post_type": "text"}, ) assert response.status_code == 201 post_ids[name] = response.json()["id"] feed = client.get("/api/v1/explore/feed", headers=viewer_headers, params={"limit": 50}) assert feed.status_code == 200 relevant_posts = [ item["id"] for item in feed.json() if item["id"] in set(post_ids.values()) ] assert relevant_posts == [ post_ids["high"], post_ids["low"], post_ids["country"], post_ids["foreign"], ] high_post = next(item for item in feed.json() if item["id"] == post_ids["high"]) assert high_post["author"]["compatibility_percentage"] == 100 assert high_post["author"]["country"] == country assert client.get( "/api/v1/explore/feed", headers=viewer_headers, params={"before_id": high_post["id"], "offset": 1}, ).status_code == 422 def test_explore_post_like_comment_and_permissions(): with TestClient(app) as client: author_id, author_token, _, _ = register_user(client, full_name="Post Author", city="上海") reader_id, reader_token, _, _ = register_user(client, full_name="Post Reader", city="北京") author_headers = {"Authorization": f"Bearer {author_token}"} reader_headers = {"Authorization": f"Bearer {reader_token}"} created = client.post( "/api/v1/explore/posts", headers=author_headers, json={"content": "今天去公园散步，天气很好", "post_type": "text"}, ) assert created.status_code == 201 post = created.json() assert post["author"]["id"] == author_id assert post["like_count"] == 0 assert post["comment_count"] == 0 feed = client.get("/api/v1/explore/feed", headers=reader_headers) assert feed.status_code == 200 assert any(item["id"] == post["id"] for item in feed.json()) liked = client.post(f"/api/v1/explore/posts/{post['id']}/like", headers=reader_headers) assert liked.status_code == 200 assert liked.json()["like_count"] == 1 assert liked.json()["liked_by_me"] is True repeated = client.post(f"/api/v1/explore/posts/{post['id']}/like", headers=reader_headers) assert repeated.status_code == 200 assert repeated.json()["like_count"] == 1 comment = client.post( f"/api/v1/explore/posts/{post['id']}/comments", headers=reader_headers, json={"content": "看起来很不错"}, ) assert comment.status_code == 201 assert comment.json()["author"]["id"] == reader_id comments = client.get(f"/api/v1/explore/posts/{post['id']}/comments", headers=author_headers) assert comments.status_code == 200 assert comments.json()[0]["content"] == "看起来很不错" forbidden = client.delete(f"/api/v1/explore/posts/{post['id']}", headers=reader_headers) assert forbidden.status_code == 403 assert client.delete(f"/api/v1/explore/posts/{post['id']}", headers=author_headers).status_code == 200 assert client.get(f"/api/v1/explore/posts/{post['id']}", headers=reader_headers).status_code == 404 def test_post_and_comment_author_avatars(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) monkeypatch.setattr( "app.main.storage.access_url", lambda _storage_path, fallback_url: f"https://media.example.test{fallback_url}?signed=1", ) with TestClient(app) as client: author_id, author_token, _, _ = register_user(client, full_name="Avatar Author") _, reader_token, _, _ = register_user(client, full_name="Avatar Reader") author_headers = {"Authorization": f"Bearer {author_token}"} reader_headers = {"Authorization": f"Bearer {reader_token}"} expected_urls = [] for headers in (author_headers, reader_headers): uploaded = client.post( "/api/v1/users/me/image", headers=headers, files={"image": ("avatar.png", b"avatar", "image/png")}, ) assert uploaded.status_code == 200 expected_urls.append(f"https://media.example.test/api/v1/media/{uploaded.json()['id']}?signed=1") post = client.post( "/api/v1/explore/posts", headers=author_headers, json={"content": "Avatar test"}, ).json() post_id = post["id"] assert post["author"]["profile_image_url"] == expected_urls[0] for path in ( f"/api/v1/explore/posts/{post_id}", "/api/v1/explore/feed?limit=50", f"/api/v1/users/{author_id}/posts", f"/api/v1/users/{author_id}", "/api/v1/auth/me", ): headers = author_headers if path == "/api/v1/auth/me" else reader_headers response = client.get(path, headers=headers) assert response.status_code == 200 result = response.json() if isinstance(result, dict) and "posts" in result: result = result["posts"] item = next(value for value in result if value["id"] == post_id) if isinstance(result, list) else result assert item["author"]["profile_image_url"] == expected_urls[0] comment = client.post( f"/api/v1/explore/posts/{post_id}/comments", headers=reader_headers, json={"content": "Hello"}, ) assert comment.status_code == 201 assert comment.json()["author"]["profile_image_url"] == expected_urls[1] comments = client.get(f"/api/v1/explore/posts/{post_id}/comments", headers=author_headers) assert comments.json()[0]["author"]["profile_image_url"] == expected_urls[1] def test_compact_post_likes_and_liker_list(): with TestClient(app) as client: _, author_token, _, _ = register_user(client, full_name="Liker Author") first_id, first_token, _, _ = register_user(client, full_name="First Liker") second_id, second_token, _, _ = register_user(client) author_headers = {"Authorization": f"Bearer {author_token}"} first_headers = {"Authorization": f"Bearer {first_token}"} second_headers = {"Authorization": f"Bearer {second_token}"} post = client.post( "/api/v1/explore/posts", headers=author_headers, json={"content": "Likers test"}, ).json() post_id = post["id"] assert post["author"]["profile_image_url"] is None assert post["liked_by_users"] == [] path = f"/api/v1/explore/posts/{post_id}/like" assert client.get(path).status_code == 401 assert client.get("/api/v1/explore/posts/2147483647/like", headers=author_headers).status_code == 404 assert client.get(path, headers=author_headers).json() == [] first_liker = {"id": first_id, "full_name": "First Liker"} second_liker = {"id": second_id, "full_name": None} liked = client.post(path, headers=first_headers) assert liked.status_code == 200 assert liked.json() == { "message": "like successful", "like_count": 1, "liked_by_me": True, "liked_by_users": [first_liker], } assert client.post(path, headers=first_headers).json() == liked.json() second_like = client.post(path, headers=second_headers).json() assert second_like["like_count"] == 2 assert second_like["liked_by_users"] == [second_liker, first_liker] assert client.get(path, headers=author_headers).json() == [second_liker, first_liker] detail = client.get(f"/api/v1/explore/posts/{post_id}", headers=first_headers).json() assert detail["liked_by_users"] == [second_liker, first_liker] assert detail["like_count"] == 2 assert detail["liked_by_me"] is True unliked = client.delete(path, headers=first_headers).json() assert unliked["message"] == "unlike successful" assert "content" not in unliked assert "media" not in unliked assert unliked["liked_by_users"] == [second_liker] assert unliked["like_count"] == 1 assert unliked["liked_by_me"] is False assert client.delete(path, headers=first_headers).json()["message"] == "unlike successful" client.post(path, headers=first_headers) with SessionLocal() as db: db.get(User, second_id).status = UserStatus.suspended db.commit() assert client.get(path, headers=author_headers).json() == [first_liker] detail = client.get(f"/api/v1/explore/posts/{post_id}", headers=author_headers).json() assert detail["liked_by_users"] == [first_liker] assert detail["like_count"] == 1 assert detail["liked_by_me"] is False feed = client.get("/api/v1/explore/feed?limit=50", headers=author_headers).json() assert next(item for item in feed if item["id"] == post_id)["liked_by_users"] == [first_liker] def test_comment_replies_and_likes(): with TestClient(app) as client: _, author_token, _, _ = register_user(client, full_name="Comment Author") _, reader_token, _, _ = register_user(client, full_name="Comment Reader") author_headers = {"Authorization": f"Bearer {author_token}"} reader_headers = {"Authorization": f"Bearer {reader_token}"} post = client.post( "/api/v1/explore/posts", headers=author_headers, json={"content": "Comment test"}, ).json() post_id = post["id"] parent = client.post( f"/api/v1/explore/posts/{post_id}/comments", headers=reader_headers, json={"content": "Parent comment"}, ) assert parent.status_code == 201 parent_id = parent.json()["id"] assert parent.json()["parent_comment_id"] is None assert parent.json()["reply_count"] == 0 reply = client.post( f"/api/v1/explore/posts/{post_id}/comments", headers=author_headers, json={"content": "Reply comment", "parent_comment_id": parent_id}, ) assert reply.status_code == 201 assert reply.json()["parent_comment_id"] == parent_id assert client.post( f"/api/v1/explore/posts/{post_id}/comments", headers=author_headers, json={"content": "Bad reply", "parent_comment_id": 2147483647}, ).status_code == 404 comments = client.get( f"/api/v1/explore/posts/{post_id}/comments", headers=reader_headers, ).json() parent_view = next(item for item in comments if item["id"] == parent_id) assert parent_view["reply_count"] == 1 assert parent_view["like_count"] == 0 like_path = f"/api/v1/explore/comments/{parent_id}/like" liked = client.post(like_path, headers=author_headers) assert liked.status_code == 200 assert liked.json() == {"message": "like successful", "like_count": 1, "liked_by_me": True} assert client.post(like_path, headers=author_headers).json()["like_count"] == 1 parent_view = next(item for item in client.get( f"/api/v1/explore/posts/{post_id}/comments", headers=reader_headers, ).json() if item["id"] == parent_id) assert parent_view["like_count"] == 1 assert parent_view["liked_by_me"] is False unliked = client.delete(like_path, headers=author_headers) assert unliked.status_code == 200 assert unliked.json() == {"message": "unlike successful", "like_count": 0, "liked_by_me": False} assert client.delete(like_path, headers=author_headers).json()["like_count"] == 0 def test_edit_post_and_comment(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: author_id, author_token, _, _ = register_user(client, full_name="Editor") _, reader_token, _, _ = register_user(client, full_name="Reader") author_headers = {"Authorization": f"Bearer {author_token}"} reader_headers = {"Authorization": f"Bearer {reader_token}"} post = client.post( "/api/v1/explore/posts", headers=author_headers, json={"content": "Old text", "post_type": "text"}, ).json() post_id = post["id"] assert client.patch( f"/api/v1/explore/posts/{post_id}", headers=reader_headers, json={"content": "Hacked"}, ).status_code == 403 updated = client.patch( f"/api/v1/explore/posts/{post_id}", headers=author_headers, json={"content": "New text", "feeling": "happy", "location": {"name": "Park"}}, ) assert updated.status_code == 200 body = updated.json() assert body["content"] == "New text" assert body["feeling"] == "happy" assert body["location"] == {"name": "Park", "latitude": None, "longitude": None} assert body["content_format"] == "plain" assert client.patch( f"/api/v1/explore/posts/{post_id}", headers=author_headers, json={"content": None}, ).status_code == 422 assert client.patch( f"/api/v1/explore/posts/{post_id}", headers=author_headers, json={"content_format": None}, ).status_code == 422 assert client.patch( f"/api/v1/explore/posts/{post_id}", headers=author_headers, json={"content": "", "feeling": None, "location": None}, ).status_code == 422 comment = client.post( f"/api/v1/explore/posts/{post_id}/comments", headers=reader_headers, json={"content": "Nice"}, ).json() comment_id = comment["id"] assert client.patch( f"/api/v1/explore/comments/{comment_id}", headers=author_headers, json={"content": "Stolen"}, ).status_code == 403 assert client.patch( "/api/v1/explore/comments/2147483647", headers=reader_headers, json={"content": "x"}, ).status_code == 404 edited = client.patch( f"/api/v1/explore/comments/{comment_id}", headers=reader_headers, json={"content": " Better "}, ) assert edited.status_code == 200 assert edited.json()["content"] == "Better" assert client.patch( f"/api/v1/explore/comments/{comment_id}", headers=reader_headers, json={"content": " "}, ).status_code == 422 def test_post_supports_image_and_video_together(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: _, token, _, _ = register_user(client, full_name="Mixed Media") headers = {"Authorization": f"Bearer {token}"} post = client.post( "/api/v1/explore/posts", headers=headers, json={"content": "Mixed post", "post_type": "text"}, ).json() post_id = post["id"] image = client.post( f"/api/v1/explore/posts/{post_id}/media", headers=headers, files={"media": ("a.png", b"img1", "image/png")}, ) assert image.status_code == 201 video = client.post( f"/api/v1/explore/posts/{post_id}/media", headers=headers, files={"media": ("b.mp4", b"vid1", "video/mp4")}, ) assert video.status_code == 201 detail = client.get(f"/api/v1/explore/posts/{post_id}", headers=headers).json() assert detail["post_type"] == "mixed" assert {item["media_type"] for item in detail["media"]} == {"image", "video"} second_video = client.post( f"/api/v1/explore/posts/{post_id}/media", headers=headers, files={"media": ("c.mp4", b"vid2", "video/mp4")}, ) assert second_video.status_code == 409 assert client.delete( f"/api/v1/explore/media/{video.json()['id']}", headers=headers, ).status_code == 200 detail = client.get(f"/api/v1/explore/posts/{post_id}", headers=headers).json() assert detail["post_type"] == "image" assert {item["media_type"] for item in detail["media"]} == {"image"} assert client.delete( f"/api/v1/explore/media/{image.json()['id']}", headers=headers, ).status_code == 200 detail = client.get(f"/api/v1/explore/posts/{post_id}", headers=headers).json() assert detail["post_type"] == "text" assert detail["media"] == [] def test_post_composer_metadata_in_feed_and_public_posts(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: author_id, author_token, _, _ = register_user(client, city="Shanghai") _, reader_token, _, _ = register_user(client, city="Shanghai") author_headers = {"Authorization": f"Bearer {author_token}"} reader_headers = {"Authorization": f"Bearer {reader_token}"} for payload in ( {"content": " "}, {"content": "Hello", "feeling": " "}, {"content": "Hello", "location": {"name": " "}}, {"content": "Hello", "location": {"name": "Soho", "latitude": 40.72}}, {"content": "Hello", "location": {"name": "Soho", "latitude": 91, "longitude": 0}}, {"content": "Hello", "content_format": "html"}, ): assert client.post("/api/v1/explore/posts", headers=author_headers, json=payload).status_code == 422 created = client.post("/api/v1/explore/posts", headers=author_headers, json={ "content": "Feeling **happy** and *relaxed* today", "content_format": "markdown", "feeling": " happy ", "location": {"name": " Blue Bottle Coffee, Soho NYC ", "latitude": 40.723, "longitude": -73.998}, }) assert created.status_code == 201 post = created.json() assert post["content_format"] == "markdown" assert post["feeling"] == "happy" assert post["location"] == { "name": "Blue Bottle Coffee, Soho NYC", "latitude": 40.723, "longitude": -73.998, } post_id = post["id"] read_paths = ( f"/api/v1/explore/posts/{post_id}", "/api/v1/explore/feed?limit=50", f"/api/v1/users/{author_id}/posts", "/api/v1/users/me/posts", ) for path in read_paths: headers = author_headers if path == "/api/v1/users/me/posts" else reader_headers response = client.get(path, headers=headers) assert response.status_code == 200 result = response.json() item = result if isinstance(result, dict) else next(value for value in result if value["id"] == post_id) assert item["content_format"] == "markdown" assert item["feeling"] == "happy" assert item["location"] == post["location"] me = client.get("/api/v1/auth/me", headers=author_headers) assert me.status_code == 200 assert next(item for item in me.json()["posts"] if item["id"] == post_id)["location"] == post["location"] legacy = client.post("/api/v1/explore/posts", headers=author_headers, json={"content": "Plain post"}) assert legacy.status_code == 201 assert legacy.json()["content_format"] == "plain" assert legacy.json()["feeling"] is None assert legacy.json()["location"] is None feeling_only = client.post("/api/v1/explore/posts", headers=author_headers, json={"feeling": "blessed"}) assert feeling_only.status_code == 201 assert feeling_only.json()["content"] == "" image_post = client.post("/api/v1/explore/posts", headers=author_headers, json={ "post_type": "image", "feeling": "excited", "location": {"name": "Soho NYC"}, }) assert image_post.status_code == 201 uploaded = client.post( f"/api/v1/explore/posts/{image_post.json()['id']}/media", headers=author_headers, files={"media": ("photo.png", b"image", "image/png")}, ) assert uploaded.status_code == 201 public_posts = client.get(f"/api/v1/users/{author_id}/posts", headers=reader_headers).json() image_result = next(item for item in public_posts if item["id"] == image_post.json()["id"]) assert image_result["feeling"] == "excited" assert image_result["location"] == {"name": "Soho NYC", "latitude": None, "longitude": None} assert len(image_result["media"]) == 1 video_post = client.post("/api/v1/explore/posts", headers=author_headers, json={ "post_type": "video", "content": "A short introduction", "feeling": "happy", }) assert video_post.status_code == 201 video_upload = client.post( f"/api/v1/explore/posts/{video_post.json()['id']}/media", headers=author_headers, files={"media": ("clip.mp4", b"video", "video/mp4")}, ) assert video_upload.status_code == 201 public_posts = client.get(f"/api/v1/users/{author_id}/posts", headers=reader_headers).json() video_result = next(item for item in public_posts if item["id"] == video_post.json()["id"]) assert video_result["post_type"] == "video" assert video_result["feeling"] == "happy" assert video_result["media"][0]["media_type"] == "video" def test_chat_requires_match_and_supports_messages(): with TestClient(app) as client: first_id, first_token, _, _ = register_user(client, full_name="Chat Alice") second_id, second_token, _, _ = register_user(client, full_name="Chat Bob") outsider_id, outsider_token, _, _ = register_user(client, full_name="Chat Outsider") first_headers = {"Authorization": f"Bearer {first_token}"} second_headers = {"Authorization": f"Bearer {second_token}"} outsider_headers = {"Authorization": f"Bearer {outsider_token}"} assert client.post( "/api/v1/chat/conversations/999999", headers=first_headers, ).status_code == 404 first_swipe = client.post( "/api/v1/meet/swipes", headers=first_headers, json={"target_user_id": second_id, "direction": "like"}, ) assert first_swipe.status_code == 200 second_swipe = client.post( "/api/v1/meet/swipes", headers=second_headers, json={"target_user_id": first_id, "direction": "like"}, ) match_id = second_swipe.json()["match_id"] assert second_swipe.json()["matched"] is True blocked = client.post( f"/api/v1/chat/conversations/{match_id}", headers=outsider_headers, ) assert blocked.status_code == 404 conversation = client.post( f"/api/v1/chat/conversations/{match_id}", headers=first_headers, ) assert conversation.status_code == 201 conversation_id = conversation.json()["id"] again = client.post( f"/api/v1/chat/conversations/{match_id}", headers=second_headers, ) assert again.status_code == 201 assert again.json()["id"] == conversation_id sent = client.post( f"/api/v1/chat/conversations/{conversation_id}/messages", headers=first_headers, json={"content": "你好，很高兴认识你"}, ) assert sent.status_code == 201 message_id = sent.json()["id"] reply = client.post( f"/api/v1/chat/conversations/{conversation_id}/messages", headers=second_headers, json={"content": "你好！"}, ) assert reply.status_code == 201 messages = client.get( f"/api/v1/chat/conversations/{conversation_id}/messages", headers=second_headers, ) assert [item["content"] for item in messages.json()] == ["你好，很高兴认识你", "你好！"] forbidden = client.delete(f"/api/v1/chat/messages/{message_id}", headers=second_headers) assert forbidden.status_code == 403 recalled = client.delete(f"/api/v1/chat/messages/{message_id}", headers=first_headers) assert recalled.status_code == 200 assert recalled.json()["is_deleted"] is True assert recalled.json()["content"] is None listed = client.get("/api/v1/chat/conversations", headers=first_headers) assert listed.status_code == 200 assert listed.json()[0]["id"] == conversation_id def test_explore_post_media_upload_access_and_cleanup(tmp_path, monkeypatch): monkeypatch.setattr("app.main.UPLOAD_ROOT", tmp_path) with TestClient(app) as client: _, author_token, _, _ = register_user(client, full_name="Media Author") _, reader_token, _, _ = register_user(client, full_name="Media Reader") author_headers = {"Authorization": f"Bearer {author_token}"} reader_headers = {"Authorization": f"Bearer {reader_token}"} created = client.post( "/api/v1/explore/posts", headers=author_headers, json={"content": "今天拍到的风景", "post_type": "image"}, ) assert created.status_code == 201 post_id = created.json()["id"] upload = client.post( f"/api/v1/explore/posts/{post_id}/media", headers=author_headers, params={"position": 0}, files={"media": ("view.png", b"image-content", "image/png")}, ) assert upload.status_code == 201 media_id = upload.json()["id"] assert upload.json()["media_type"] == "image" assert upload.json()["size_bytes"] == len(b"image-content") detail = client.get(f"/api/v1/explore/posts/{post_id}", headers=reader_headers) assert detail.status_code == 200 assert detail.json()["media"][0]["id"] == media_id downloaded = client.get(f"/api/v1/explore/media/{media_id}", headers=reader_headers) assert downloaded.status_code == 200 assert downloaded.content == b"image-content" assert client.delete(f"/api/v1/explore/media/{media_id}", headers=reader_headers).status_code == 403 assert client.delete(f"/api/v1/explore/posts/{post_id}", headers=author_headers).status_code == 200 assert client.get(f"/api/v1/explore/media/{media_id}", headers=reader_headers).status_code == 404 assert not list(tmp_path.rglob("*.png")) def test_feed_questions_answer_update_filter_and_delete(): with TestClient(app) as client: _, token, _, _ = register_user(client, full_name="Question User") headers = {"Authorization": f"Bearer {token}"} questions = client.get("/api/v1/feed/questions", headers=headers) assert questions.status_code == 200 assert len(questions.json()) >= 5 boolean_question = next(item for item in questions.json() if item["answer_type"] == "boolean") text_question = next(item for item in questions.json() if item["answer_type"] == "text") invalid = client.put( f"/api/v1/feed/questions/{boolean_question['id']}/answer", headers=headers, json={"answer": "yes"}, ) assert invalid.status_code == 422 boolean_answer = client.put( f"/api/v1/feed/questions/{boolean_question['id']}/answer", headers=headers, json={"answer": True}, ) assert boolean_answer.status_code == 200 assert boolean_answer.json()["answer"] is True updated = client.put( f"/api/v1/feed/questions/{boolean_question['id']}/answer", headers=headers, json={"answer": False}, ) assert updated.status_code == 200 assert updated.json()["answer"] is False text_answer = client.put( f"/api/v1/feed/questions/{text_question['id']}/answer", headers=headers, json={"answer": "一起喝咖啡并散步"}, ) assert text_answer.status_code == 200 assert text_answer.json()["answer"] == "一起喝咖啡并散步" answered = client.get("/api/v1/feed/answers/me", headers=headers) assert answered.status_code == 200 assert len(answered.json()) == 2 unanswered = client.get("/api/v1/feed/questions", headers=headers, params={"unanswered_only": True}) answered_ids = {boolean_question["id"], text_question["id"]} assert answered_ids.isdisjoint({item["id"] for item in unanswered.json()}) deleted = client.delete( f"/api/v1/feed/questions/{boolean_question['id']}/answer", headers=headers, ) assert deleted.status_code == 200 remaining = client.get("/api/v1/feed/answers/me", headers=headers) assert len(remaining.json()) == 1 def test_public_profile_includes_interests_and_recent_posts_without_private_activity(): with TestClient(app) as client: _, viewer_token, _, _ = register_user(client, full_name="Public Viewer") target_id, target_token, _, _ = register_user( client, full_name="Public Target", interests=["Travel", "Photography"], bio="A public profile bio", ) viewer_headers = {"Authorization": f"Bearer {viewer_token}"} target_headers = {"Authorization": f"Bearer {target_token}"} created = client.post( "/api/v1/explore/posts", headers=target_headers, json={"content": "Target profile post", "post_type": "text"}, ) assert created.status_code == 201 profile = client.get(f"/api/v1/users/{target_id}", headers=viewer_headers) assert profile.status_code == 200 payload = profile.json() assert payload["interests"] == ["Travel", "Photography"] assert len(payload["posts"]) == 1 assert payload["posts"][0]["content"] == "Target profile post" assert "email_or_phone" not in payload assert "status" not in payload assert "liked_users" not in payload assert "matches" not in payload