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