247 lines
8.9 KiB
Python
247 lines
8.9 KiB
Python
"""Integration tests for in-experience world/room chat:
|
|
- POST /api/chat/world/{world_id}/messages
|
|
- GET /api/chat/world/{world_id}/messages
|
|
- POST /api/chat/room/{room_id}/messages
|
|
- GET /api/chat/room/{room_id}/messages
|
|
"""
|
|
|
|
import pytest
|
|
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"], "username": username}
|
|
|
|
|
|
def _make_world(client: TestClient, headers: dict, visibility: str = "public") -> int:
|
|
r = client.post("/api/worlds/", headers=headers, json={"name": "TestWorld", "visibility": visibility})
|
|
assert r.status_code == 201, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
def _make_friends(client: TestClient, user_a: dict, user_b: dict) -> None:
|
|
req = client.post("/api/friends/requests", json={"user_id": user_b["id"]}, headers=user_a["headers"])
|
|
assert req.status_code == 201, req.text
|
|
req_id = req.json()["id"]
|
|
acc = client.post(f"/api/friends/requests/{req_id}/accept", headers=user_b["headers"])
|
|
assert acc.status_code == 200, acc.text
|
|
|
|
|
|
class TestWorldChat:
|
|
def test_send_and_get_world_message(self):
|
|
client = TestClient(app)
|
|
alice = _register(client, "wch_alice1")
|
|
bob = _register(client, "wch_bob1")
|
|
|
|
world_id = _make_world(client, alice["headers"], visibility="public")
|
|
|
|
# Bob posts to Alice's public world
|
|
post_resp = client.post(
|
|
f"/api/chat/world/{world_id}/messages",
|
|
headers=bob["headers"],
|
|
json={"body": "Hello world from Bob!"},
|
|
)
|
|
assert post_resp.status_code == 201, post_resp.text
|
|
msg_data = post_resp.json()
|
|
assert msg_data["world_id"] == world_id
|
|
assert msg_data["room_id"] is None
|
|
assert msg_data["sender_id"] == bob["id"]
|
|
assert msg_data["body"] == "Hello world from Bob!"
|
|
assert msg_data["filtered"] is False
|
|
assert msg_data["author"]["username"] == "wch_bob1"
|
|
assert msg_data["author"]["id"] == bob["id"]
|
|
|
|
# Alice gets world chat history
|
|
get_resp = client.get(
|
|
f"/api/chat/world/{world_id}/messages",
|
|
headers=alice["headers"],
|
|
)
|
|
assert get_resp.status_code == 200, get_resp.text
|
|
history = get_resp.json()
|
|
assert len(history["messages"]) == 1
|
|
assert history["messages"][0]["id"] == msg_data["id"]
|
|
assert history["messages"][0]["author"]["username"] == "wch_bob1"
|
|
|
|
def test_room_scoped_messages(self):
|
|
client = TestClient(app)
|
|
alice = _register(client, "wch_alice2")
|
|
world_id = _make_world(client, alice["headers"])
|
|
|
|
# Fetch auto-created main room
|
|
rooms_resp = client.get(f"/api/worlds/{world_id}/rooms", headers=alice["headers"])
|
|
assert rooms_resp.status_code == 200, rooms_resp.text
|
|
rooms = rooms_resp.json()
|
|
assert len(rooms) >= 1
|
|
room_id = rooms[0]["id"]
|
|
|
|
# Post directly via room endpoint
|
|
post_room = client.post(
|
|
f"/api/chat/room/{room_id}/messages",
|
|
headers=alice["headers"],
|
|
json={"body": "Message in room"},
|
|
)
|
|
assert post_room.status_code == 201, post_room.text
|
|
msg_data = post_room.json()
|
|
assert msg_data["world_id"] == world_id
|
|
assert msg_data["room_id"] == room_id
|
|
assert msg_data["body"] == "Message in room"
|
|
|
|
# Get room history
|
|
get_room = client.get(
|
|
f"/api/chat/room/{room_id}/messages",
|
|
headers=alice["headers"],
|
|
)
|
|
assert get_room.status_code == 200, get_room.text
|
|
messages = get_room.json()["messages"]
|
|
assert len(messages) == 1
|
|
assert messages[0]["id"] == msg_data["id"]
|
|
|
|
# Also visible when querying world history with room_id filter
|
|
get_world_room = client.get(
|
|
f"/api/chat/world/{world_id}/messages?room_id={room_id}",
|
|
headers=alice["headers"],
|
|
)
|
|
assert get_world_room.status_code == 200, get_world_room.text
|
|
assert len(get_world_room.json()["messages"]) == 1
|
|
|
|
def test_visibility_access_control(self):
|
|
client = TestClient(app)
|
|
alice = _register(client, "wch_alice3")
|
|
bob = _register(client, "wch_bob3")
|
|
charlie = _register(client, "wch_charlie3")
|
|
|
|
# Private world: only Alice
|
|
priv_world_id = _make_world(client, alice["headers"], visibility="private")
|
|
|
|
# Bob cannot send or read
|
|
bob_send = client.post(
|
|
f"/api/chat/world/{priv_world_id}/messages",
|
|
headers=bob["headers"],
|
|
json={"body": "Trying to trespass"},
|
|
)
|
|
assert bob_send.status_code == 403, bob_send.text
|
|
|
|
bob_read = client.get(
|
|
f"/api/chat/world/{priv_world_id}/messages",
|
|
headers=bob["headers"],
|
|
)
|
|
assert bob_read.status_code == 403, bob_read.text
|
|
|
|
# Alice (owner) can send and read
|
|
alice_send = client.post(
|
|
f"/api/chat/world/{priv_world_id}/messages",
|
|
headers=alice["headers"],
|
|
json={"body": "My private sanctum"},
|
|
)
|
|
assert alice_send.status_code == 201, alice_send.text
|
|
|
|
# Friends-only world
|
|
fr_world_id = _make_world(client, alice["headers"], visibility="friends_only")
|
|
_make_friends(client, alice, bob)
|
|
|
|
# Bob (friend) can post
|
|
bob_fr_send = client.post(
|
|
f"/api/chat/world/{fr_world_id}/messages",
|
|
headers=bob["headers"],
|
|
json={"body": "Hello friend!"},
|
|
)
|
|
assert bob_fr_send.status_code == 201, bob_fr_send.text
|
|
|
|
# Charlie (not friend) cannot post or read
|
|
charlie_send = client.post(
|
|
f"/api/chat/world/{fr_world_id}/messages",
|
|
headers=charlie["headers"],
|
|
json={"body": "Can I join?"},
|
|
)
|
|
assert charlie_send.status_code == 403, charlie_send.text
|
|
|
|
charlie_read = client.get(
|
|
f"/api/chat/world/{fr_world_id}/messages",
|
|
headers=charlie["headers"],
|
|
)
|
|
assert charlie_read.status_code == 403, charlie_read.text
|
|
|
|
def test_not_found_validation(self):
|
|
client = TestClient(app)
|
|
alice = _register(client, "wch_alice4")
|
|
|
|
# Non-existent world
|
|
res = client.post(
|
|
"/api/chat/world/999999/messages",
|
|
headers=alice["headers"],
|
|
json={"body": "Ghost world"},
|
|
)
|
|
assert res.status_code == 404, res.text
|
|
|
|
# Non-existent room
|
|
world_id = _make_world(client, alice["headers"])
|
|
res2 = client.post(
|
|
f"/api/chat/world/{world_id}/messages",
|
|
headers=alice["headers"],
|
|
json={"body": "Ghost room", "room_id": 999999},
|
|
)
|
|
assert res2.status_code == 404, res2.text
|
|
|
|
def test_text_filtering_in_world_chat(self):
|
|
client = TestClient(app)
|
|
alice = _register(client, "wch_alice5")
|
|
world_id = _make_world(client, alice["headers"])
|
|
|
|
# Default filter redacts forbidden words
|
|
res = client.post(
|
|
f"/api/chat/world/{world_id}/messages",
|
|
headers=alice["headers"],
|
|
json={"body": "you are a bastard"},
|
|
)
|
|
assert res.status_code == 201, res.text
|
|
data = res.json()
|
|
assert data["filtered"] is True
|
|
assert "bastard" not in data["body"]
|
|
assert "*" in data["body"]
|
|
|
|
def test_pagination_and_next_cursor(self):
|
|
client = TestClient(app)
|
|
alice = _register(client, "wch_alice6")
|
|
world_id = _make_world(client, alice["headers"])
|
|
|
|
for i in range(5):
|
|
r = client.post(
|
|
f"/api/chat/world/{world_id}/messages",
|
|
headers=alice["headers"],
|
|
json={"body": f"Msg {i}"},
|
|
)
|
|
assert r.status_code == 201
|
|
|
|
page1 = client.get(
|
|
f"/api/chat/world/{world_id}/messages?limit=2",
|
|
headers=alice["headers"],
|
|
).json()
|
|
assert len(page1["messages"]) == 2
|
|
assert page1["next_cursor"] is not None
|
|
assert page1["messages"][0]["body"] == "Msg 4"
|
|
assert page1["messages"][1]["body"] == "Msg 3"
|
|
|
|
page2 = client.get(
|
|
f"/api/chat/world/{world_id}/messages?limit=2&before={page1['next_cursor']}",
|
|
headers=alice["headers"],
|
|
).json()
|
|
assert len(page2["messages"]) == 2
|
|
assert page2["messages"][0]["body"] == "Msg 2"
|
|
assert page2["messages"][1]["body"] == "Msg 1"
|