Coverage for tests / test_tutorial / test_websockets / test_tutorial002.py: 100%
75 statements
« prev ^ index » next coverage.py v7.13.3, created at 2026-02-12 18:15 +0000
1import importlib 1demf
3import pytest 1demf
4from fastapi import FastAPI 1demf
5from fastapi.testclient import TestClient 1demf
6from fastapi.websockets import WebSocketDisconnect 1demf
8from ...utils import needs_py310 1demf
11@pytest.fixture( 1demf
12 name="app",
13 params=[
14 pytest.param("tutorial002_py310", marks=needs_py310),
15 pytest.param("tutorial002_an_py310", marks=needs_py310),
16 ],
17)
18def get_app(request: pytest.FixtureRequest): 1demf
19 mod = importlib.import_module(f"docs_src.websockets.{request.param}") 1def
21 return mod.app 1def
24def test_main(app: FastAPI): 1demf
25 client = TestClient(app) 1nop
26 response = client.get("/") 1nop
27 assert response.status_code == 200, response.text 1nop
28 assert b"<!DOCTYPE html>" in response.content 1nop
31def test_websocket_with_cookie(app: FastAPI): 1demf
32 client = TestClient(app, cookies={"session": "fakesession"}) 1ghi
33 with pytest.raises(WebSocketDisconnect): 1ghi
34 with client.websocket_connect("/items/foo/ws") as websocket: 1ghi
35 message = "Message one" 1ghi
36 websocket.send_text(message) 1ghi
37 data = websocket.receive_text() 1ghi
38 assert data == "Session cookie or query token value is: fakesession" 1ghi
39 data = websocket.receive_text() 1ghi
40 assert data == f"Message text was: {message}, for item ID: foo" 1ghi
41 message = "Message two" 1ghi
42 websocket.send_text(message) 1ghi
43 data = websocket.receive_text() 1ghi
44 assert data == "Session cookie or query token value is: fakesession" 1ghi
45 data = websocket.receive_text() 1ghi
46 assert data == f"Message text was: {message}, for item ID: foo" 1ghi
49def test_websocket_with_header(app: FastAPI): 1demf
50 client = TestClient(app) 1jkl
51 with pytest.raises(WebSocketDisconnect): 1jkl
52 with client.websocket_connect("/items/bar/ws?token=some-token") as websocket: 1jkl
53 message = "Message one" 1jkl
54 websocket.send_text(message) 1jkl
55 data = websocket.receive_text() 1jkl
56 assert data == "Session cookie or query token value is: some-token" 1jkl
57 data = websocket.receive_text() 1jkl
58 assert data == f"Message text was: {message}, for item ID: bar" 1jkl
59 message = "Message two" 1jkl
60 websocket.send_text(message) 1jkl
61 data = websocket.receive_text() 1jkl
62 assert data == "Session cookie or query token value is: some-token" 1jkl
63 data = websocket.receive_text() 1jkl
64 assert data == f"Message text was: {message}, for item ID: bar" 1jkl
67def test_websocket_with_header_and_query(app: FastAPI): 1demf
68 client = TestClient(app) 1abc
69 with pytest.raises(WebSocketDisconnect): 1abc
70 with client.websocket_connect("/items/2/ws?q=3&token=some-token") as websocket: 1abc
71 message = "Message one" 1abc
72 websocket.send_text(message) 1abc
73 data = websocket.receive_text() 1abc
74 assert data == "Session cookie or query token value is: some-token" 1abc
75 data = websocket.receive_text() 1abc
76 assert data == "Query parameter q is: 3" 1abc
77 data = websocket.receive_text() 1abc
78 assert data == f"Message text was: {message}, for item ID: 2" 1abc
79 message = "Message two" 1abc
80 websocket.send_text(message) 1abc
81 data = websocket.receive_text() 1abc
82 assert data == "Session cookie or query token value is: some-token" 1abc
83 data = websocket.receive_text() 1abc
84 assert data == "Query parameter q is: 3" 1abc
85 data = websocket.receive_text() 1abc
86 assert data == f"Message text was: {message}, for item ID: 2" 1abc
89def test_websocket_no_credentials(app: FastAPI): 1demf
90 client = TestClient(app) 1qrs
91 with pytest.raises(WebSocketDisconnect): 1qrs
92 with client.websocket_connect("/items/foo/ws"): 1qrs
93 pytest.fail(
94 "did not raise WebSocketDisconnect on __enter__"
95 ) # pragma: no cover
98def test_websocket_invalid_data(app: FastAPI): 1demf
99 client = TestClient(app) 1tuv
100 with pytest.raises(WebSocketDisconnect): 1tuv
101 with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"): 1tuv
102 pytest.fail(
103 "did not raise WebSocketDisconnect on __enter__"
104 ) # pragma: no cover