362 lines
16 KiB
Python
362 lines
16 KiB
Python
"""Integration tests for the friends/follow/block API:
|
|
|
|
- POST /api/friends/requests
|
|
- GET /api/friends/requests
|
|
- POST /api/friends/requests/{id}/accept
|
|
- POST /api/friends/requests/{id}/decline
|
|
- DELETE /api/friends/{user_id}
|
|
- POST /api/friends/{user_id}/follow
|
|
- DELETE /api/friends/{user_id}/follow
|
|
- POST /api/friends/{user_id}/block
|
|
- DELETE /api/friends/{user_id}/block
|
|
- GET /api/friends
|
|
- GET /api/friends/following
|
|
- GET /api/friends/blocked
|
|
|
|
app/routers/social_friends.py, built on the merged models in
|
|
app/models/social.py (Friendship/Follow/Block).
|
|
"""
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
|
|
def _register(client: TestClient, username: str) -> dict:
|
|
resp = client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"username": username,
|
|
"email": f"{username}@example.com",
|
|
"password": "secret123",
|
|
"date_of_birth": "1990-01-01",
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
token = resp.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
me = client.get("/api/users/me", headers=headers)
|
|
assert me.status_code == 200, me.text
|
|
return {"headers": headers, "id": me.json()["id"]}
|
|
|
|
|
|
class TestFriendRequestFlow:
|
|
def test_send_list_accept_makes_mutual_friendship(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "fr_flow_a")
|
|
b = _register(client, "fr_flow_b")
|
|
|
|
resp = client.post("/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"])
|
|
assert resp.status_code == 201, resp.text
|
|
request_id = resp.json()["id"]
|
|
assert resp.json()["status"] == "pending"
|
|
|
|
incoming = client.get("/api/friends/requests", headers=b["headers"])
|
|
assert incoming.status_code == 200
|
|
ids = [r["id"] for r in incoming.json()["requests"]]
|
|
assert request_id in ids
|
|
|
|
# Requester doesn't see it in their own incoming list.
|
|
a_incoming = client.get("/api/friends/requests", headers=a["headers"])
|
|
assert a_incoming.json()["requests"] == []
|
|
|
|
accept = client.post(f"/api/friends/requests/{request_id}/accept", headers=b["headers"])
|
|
assert accept.status_code == 200, accept.text
|
|
assert accept.json()["status"] == "accepted"
|
|
|
|
# Mutual: shows up in both users' friend lists.
|
|
a_friends = client.get("/api/friends", headers=a["headers"]).json()["friends"]
|
|
b_friends = client.get("/api/friends", headers=b["headers"]).json()["friends"]
|
|
assert any(u["id"] == b["id"] for u in a_friends)
|
|
assert any(u["id"] == a["id"] for u in b_friends)
|
|
|
|
def test_decline_removes_request_and_allows_re_request(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "fr_decl_a")
|
|
b = _register(client, "fr_decl_b")
|
|
|
|
resp = client.post("/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"])
|
|
request_id = resp.json()["id"]
|
|
|
|
decline = client.post(f"/api/friends/requests/{request_id}/decline", headers=b["headers"])
|
|
assert decline.status_code == 200, decline.text
|
|
|
|
assert client.get("/api/friends/requests", headers=b["headers"]).json()["requests"] == []
|
|
|
|
# A can send a new request now that the old one is gone.
|
|
resp2 = client.post("/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"])
|
|
assert resp2.status_code == 201, resp2.text
|
|
|
|
def test_accept_or_decline_of_others_request_is_404_not_403(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "fr_enum_a")
|
|
b = _register(client, "fr_enum_b")
|
|
c = _register(client, "fr_enum_c")
|
|
|
|
resp = client.post("/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"])
|
|
request_id = resp.json()["id"]
|
|
|
|
# c is neither requester nor addressee -- must not be able to act on it,
|
|
# and must not be able to distinguish "not yours" from "doesn't exist".
|
|
accept = client.post(f"/api/friends/requests/{request_id}/accept", headers=c["headers"])
|
|
assert accept.status_code == 404
|
|
|
|
decline = client.post(f"/api/friends/requests/{request_id}/decline", headers=c["headers"])
|
|
assert decline.status_code == 404
|
|
|
|
nonexistent = client.post("/api/friends/requests/999999999/accept", headers=c["headers"])
|
|
assert nonexistent.status_code == 404
|
|
|
|
def test_unfriend_removes_mutual_friendship(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "fr_unfr_a")
|
|
b = _register(client, "fr_unfr_b")
|
|
|
|
request_id = client.post(
|
|
"/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"]
|
|
).json()["id"]
|
|
client.post(f"/api/friends/requests/{request_id}/accept", headers=b["headers"])
|
|
|
|
unfriend = client.delete(f"/api/friends/{a['id']}", headers=b["headers"])
|
|
assert unfriend.status_code == 200, unfriend.text
|
|
|
|
assert client.get("/api/friends", headers=a["headers"]).json()["friends"] == []
|
|
assert client.get("/api/friends", headers=b["headers"]).json()["friends"] == []
|
|
|
|
# Idempotency / no dangling state: unfriending again is a clean 404, not 500.
|
|
again = client.delete(f"/api/friends/{a['id']}", headers=b["headers"])
|
|
assert again.status_code == 404
|
|
|
|
def test_send_request_by_username_populates_requester_username(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "fr_uname_a")
|
|
b = _register(client, "fr_uname_b")
|
|
|
|
resp = client.post(
|
|
"/api/friends/requests", json={"username": "fr_uname_b"}, headers=a["headers"]
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
req_data = resp.json()
|
|
assert req_data["status"] == "pending"
|
|
assert req_data.get("requester_username") == "fr_uname_a"
|
|
|
|
incoming = client.get("/api/friends/requests", headers=b["headers"])
|
|
assert incoming.status_code == 200, incoming.text
|
|
reqs = incoming.json()["requests"]
|
|
matching = [r for r in reqs if r["id"] == req_data["id"]]
|
|
assert len(matching) == 1
|
|
assert matching[0]["requester_username"] == "fr_uname_a"
|
|
|
|
|
|
class TestSelfAndDuplicateRejection:
|
|
def test_self_request_rejected(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "self_req_a")
|
|
resp = client.post("/api/friends/requests", json={"user_id": a["id"]}, headers=a["headers"])
|
|
assert resp.status_code == 400
|
|
|
|
def test_self_follow_rejected(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "self_follow_a")
|
|
resp = client.post(f"/api/friends/{a['id']}/follow", headers=a["headers"])
|
|
assert resp.status_code == 400
|
|
|
|
def test_self_block_rejected(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "self_block_a")
|
|
resp = client.post(f"/api/friends/{a['id']}/block", headers=a["headers"])
|
|
assert resp.status_code == 400
|
|
|
|
def test_duplicate_friend_request_is_conflict_not_500(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "dup_req_a")
|
|
b = _register(client, "dup_req_b")
|
|
assert client.post("/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"]).status_code == 201
|
|
dup = client.post("/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"])
|
|
assert dup.status_code == 409
|
|
|
|
def test_reverse_duplicate_friend_request_is_conflict(self):
|
|
"""b requesting a while a's request to b is still pending is treated
|
|
as a duplicate of the same pair, not a second independent request."""
|
|
client = TestClient(app)
|
|
a = _register(client, "dup_rev_a")
|
|
b = _register(client, "dup_rev_b")
|
|
assert client.post("/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"]).status_code == 201
|
|
reverse = client.post("/api/friends/requests", json={"user_id": a["id"]}, headers=b["headers"])
|
|
assert reverse.status_code == 409
|
|
|
|
def test_duplicate_follow_is_conflict_not_500(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "dup_follow_a")
|
|
b = _register(client, "dup_follow_b")
|
|
assert client.post(f"/api/friends/{b['id']}/follow", headers=a["headers"]).status_code == 201
|
|
dup = client.post(f"/api/friends/{b['id']}/follow", headers=a["headers"])
|
|
assert dup.status_code == 409
|
|
|
|
def test_duplicate_block_is_conflict_not_500(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "dup_block_a")
|
|
b = _register(client, "dup_block_b")
|
|
assert client.post(f"/api/friends/{b['id']}/block", headers=a["headers"]).status_code == 201
|
|
dup = client.post(f"/api/friends/{b['id']}/block", headers=a["headers"])
|
|
assert dup.status_code == 409
|
|
|
|
def test_friend_request_to_nonexistent_user_is_404(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "req_missing_a")
|
|
resp = client.post("/api/friends/requests", json={"user_id": 999999999}, headers=a["headers"])
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestFollow:
|
|
def test_follow_unfollow_and_list(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "follow_flow_a")
|
|
b = _register(client, "follow_flow_b")
|
|
|
|
follow = client.post(f"/api/friends/{b['id']}/follow", headers=a["headers"])
|
|
assert follow.status_code == 201, follow.text
|
|
|
|
following = client.get("/api/friends/following", headers=a["headers"]).json()["following"]
|
|
assert any(u["id"] == b["id"] for u in following)
|
|
|
|
# One-way: b does not follow a.
|
|
b_following = client.get("/api/friends/following", headers=b["headers"]).json()["following"]
|
|
assert b_following == []
|
|
|
|
unfollow = client.delete(f"/api/friends/{b['id']}/follow", headers=a["headers"])
|
|
assert unfollow.status_code == 200, unfollow.text
|
|
|
|
following_after = client.get("/api/friends/following", headers=a["headers"]).json()["following"]
|
|
assert following_after == []
|
|
|
|
# Unfollowing again is a clean 404, not 500.
|
|
again = client.delete(f"/api/friends/{b['id']}/follow", headers=a["headers"])
|
|
assert again.status_code == 404
|
|
|
|
def test_follow_requires_auth(self):
|
|
client = TestClient(app)
|
|
b = _register(client, "follow_noauth_b")
|
|
resp = client.post(f"/api/friends/{b['id']}/follow", json={})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
class TestBlock:
|
|
def test_block_clears_existing_friendship_and_follow(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "block_clear_a")
|
|
b = _register(client, "block_clear_b")
|
|
|
|
request_id = client.post(
|
|
"/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"]
|
|
).json()["id"]
|
|
client.post(f"/api/friends/requests/{request_id}/accept", headers=b["headers"])
|
|
client.post(f"/api/friends/{a['id']}/follow", headers=b["headers"])
|
|
|
|
assert any(
|
|
u["id"] == b["id"] for u in client.get("/api/friends", headers=a["headers"]).json()["friends"]
|
|
)
|
|
assert any(
|
|
u["id"] == a["id"]
|
|
for u in client.get("/api/friends/following", headers=b["headers"]).json()["following"]
|
|
)
|
|
|
|
block = client.post(f"/api/friends/{b['id']}/block", headers=a["headers"])
|
|
assert block.status_code == 201, block.text
|
|
|
|
assert client.get("/api/friends", headers=a["headers"]).json()["friends"] == []
|
|
assert client.get("/api/friends", headers=b["headers"]).json()["friends"] == []
|
|
assert client.get("/api/friends/following", headers=b["headers"]).json()["following"] == []
|
|
|
|
blocked_list = client.get("/api/friends/blocked", headers=a["headers"]).json()["blocked"]
|
|
assert any(u["id"] == b["id"] for u in blocked_list)
|
|
|
|
def test_block_prevents_new_friend_request_and_follow(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "block_prevent_a")
|
|
b = _register(client, "block_prevent_b")
|
|
|
|
client.post(f"/api/friends/{b['id']}/block", headers=a["headers"])
|
|
|
|
req = client.post("/api/friends/requests", json={"user_id": a["id"]}, headers=b["headers"])
|
|
assert req.status_code == 403
|
|
|
|
follow = client.post(f"/api/friends/{a['id']}/follow", headers=b["headers"])
|
|
assert follow.status_code == 403
|
|
|
|
def test_unblock_removes_block(self):
|
|
client = TestClient(app)
|
|
a = _register(client, "unblock_a")
|
|
b = _register(client, "unblock_b")
|
|
|
|
client.post(f"/api/friends/{b['id']}/block", headers=a["headers"])
|
|
unblock = client.delete(f"/api/friends/{b['id']}/block", headers=a["headers"])
|
|
assert unblock.status_code == 200, unblock.text
|
|
|
|
assert client.get("/api/friends/blocked", headers=a["headers"]).json()["blocked"] == []
|
|
|
|
# Idempotency: unblocking again is a clean 404, not 500.
|
|
again = client.delete(f"/api/friends/{b['id']}/block", headers=a["headers"])
|
|
assert again.status_code == 404
|
|
|
|
# After unblock, a fresh follow/request works again.
|
|
assert client.post(f"/api/friends/{a['id']}/follow", headers=b["headers"]).status_code == 201
|
|
|
|
|
|
class TestDataIsolation:
|
|
def test_list_endpoints_are_scoped_to_caller(self):
|
|
"""A third party's friends/following/blocked lists must never leak
|
|
into another user's response."""
|
|
client = TestClient(app)
|
|
a = _register(client, "iso_a")
|
|
b = _register(client, "iso_b")
|
|
c = _register(client, "iso_c")
|
|
|
|
request_id = client.post(
|
|
"/api/friends/requests", json={"user_id": b["id"]}, headers=a["headers"]
|
|
).json()["id"]
|
|
client.post(f"/api/friends/requests/{request_id}/accept", headers=b["headers"])
|
|
client.post(f"/api/friends/{b['id']}/follow", headers=a["headers"])
|
|
client.post(f"/api/friends/{b['id']}/block", headers=c["headers"])
|
|
|
|
c_friends = client.get("/api/friends", headers=c["headers"]).json()["friends"]
|
|
c_following = client.get("/api/friends/following", headers=c["headers"]).json()["following"]
|
|
assert c_friends == []
|
|
assert c_following == []
|
|
|
|
# c's block of b must not appear in a's or b's blocked list.
|
|
a_blocked = client.get("/api/friends/blocked", headers=a["headers"]).json()["blocked"]
|
|
assert a_blocked == []
|
|
|
|
def test_requires_auth_on_all_endpoints(self):
|
|
client = TestClient(app)
|
|
assert client.get("/api/friends").status_code == 401
|
|
assert client.get("/api/friends/requests").status_code == 401
|
|
assert client.get("/api/friends/following").status_code == 401
|
|
assert client.get("/api/friends/blocked").status_code == 401
|
|
assert client.post("/api/friends/requests", json={"user_id": 1}).status_code == 401
|
|
|
|
|
|
class TestUserSearchAndLookup:
|
|
def test_search_and_lookup_by_username(self):
|
|
client = TestClient(app)
|
|
u1 = _register(client, "usr_srch_1")
|
|
u2 = _register(client, "usr_srch_2")
|
|
|
|
# Search excludes current user
|
|
search_res = client.get("/api/users/search?q=usr_srch", headers=u1["headers"])
|
|
assert search_res.status_code == 200, search_res.text
|
|
names = [u["username"] for u in search_res.json()]
|
|
assert "usr_srch_2" in names
|
|
assert "usr_srch_1" not in names
|
|
|
|
# Lookup by username
|
|
lookup_res = client.get("/api/users/by-username/usr_srch_2", headers=u1["headers"])
|
|
assert lookup_res.status_code == 200, lookup_res.text
|
|
assert lookup_res.json()["username"] == "usr_srch_2"
|
|
assert lookup_res.json()["id"] == u2["id"]
|
|
|
|
# Lookup 404 for unknown user
|
|
missing = client.get("/api/users/by-username/nonexistent_usr_xyz", headers=u1["headers"])
|
|
assert missing.status_code == 404
|