fastapi

Форк
0
/
test_ws_dependencies.py 
73 строки · 2.1 Кб
1
import json
2
from typing import List
3

4
from fastapi import APIRouter, Depends, FastAPI, WebSocket
5
from fastapi.testclient import TestClient
6
from typing_extensions import Annotated
7

8

9
def dependency_list() -> List[str]:
10
    return []
11

12

13
DepList = Annotated[List[str], Depends(dependency_list)]
14

15

16
def create_dependency(name: str):
17
    def fun(deps: DepList):
18
        deps.append(name)
19

20
    return Depends(fun)
21

22

23
router = APIRouter(dependencies=[create_dependency("router")])
24
prefix_router = APIRouter(dependencies=[create_dependency("prefix_router")])
25
app = FastAPI(dependencies=[create_dependency("app")])
26

27

28
@app.websocket("/", dependencies=[create_dependency("index")])
29
async def index(websocket: WebSocket, deps: DepList):
30
    await websocket.accept()
31
    await websocket.send_text(json.dumps(deps))
32
    await websocket.close()
33

34

35
@router.websocket("/router", dependencies=[create_dependency("routerindex")])
36
async def routerindex(websocket: WebSocket, deps: DepList):
37
    await websocket.accept()
38
    await websocket.send_text(json.dumps(deps))
39
    await websocket.close()
40

41

42
@prefix_router.websocket("/", dependencies=[create_dependency("routerprefixindex")])
43
async def routerprefixindex(websocket: WebSocket, deps: DepList):
44
    await websocket.accept()
45
    await websocket.send_text(json.dumps(deps))
46
    await websocket.close()
47

48

49
app.include_router(router, dependencies=[create_dependency("router2")])
50
app.include_router(
51
    prefix_router, prefix="/prefix", dependencies=[create_dependency("prefix_router2")]
52
)
53

54

55
def test_index():
56
    client = TestClient(app)
57
    with client.websocket_connect("/") as websocket:
58
        data = json.loads(websocket.receive_text())
59
        assert data == ["app", "index"]
60

61

62
def test_routerindex():
63
    client = TestClient(app)
64
    with client.websocket_connect("/router") as websocket:
65
        data = json.loads(websocket.receive_text())
66
        assert data == ["app", "router2", "router", "routerindex"]
67

68

69
def test_routerprefixindex():
70
    client = TestClient(app)
71
    with client.websocket_connect("/prefix/") as websocket:
72
        data = json.loads(websocket.receive_text())
73
        assert data == ["app", "prefix_router2", "prefix_router", "routerprefixindex"]
74

Использование cookies

Мы используем файлы cookie в соответствии с Политикой конфиденциальности и Политикой использования cookies.

Нажимая кнопку «Принимаю», Вы даете АО «СберТех» согласие на обработку Ваших персональных данных в целях совершенствования нашего веб-сайта и Сервиса GitVerse, а также повышения удобства их использования.

Запретить использование cookies Вы можете самостоятельно в настройках Вашего браузера.