File: test_web_exceptions.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 (187 lines) | stat: -rw-r--r-- 5,696 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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
import collections
import re
from traceback import format_exception
from unittest import mock

import pytest

from aiohttp import helpers, signals, web
from aiohttp.test_utils import make_mocked_request


@pytest.fixture
def buf():
    return bytearray()


@pytest.fixture
def http_request(buf):
    method = 'GET'
    path = '/'
    writer = mock.Mock()
    writer.drain.return_value = ()

    def append(data=b''):
        buf.extend(data)
        return helpers.noop()

    async def write_headers(status_line, headers):
        headers = status_line + '\r\n' + ''.join(
            [k + ': ' + v + '\r\n' for k, v in headers.items()])
        headers = headers.encode('utf-8') + b'\r\n'
        buf.extend(headers)

    writer.buffer_data.side_effect = append
    writer.write.side_effect = append
    writer.write_eof.side_effect = append
    writer.write_headers.side_effect = write_headers

    app = mock.Mock()
    app._debug = False
    app.on_response_prepare = signals.Signal(app)
    app.on_response_prepare.freeze()
    req = make_mocked_request(method, path, app=app, writer=writer)
    return req


def test_all_http_exceptions_exported() -> None:
    assert 'HTTPException' in web.__all__
    for name in dir(web):
        if name.startswith('_'):
            continue
        obj = getattr(web, name)
        if isinstance(obj, type) and issubclass(obj, web.HTTPException):
            assert name in web.__all__


async def test_HTTPOk(buf, http_request) -> None:
    resp = web.HTTPOk()
    await resp.prepare(http_request)
    await resp.write_eof()
    txt = buf.decode('utf8')
    assert re.match(('HTTP/1.1 200 OK\r\n'
                     'Content-Type: text/plain; charset=utf-8\r\n'
                     'Content-Length: 7\r\n'
                     'Date: .+\r\n'
                     'Server: .+\r\n\r\n'
                     '200: OK'), txt)


def test_terminal_classes_has_status_code() -> None:
    terminals = set()
    for name in dir(web):
        obj = getattr(web, name)
        if isinstance(obj, type) and issubclass(obj, web.HTTPException):
            terminals.add(obj)

    dup = frozenset(terminals)
    for cls1 in dup:
        for cls2 in dup:
            if cls1 in cls2.__bases__:
                terminals.discard(cls1)

    for cls in terminals:
        assert cls.status_code is not None
    codes = collections.Counter(cls.status_code for cls in terminals)
    assert None not in codes
    assert 1 == codes.most_common(1)[0][1]


async def test_HTTPFound(buf, http_request) -> None:
    resp = web.HTTPFound(location='/redirect')
    assert '/redirect' == resp.location
    assert '/redirect' == resp.headers['location']
    await resp.prepare(http_request)
    await resp.write_eof()
    txt = buf.decode('utf8')
    assert re.match('HTTP/1.1 302 Found\r\n'
                    'Content-Type: text/plain; charset=utf-8\r\n'
                    'Location: /redirect\r\n'
                    'Content-Length: 10\r\n'
                    'Date: .+\r\n'
                    'Server: .+\r\n\r\n'
                    '302: Found', txt)


def test_HTTPFound_empty_location() -> None:
    with pytest.raises(ValueError):
        web.HTTPFound(location='')

    with pytest.raises(ValueError):
        web.HTTPFound(location=None)


async def test_HTTPMethodNotAllowed(buf, http_request) -> None:
    resp = web.HTTPMethodNotAllowed('get', ['POST', 'PUT'])
    assert 'GET' == resp.method
    assert {'POST', 'PUT'} == resp.allowed_methods
    assert 'POST,PUT' == resp.headers['allow']
    await resp.prepare(http_request)
    await resp.write_eof()
    txt = buf.decode('utf8')
    assert re.match('HTTP/1.1 405 Method Not Allowed\r\n'
                    'Content-Type: text/plain; charset=utf-8\r\n'
                    'Allow: POST,PUT\r\n'
                    'Content-Length: 23\r\n'
                    'Date: .+\r\n'
                    'Server: .+\r\n\r\n'
                    '405: Method Not Allowed', txt)


def test_override_body_with_text() -> None:
    resp = web.HTTPNotFound(text="Page not found")
    assert 404 == resp.status
    assert "Page not found".encode('utf-8') == resp.body
    assert "Page not found" == resp.text
    assert "text/plain" == resp.content_type
    assert "utf-8" == resp.charset


def test_override_body_with_binary() -> None:
    txt = "<html><body>Page not found</body></html>"
    with pytest.warns(DeprecationWarning):
        resp = web.HTTPNotFound(body=txt.encode('utf-8'),
                                content_type="text/html")
    assert 404 == resp.status
    assert txt.encode('utf-8') == resp.body
    assert txt == resp.text
    assert "text/html" == resp.content_type
    assert resp.charset is None


def test_default_body() -> None:
    resp = web.HTTPOk()
    assert b'200: OK' == resp.body


def test_empty_body_204() -> None:
    resp = web.HTTPNoContent()
    assert resp.body is None


def test_empty_body_205() -> None:
    resp = web.HTTPNoContent()
    assert resp.body is None


def test_empty_body_304() -> None:
    resp = web.HTTPNoContent()
    resp.body is None


def test_link_header_451(buf) -> None:
    resp = web.HTTPUnavailableForLegalReasons(link='http://warning.or.kr/')

    assert 'http://warning.or.kr/' == resp.link
    assert '<http://warning.or.kr/>; rel="blocked-by"' == resp.headers['Link']


def test_HTTPException_retains_cause() -> None:
    with pytest.raises(web.HTTPException) as ei:
        try:
            raise Exception('CustomException')
        except Exception as exc:
            raise web.HTTPException() from exc
    tb = ''.join(format_exception(ei.type, ei.value, ei.tb))
    assert 'CustomException' in tb
    assert 'direct cause' in tb