Coverage for tests / test_tutorial / test_websockets / test_tutorial003.py: 100%

1import importlib 4 ctx1bcde

2import time 4 ctx1bcde

3from types import ModuleType 4 ctx1bcde

4 

5import pytest 4 ctx1bcde

6from fastapi.testclient import TestClient 4 ctx1bcde

7 

8 

9@pytest.fixture( 4 ctx1bcde

10 name="mod", 

11 params=[ 

12 pytest.param("tutorial003_py310"), 

13 ], 

14) 

15def get_mod(request: pytest.FixtureRequest): 4 ctx1bcde

16 mod = importlib.import_module(f"docs_src.websockets_.{request.param}") 8 ctx1imjnkolp

17 

18 return mod 8 ctx1imjnkolp

19 

20 

21@pytest.fixture(name="html") 4 ctx1bcde

22def get_html(mod: ModuleType): 4 ctx1bcde

23 return mod.html 4 ctx1ijkl

24 

25 

26@pytest.fixture(name="client") 4 ctx1bcde

27def get_client(mod: ModuleType): 4 ctx1bcde

28 client = TestClient(mod.app) 8 ctx1imjnkolp

29 

30 return client 8 ctx1imjnkolp

31 

32 

33def test_get(client: TestClient, html: str): 4 ctx1bcde

34 response = client.get("/") 4 ctx1qrst

35 assert response.text == html 4 ctx1qrst

36 

37 

38def test_websocket_handle_disconnection(client: TestClient): 4 ctx1bcde

39 with ( 1 ctx1a

40 client.websocket_connect("/ws/1234") as connection, 

41 client.websocket_connect("/ws/5678") as connection_two, 

42 ): 

43 connection.send_text("Hello from 1234") 4 ctx1fgha

44 data1 = connection.receive_text() 4 ctx1fgha

45 assert data1 == "You wrote: Hello from 1234" 4 ctx1fgha

46 time.sleep(0.01) # Give server time to process broadcast 4 ctx1fgha

47 data2 = connection_two.receive_text() 4 ctx1fgha

48 client1_says = "Client #1234 says: Hello from 1234" 4 ctx1fgha

49 assert data2 == client1_says 4 ctx1fgha

50 data1 = connection.receive_text() 4 ctx1fgha

51 assert data1 == client1_says 4 ctx1fgha

52 connection_two.close() 4 ctx1fgha

53 time.sleep(0.01) # Give server time to process broadcast 4 ctx1fgha

54 data1 = connection.receive_text() 4 ctx1fgha

55 assert data1 == "Client #5678 left the chat" 4 ctx1fgha