You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1540 lines
68 KiB
1540 lines
68 KiB
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
|
|
|