diff --git a/backend/tests/test_collaboration_ws.py b/backend/tests/test_collaboration_ws.py index 4f2b06fe..5361033d 100644 --- a/backend/tests/test_collaboration_ws.py +++ b/backend/tests/test_collaboration_ws.py @@ -1,7 +1,9 @@ """Tests for real-time collaboration WebSocket sessions.""" +import pytest from app import main as app_main from app.routers.collaboration import manager +from fastapi import WebSocketDisconnect from fastapi.testclient import TestClient client = TestClient(app_main.app) @@ -162,3 +164,87 @@ def test_collaboration_ping_returns_pong(): response = websocket.receive_json() assert response["type"] == "pong" + + +def test_presence_sync_broadcasts_on_join_and_leave(): + # Alice connects + with client.websocket_connect( + "/collaboration/ws/presence-test?name=Alice" + ) as alice: + alice_state = alice.receive_json() + alice_presence1 = alice.receive_json() + + assert alice_state["type"] == "session_state" + assert alice_presence1["type"] == "presence_update" + assert len(alice_presence1["users"]) == 1 + assert alice_presence1["users"][0]["name"] == "Alice" + + # Bob connects + with client.websocket_connect( + "/collaboration/ws/presence-test?name=Bob" + ) as bob: + bob_state = bob.receive_json() + + # Alice receives presence update for Bob's join + alice_presence2 = alice.receive_json() + # Bob receives presence update after joining + bob_presence = bob.receive_json() + + assert alice_presence2["type"] == "presence_update" + assert len(alice_presence2["users"]) == 2 + names = {u["name"] for u in alice_presence2["users"]} + assert names == {"Alice", "Bob"} + + assert bob_presence["type"] == "presence_update" + assert len(bob_presence["users"]) == 2 + + # Bob has disconnected. Alice should receive a presence update with only Alice remaining. + alice_presence3 = alice.receive_json() + assert alice_presence3["type"] == "presence_update" + assert len(alice_presence3["users"]) == 1 + assert alice_presence3["users"][0]["name"] == "Alice" + + +def test_presence_sync_handles_name_sanitization(): + # 1. Empty name parameter (or not provided) + with client.websocket_connect("/collaboration/ws/name-test") as ws1: + state1 = ws1.receive_json() + assert state1["users"][0]["name"] == "Anonymous" + + # 2. Whitespace-only name parameter + with client.websocket_connect("/collaboration/ws/name-test?name=%20%20%20") as ws2: + state2 = ws2.receive_json() + assert state2["users"][0]["name"] == "Anonymous" + + # 3. Truncate long name parameter (>40 characters) - FastAPI constraint raises WebSocketDisconnect (1008) + long_name = "A" * 50 + with pytest.raises(WebSocketDisconnect) as exc_info: + with client.websocket_connect(f"/collaboration/ws/name-test?name={long_name}"): + pass + assert exc_info.value.code == 1008 + + +def test_presence_sync_session_cleanup(): + # Alice joins and updates code + with client.websocket_connect("/collaboration/ws/cleanup-test?name=Alice") as alice: + alice_state = alice.receive_json() + alice.receive_json() # presence_update + + alice.send_json( + { + "type": "code_update", + "code": "print('hello')", + "language": "python", + "version": alice_state["version"], + } + ) + alice.receive_json() # receive echo of code update + + # Alice disconnected, room should be deleted. + # Bob joins the same room. Room should be recreated with fresh state. + with client.websocket_connect("/collaboration/ws/cleanup-test?name=Bob") as bob: + bob_state = bob.receive_json() + assert bob_state["code"] == "" + assert bob_state["version"] == 0 + assert len(bob_state["users"]) == 1 + assert bob_state["users"][0]["name"] == "Bob"