File: test_payload.py

package info (click to toggle)
python-aiohttp 3.5.1-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, buster, sid
  • size: 5,612 kB
  • sloc: python: 36,917; ansic: 15,734; makefile: 365; sh: 83
file content (133 lines) | stat: -rw-r--r-- 3,606 bytes parent folder | download
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
import asyncio
from io import StringIO
from unittest import mock

import pytest
from async_generator import async_generator

from aiohttp import payload, streams


@pytest.fixture
def registry():
    old = payload.PAYLOAD_REGISTRY
    reg = payload.PAYLOAD_REGISTRY = payload.PayloadRegistry()
    yield reg
    payload.PAYLOAD_REGISTRY = old


class Payload(payload.Payload):

    async def write(self, writer):
        pass


def test_register_type(registry) -> None:
    class TestProvider:
        pass

    payload.register_payload(Payload, TestProvider)
    p = payload.get_payload(TestProvider())
    assert isinstance(p, Payload)


def test_register_unsupported_order(registry) -> None:
    class TestProvider:
        pass

    with pytest.raises(ValueError):
        payload.register_payload(Payload, TestProvider, order=object())


def test_payload_ctor() -> None:
    p = Payload('test', encoding='utf-8', filename='test.txt')
    assert p._value == 'test'
    assert p._encoding == 'utf-8'
    assert p.size is None
    assert p.filename == 'test.txt'
    assert p.content_type == 'text/plain'


def test_payload_content_type() -> None:
    p = Payload('test', headers={'content-type': 'application/json'})
    assert p.content_type == 'application/json'


def test_bytes_payload_default_content_type() -> None:
    p = payload.BytesPayload(b'data')
    assert p.content_type == 'application/octet-stream'


def test_bytes_payload_explicit_content_type() -> None:
    p = payload.BytesPayload(b'data', content_type='application/custom')
    assert p.content_type == 'application/custom'


def test_bytes_payload_bad_type() -> None:
    with pytest.raises(TypeError):
        payload.BytesPayload(object())


def test_string_payload() -> None:
    p = payload.StringPayload('test')
    assert p.encoding == 'utf-8'
    assert p.content_type == 'text/plain; charset=utf-8'

    p = payload.StringPayload('test', encoding='koi8-r')
    assert p.encoding == 'koi8-r'
    assert p.content_type == 'text/plain; charset=koi8-r'

    p = payload.StringPayload(
        'test', content_type='text/plain; charset=koi8-r')
    assert p.encoding == 'koi8-r'
    assert p.content_type == 'text/plain; charset=koi8-r'


def test_string_io_payload() -> None:
    s = StringIO('ű' * 5000)
    p = payload.StringIOPayload(s)
    assert p.encoding == 'utf-8'
    assert p.content_type == 'text/plain; charset=utf-8'
    assert p.size == 10000


def test_async_iterable_payload_default_content_type() -> None:
    @async_generator
    async def gen():
        pass

    p = payload.AsyncIterablePayload(gen())
    assert p.content_type == 'application/octet-stream'


def test_async_iterable_payload_explicit_content_type() -> None:
    @async_generator
    async def gen():
        pass

    p = payload.AsyncIterablePayload(gen(), content_type='application/custom')
    assert p.content_type == 'application/custom'


def test_async_iterable_payload_not_async_iterable() -> None:

    with pytest.raises(TypeError):
        payload.AsyncIterablePayload(object())


async def test_stream_reader_long_lines() -> None:
    loop = asyncio.get_event_loop()
    DATA = b'0' * 1024 ** 3

    stream = streams.StreamReader(mock.Mock(), loop=loop)
    stream.feed_data(DATA)
    stream.feed_eof()
    body = payload.get_payload(stream)

    writer = mock.Mock()
    writer.write.return_value = loop.create_future()
    writer.write.return_value.set_result(None)
    await body.write(writer)
    writer.write.assert_called_once_with(mock.ANY)
    (chunk,), _ = writer.write.call_args
    assert len(chunk) == len(DATA)