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
37
38
39
40
41
42
43
44
45
46
47
48
49
|
import pytest
from aioredis.stream import StreamReader
from aioredis.parser import PyReader
from aioredis.errors import (
ProtocolError,
ReplyError
)
@pytest.fixture
def reader(loop):
reader = StreamReader(loop=loop)
reader.set_parser(
PyReader(protocolError=ProtocolError, replyError=ReplyError)
)
return reader
async def test_feed_and_parse(reader):
reader.feed_data(b'+PONG\r\n')
assert (await reader.readobj()) == b'PONG'
async def test_buffer_available_after_RST(reader):
reader.feed_data(b'+PONG\r\n')
reader.set_exception(Exception())
assert (await reader.readobj()) == b'PONG'
with pytest.raises(Exception):
await reader.readobj()
def test_feed_with_eof(reader):
reader.feed_eof()
with pytest.raises(AssertionError):
reader.feed_data(b'+PONG\r\n')
def test_feed_no_data(reader):
assert not reader.feed_data(None)
@pytest.mark.parametrize(
'read_method',
['read', 'readline', 'readuntil', 'readexactly']
)
async def test_read_flavors_not_supported(reader, read_method):
with pytest.raises(RuntimeError):
await getattr(reader, read_method)()
|
