| from http import HTTPStatus |
|
|
| import pytest |
|
|
| from .. import _events |
| from .._events import * |
| from .._util import LocalProtocolError |
|
|
|
|
| def test_event_bundle(): |
| class T(_events._EventBundle): |
| _fields = ["a", "b"] |
| _defaults = {"b": 1} |
|
|
| def _validate(self): |
| if self.a == 0: |
| raise ValueError |
|
|
| |
| t = T(a=1, b=0) |
| assert repr(t) == "T(a=1, b=0)" |
| assert t == T(a=1, b=0) |
| assert not (t == T(a=2, b=0)) |
| assert not (t != T(a=1, b=0)) |
| assert t != T(a=2, b=0) |
| with pytest.raises(TypeError): |
| hash(t) |
|
|
| |
| t = T(a=10) |
| assert t.a == 10 |
| assert t.b == 1 |
|
|
| |
| with pytest.raises(TypeError): |
| T(1) |
|
|
| with pytest.raises(TypeError): |
| T(1, a=1, b=0) |
|
|
| |
| with pytest.raises(TypeError): |
| T(a=1, b=0, c=10) |
|
|
| |
| with pytest.raises(TypeError) as exc: |
| T(b=0) |
| |
| assert "kwarg a" in str(exc.value) |
|
|
| |
| with pytest.raises(ValueError): |
| T(a=0, b=0) |
|
|
|
|
| def test_events(): |
| with pytest.raises(LocalProtocolError): |
| |
| req = Request( |
| method="GET", target="/", headers=[("a", "b")], http_version="1.1" |
| ) |
| |
| req = Request(method="GET", target="/", headers=[("a", "b")], http_version="1.0") |
| |
| assert req.method == b"GET" |
| assert req.target == b"/" |
| assert req.headers == [(b"a", b"b")] |
| assert req.http_version == b"1.0" |
|
|
| |
| req = Request( |
| method="GET", |
| target="/", |
| headers=[("a", "b"), ("hOSt", "example.com")], |
| http_version="1.1", |
| ) |
| |
| assert req.headers == [(b"a", b"b"), (b"host", b"example.com")] |
|
|
| |
| with pytest.raises(LocalProtocolError): |
| req = Request( |
| method="GET", |
| target="/", |
| headers=[("Host", "a"), ("Host", "a")], |
| http_version="1.1", |
| ) |
| |
| with pytest.raises(LocalProtocolError): |
| req = Request( |
| method="GET", |
| target="/", |
| headers=[("Host", "a"), ("Host", "a")], |
| http_version="1.0", |
| ) |
|
|
| |
| for bad_char in "\x00\r\n\f\v": |
| with pytest.raises(LocalProtocolError): |
| req = Request( |
| method="GET", |
| target="/", |
| headers=[("Host", "a"), ("Foo", "asd" + bad_char)], |
| http_version="1.0", |
| ) |
|
|
| |
| |
| Request( |
| method="GET", |
| target="/", |
| headers=[("Host", "a"), ("Foo", "asd\x01\x02\x7f")], |
| http_version="1.0", |
| ) |
|
|
| |
| for bad_char in b"\x00\x20\x7f\xee": |
| target = bytearray(b"/") |
| target.append(bad_char) |
| with pytest.raises(LocalProtocolError): |
| Request( |
| method="GET", target=target, headers=[("Host", "a")], http_version="1.1" |
| ) |
|
|
| ir = InformationalResponse(status_code=100, headers=[("Host", "a")]) |
| assert ir.status_code == 100 |
| assert ir.headers == [(b"host", b"a")] |
| assert ir.http_version == b"1.1" |
|
|
| with pytest.raises(LocalProtocolError): |
| InformationalResponse(status_code=200, headers=[("Host", "a")]) |
|
|
| resp = Response(status_code=204, headers=[], http_version="1.0") |
| assert resp.status_code == 204 |
| assert resp.headers == [] |
| assert resp.http_version == b"1.0" |
|
|
| with pytest.raises(LocalProtocolError): |
| resp = Response(status_code=100, headers=[], http_version="1.0") |
|
|
| with pytest.raises(LocalProtocolError): |
| Response(status_code="100", headers=[], http_version="1.0") |
|
|
| with pytest.raises(LocalProtocolError): |
| InformationalResponse(status_code=b"100", headers=[], http_version="1.0") |
|
|
| d = Data(data=b"asdf") |
| assert d.data == b"asdf" |
|
|
| eom = EndOfMessage() |
| assert eom.headers == [] |
|
|
| cc = ConnectionClosed() |
| assert repr(cc) == "ConnectionClosed()" |
|
|
|
|
| def test_intenum_status_code(): |
| |
|
|
| r = Response(status_code=HTTPStatus.OK, headers=[], http_version="1.0") |
| assert r.status_code == HTTPStatus.OK |
| assert type(r.status_code) is not type(HTTPStatus.OK) |
| assert type(r.status_code) is int |
|
|
|
|
| def test_header_casing(): |
| r = Request( |
| method="GET", |
| target="/", |
| headers=[("Host", "example.org"), ("Connection", "keep-alive")], |
| http_version="1.1", |
| ) |
| assert len(r.headers) == 2 |
| assert r.headers[0] == (b"host", b"example.org") |
| assert r.headers == [(b"host", b"example.org"), (b"connection", b"keep-alive")] |
| assert r.headers.raw_items() == [ |
| (b"Host", b"example.org"), |
| (b"Connection", b"keep-alive"), |
| ] |
|
|