Coverage for src/lexigram/web/transport/websockets.py: 61%

1"""WebSocket support""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from starlette.websockets import WebSocket as StarletteWebSocket 

8 

9 

10class WebSocket: 

11 """Lexigram WebSocket wrapper around Starlette WebSocket""" 

12 

13 def __init__(self, starlette_ws: StarletteWebSocket): 

14 self._starlette = starlette_ws 

15 

16 async def accept(self, subprotocol: str | None = None) -> None: 

17 await self._starlette.accept(subprotocol=subprotocol) 

18 

19 async def close(self, code: int = 1000) -> None: 

20 await self._starlette.close(code=code) 

21 

22 async def receive_text(self) -> str: 

23 return await self._starlette.receive_text() 

24 

25 async def receive_json(self) -> Any: 

26 return await self._starlette.receive_json() 

27 

28 async def send_text(self, data: str) -> None: 

29 await self._starlette.send_text(data) 

30 

31 async def send_json(self, data: Any) -> None: 

32 await self._starlette.send_json(data) 

33 

34 async def send_bytes(self, data: bytes) -> None: 

35 await self._starlette.send_bytes(data) 

36 

37 @property 

38 def state(self) -> Any: 

39 """WebSocket state for middleware""" 

40 return self._starlette.state