1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
import asyncio
from unittest import mock
import pytest
from aiohttp._websocket.models import WSMessage, WSMsgType
from aiohttp._websocket.reader import WebSocketDataQueue
from aiohttp.base_protocol import BaseProtocol
@pytest.fixture
def protocol() -> BaseProtocol:
return mock.create_autospec(BaseProtocol, spec_set=True, instance=True, _reading_paused=False) # type: ignore[no-any-return]
@pytest.fixture
def buffer(
loop: asyncio.AbstractEventLoop, protocol: BaseProtocol
) -> WebSocketDataQueue:
return WebSocketDataQueue(protocol, limit=1, loop=loop)
class TestWebSocketDataQueue:
def test_feed_pause(self, buffer: WebSocketDataQueue) -> None:
buffer._protocol._reading_paused = False
for _ in range(3):
buffer.feed_data(WSMessage(data=b"x", type=WSMsgType.BINARY, extra=""), 1)
assert buffer._protocol.pause_reading.called # type: ignore[attr-defined]
async def test_resume_on_read(self, buffer: WebSocketDataQueue) -> None:
buffer.feed_data(WSMessage(data=b"x", type=WSMsgType.BINARY, extra=""), 1)
buffer._protocol._reading_paused = True
await buffer.read()
assert buffer._protocol.resume_reading.called # type: ignore[attr-defined]
|
