mirror
/
tiangolo.fastapi
mirror of https://github.com/tiangolo/fastapi
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

✅ Simplify tests for websockets (#13202)

pull/13237/head
Alejandra 2 months ago
committed by GitHub
parent
9f3edbf537
commit
280fe73c03
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
5 changed files with 26 additions and 401 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 33
      tests/test_tutorial/test_websockets/test_tutorial002.py
  2. 88
      tests/test_tutorial/test_websockets/test_tutorial002_an.py
  3. 102
      tests/test_tutorial/test_websockets/test_tutorial002_an_py310.py
  4. 102
      tests/test_tutorial/test_websockets/test_tutorial002_an_py39.py
  5. 102
      tests/test_tutorial/test_websockets/test_tutorial002_py310.py

33
tests/test_tutorial/test_websockets/test_tutorial002.py
View File

@ -1,18 +1,37 @@
import importlib
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect
from docs_src.websockets.tutorial002 import app
from ...utils import needs_py39, needs_py310
@pytest.fixture(
name="app",
params=[
"tutorial002",
pytest.param("tutorial002_py310", marks=needs_py310),
"tutorial002_an",
pytest.param("tutorial002_an_py39", marks=needs_py39),
pytest.param("tutorial002_an_py310", marks=needs_py310),
],
)
def get_app(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.websockets.{request.param}")
return mod.app
def test_main():
def test_main(app: FastAPI):
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200, response.text
assert b"<!DOCTYPE html>" in response.content
def test_websocket_with_cookie():
def test_websocket_with_cookie(app: FastAPI):
client = TestClient(app, cookies={"session": "fakesession"})
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws") as websocket:
@ -30,7 +49,7 @@ def test_websocket_with_cookie():
assert data == f"Message text was: {message}, for item ID: foo"
def test_websocket_with_header():
def test_websocket_with_header(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/bar/ws?token=some-token") as websocket:
@ -48,7 +67,7 @@ def test_websocket_with_header():
assert data == f"Message text was: {message}, for item ID: bar"
def test_websocket_with_header_and_query():
def test_websocket_with_header_and_query(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/2/ws?q=3&token=some-token") as websocket:
@ -70,7 +89,7 @@ def test_websocket_with_header_and_query():
assert data == f"Message text was: {message}, for item ID: 2"
def test_websocket_no_credentials():
def test_websocket_no_credentials(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws"):
@ -79,7 +98,7 @@ def test_websocket_no_credentials():
) # pragma: no cover
def test_websocket_invalid_data():
def test_websocket_invalid_data(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"):

88
tests/test_tutorial/test_websockets/test_tutorial002_an.py
View File

@ -1,88 +0,0 @@
import pytest
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect
from docs_src.websockets.tutorial002_an import app
def test_main():
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200, response.text
assert b"<!DOCTYPE html>" in response.content
def test_websocket_with_cookie():
client = TestClient(app, cookies={"session": "fakesession"})
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: fakesession"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: foo"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: fakesession"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: foo"
def test_websocket_with_header():
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/bar/ws?token=some-token") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: bar"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: bar"
def test_websocket_with_header_and_query():
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/2/ws?q=3&token=some-token") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == "Query parameter q is: 3"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: 2"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == "Query parameter q is: 3"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: 2"
def test_websocket_no_credentials():
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws"):
pytest.fail(
"did not raise WebSocketDisconnect on __enter__"
) # pragma: no cover
def test_websocket_invalid_data():
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"):
pytest.fail(
"did not raise WebSocketDisconnect on __enter__"
) # pragma: no cover

102
tests/test_tutorial/test_websockets/test_tutorial002_an_py310.py
View File

@ -1,102 +0,0 @@
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect
from ...utils import needs_py310
@pytest.fixture(name="app")
def get_app():
from docs_src.websockets.tutorial002_an_py310 import app
return app
@needs_py310
def test_main(app: FastAPI):
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200, response.text
assert b"<!DOCTYPE html>" in response.content
@needs_py310
def test_websocket_with_cookie(app: FastAPI):
client = TestClient(app, cookies={"session": "fakesession"})
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: fakesession"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: foo"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: fakesession"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: foo"
@needs_py310
def test_websocket_with_header(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/bar/ws?token=some-token") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: bar"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: bar"
@needs_py310
def test_websocket_with_header_and_query(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/2/ws?q=3&token=some-token") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == "Query parameter q is: 3"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: 2"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == "Query parameter q is: 3"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: 2"
@needs_py310
def test_websocket_no_credentials(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws"):
pytest.fail(
"did not raise WebSocketDisconnect on __enter__"
) # pragma: no cover
@needs_py310
def test_websocket_invalid_data(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"):
pytest.fail(
"did not raise WebSocketDisconnect on __enter__"
) # pragma: no cover

102
tests/test_tutorial/test_websockets/test_tutorial002_an_py39.py
View File

@ -1,102 +0,0 @@
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect
from ...utils import needs_py39
@pytest.fixture(name="app")
def get_app():
from docs_src.websockets.tutorial002_an_py39 import app
return app
@needs_py39
def test_main(app: FastAPI):
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200, response.text
assert b"<!DOCTYPE html>" in response.content
@needs_py39
def test_websocket_with_cookie(app: FastAPI):
client = TestClient(app, cookies={"session": "fakesession"})
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: fakesession"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: foo"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: fakesession"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: foo"
@needs_py39
def test_websocket_with_header(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/bar/ws?token=some-token") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: bar"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: bar"
@needs_py39
def test_websocket_with_header_and_query(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/2/ws?q=3&token=some-token") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == "Query parameter q is: 3"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: 2"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == "Query parameter q is: 3"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: 2"
@needs_py39
def test_websocket_no_credentials(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws"):
pytest.fail(
"did not raise WebSocketDisconnect on __enter__"
) # pragma: no cover
@needs_py39
def test_websocket_invalid_data(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"):
pytest.fail(
"did not raise WebSocketDisconnect on __enter__"
) # pragma: no cover

102
tests/test_tutorial/test_websockets/test_tutorial002_py310.py
View File

@ -1,102 +0,0 @@
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect
from ...utils import needs_py310
@pytest.fixture(name="app")
def get_app():
from docs_src.websockets.tutorial002_py310 import app
return app
@needs_py310
def test_main(app: FastAPI):
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200, response.text
assert b"<!DOCTYPE html>" in response.content
@needs_py310
def test_websocket_with_cookie(app: FastAPI):
client = TestClient(app, cookies={"session": "fakesession"})
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: fakesession"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: foo"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: fakesession"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: foo"
@needs_py310
def test_websocket_with_header(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/bar/ws?token=some-token") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: bar"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: bar"
@needs_py310
def test_websocket_with_header_and_query(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/2/ws?q=3&token=some-token") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == "Query parameter q is: 3"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: 2"
message = "Message two"
websocket.send_text(message)
data = websocket.receive_text()
assert data == "Session cookie or query token value is: some-token"
data = websocket.receive_text()
assert data == "Query parameter q is: 3"
data = websocket.receive_text()
assert data == f"Message text was: {message}, for item ID: 2"
@needs_py310
def test_websocket_no_credentials(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws"):
pytest.fail(
"did not raise WebSocketDisconnect on __enter__"
) # pragma: no cover
@needs_py310
def test_websocket_invalid_data(app: FastAPI):
client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"):
pytest.fail(
"did not raise WebSocketDisconnect on __enter__"
) # pragma: no cover
Write Preview
Loading…
Cancel
Save