Coverage for backend/tests/test_ws.py: 33%

52 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-04-17 17:55 +0000

1import pytest 

2from fastapi.testclient import TestClient 

3 

4from backend.dependencies import LobbyManager, lobby_manager 

5from backend.game import Message 

6from backend.hello import app 

7from backend.models import Burger, Order, OrderSubmission 

8 

9lm = LobbyManager(["Lettuce"]) 

10 

11 

12def lobby_manager_override() -> LobbyManager: 

13 """Override lobby manager.""" 

14 return lm 

15 

16 

17app.dependency_overrides[lobby_manager] = lobby_manager_override 

18 

19 

20@pytest.fixture 

21def lobby_client() -> TestClient: 

22 """Create a TestClient with a lobby created.""" 

23 client = TestClient(app) 

24 client.post("lobby") 

25 return client 

26 

27 

28@pytest.mark.skip(reason="Test function hangs occasionally") 

29def test_websocket(lobby_client: TestClient) -> None: 

30 """Test that websocket connection does not hang.""" 

31 player_1_response = lobby_client.post("/lobby/join", json={"code": ["Lettuce", "Lettuce", "Lettuce"]}) 

32 lobby_client.post("/lobby/join", json={"code": ["Lettuce", "Lettuce", "Lettuce"]}) 

33 

34 player_1_id = player_1_response.json()["id"] 

35 

36 with lobby_client.websocket_connect("/ws") as websocket: 

37 websocket.send_text( 

38 '{"data": {"type": "initializer", "code": ["Lettuce", "Lettuce", "Lettuce"], "id": "%s"}}' % player_1_id 

39 ) 

40 websocket.send_text('{"data": {"type": "lobby_lifecycle", "lifecycle_type": "game_start"}}') 

41 msg = websocket.receive_json()["data"] 

42 

43 order = msg["order"] 

44 burger = order["burger"] 

45 

46 # since we have only one cook, we should have a burger, but no drink or 

47 # side. 

48 # technically speaking, we _could_ inject a random.Random for testing, 

49 # but that seems like too much effort for not much gain 

50 assert burger["ingredients"][0] == "bottom bun" 

51 assert burger["ingredients"][-1] == "top bun" 

52 

53 assert order["drink"] is None 

54 assert order["side"] is None 

55 

56 websocket.close() 

57 

58 

59@pytest.mark.skip(reason="Test function hangs occasionally") 

60def test_websocket_spam_chat(lobby_client: TestClient) -> None: 

61 """Test that a bunch of chat messages are received and broadcast without pausing.""" 

62 id = lobby_client.post("/lobby/join", json={"code": ["Lettuce", "Lettuce", "Lettuce"]}).json()["id"] 

63 

64 client_2 = TestClient(app) 

65 id_2 = client_2.post("/lobby/join", json={"code": ["Lettuce", "Lettuce", "Lettuce"]}).json()["id"] 

66 

67 with ( 

68 lobby_client.websocket_connect("/ws") as websocket_1, 

69 client_2.websocket_connect("/ws") as websocket_2, 

70 ): 

71 websocket_1.send_text( 

72 '{"data": {"type": "initializer", "code": ["Lettuce", "Lettuce", "Lettuce"], "id": "%s"}}' % id 

73 ) 

74 websocket_2.send_text( 

75 '{"data": {"type": "initializer", "code": ["Lettuce", "Lettuce", "Lettuce"], "id": "%s"}}' % id_2 

76 ) 

77 

78 for _ in range(10): 

79 websocket_1.send_text('{"data": {"type": "chat", "typing": true, "id": "%s"}}' % id) 

80 

81 for _ in range(10): 

82 assert websocket_2.receive_json()["data"]["id"] == id 

83 

84 

85@pytest.mark.skip(reason="Test function hangs occasionally") 

86def test_websocket_submit_burger_order(lobby_client: TestClient) -> None: 

87 """Test that we get a score after submitting an order.""" 

88 id = lobby_client.post("/lobby/join", json={"code": ["Lettuce", "Lettuce", "Lettuce"]}).json()["id"] 

89 with lobby_client.websocket_connect("/ws") as websocket: 

90 websocket.send_text( 

91 '{"data": {"type": "initializer", "code": ["Lettuce", "Lettuce", "Lettuce"], "id": "%s"}}' % id 

92 ) 

93 websocket.send_text('{"data": {"type": "lobby_lifecycle", "lifecycle_type": "game_start"}}') 

94 

95 order = websocket.receive_json()["data"]["order"] 

96 m = Message( 

97 data=OrderSubmission( 

98 order=Order(burger=Burger(ingredients=order["burger"]["ingredients"]), drink=None, side=None) 

99 ) 

100 ) 

101 websocket.send_text(m.model_dump_json()) 

102 # print(websocket.receive_json())