aboutsummaryrefslogtreecommitdiffstats
path: root/test/netlib/http
diff options
context:
space:
mode:
Diffstat (limited to 'test/netlib/http')
-rw-r--r--test/netlib/http/http1/test_read.py13
-rw-r--r--test/netlib/http/http2/test_connections.py26
-rw-r--r--test/netlib/http/test_headers.py10
-rw-r--r--test/netlib/http/test_multipart.py24
-rw-r--r--test/netlib/http/test_response.py2
-rw-r--r--test/netlib/http/test_url.py66
6 files changed, 126 insertions, 15 deletions
diff --git a/test/netlib/http/http1/test_read.py b/test/netlib/http/http1/test_read.py
index 974aa895..5285ac1d 100644
--- a/test/netlib/http/http1/test_read.py
+++ b/test/netlib/http/http1/test_read.py
@@ -7,11 +7,22 @@ from netlib.http.http1.read import (
read_request, read_response, read_request_head,
read_response_head, read_body, connection_close, expected_http_body_size, _get_first_line,
_read_request_line, _parse_authority_form, _read_response_line, _check_http_version,
- _read_headers, _read_chunked
+ _read_headers, _read_chunked, get_header_tokens
)
from netlib.tutils import treq, tresp, raises
+def test_get_header_tokens():
+ headers = Headers()
+ assert get_header_tokens(headers, "foo") == []
+ headers["foo"] = "bar"
+ assert get_header_tokens(headers, "foo") == ["bar"]
+ headers["foo"] = "bar, voing"
+ assert get_header_tokens(headers, "foo") == ["bar", "voing"]
+ headers.set_all("foo", ["bar, voing", "oink"])
+ assert get_header_tokens(headers, "foo") == ["bar", "voing", "oink"]
+
+
def test_read_request():
rfile = BytesIO(b"GET / HTTP/1.1\r\n\r\nskip")
r = read_request(rfile)
diff --git a/test/netlib/http/http2/test_connections.py b/test/netlib/http/http2/test_connections.py
index 69667d1c..27cc30ba 100644
--- a/test/netlib/http/http2/test_connections.py
+++ b/test/netlib/http/http2/test_connections.py
@@ -1,12 +1,12 @@
import mock
import codecs
-from hyperframe import frame
-
-from netlib import tcp, http, utils
+import hyperframe
+from netlib import tcp, http
from netlib.tutils import raises
from netlib.exceptions import TcpDisconnect
from netlib.http.http2.connections import HTTP2Protocol, TCPHandler
+from netlib.http.http2 import framereader
from ... import tservers
@@ -111,11 +111,11 @@ class TestPerformServerConnectionPreface(tservers.ServerTestBase):
self.wfile.flush()
# check empty settings frame
- raw = utils.http2_read_raw_frame(self.rfile)
+ raw = framereader.http2_read_raw_frame(self.rfile)
assert raw == codecs.decode('00000c040000000000000200000000000300000001', 'hex_codec')
# check settings acknowledgement
- raw = utils.http2_read_raw_frame(self.rfile)
+ raw = framereader.http2_read_raw_frame(self.rfile)
assert raw == codecs.decode('000000040100000000', 'hex_codec')
# send settings acknowledgement
@@ -214,19 +214,19 @@ class TestApplySettings(tservers.ServerTestBase):
protocol = HTTP2Protocol(c)
protocol._apply_settings({
- frame.SettingsFrame.ENABLE_PUSH: 'foo',
- frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 'bar',
- frame.SettingsFrame.INITIAL_WINDOW_SIZE: 'deadbeef',
+ hyperframe.frame.SettingsFrame.ENABLE_PUSH: 'foo',
+ hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 'bar',
+ hyperframe.frame.SettingsFrame.INITIAL_WINDOW_SIZE: 'deadbeef',
})
assert c.rfile.safe_read(2) == b"OK"
assert protocol.http2_settings[
- frame.SettingsFrame.ENABLE_PUSH] == 'foo'
+ hyperframe.frame.SettingsFrame.ENABLE_PUSH] == 'foo'
assert protocol.http2_settings[
- frame.SettingsFrame.MAX_CONCURRENT_STREAMS] == 'bar'
+ hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS] == 'bar'
assert protocol.http2_settings[
- frame.SettingsFrame.INITIAL_WINDOW_SIZE] == 'deadbeef'
+ hyperframe.frame.SettingsFrame.INITIAL_WINDOW_SIZE] == 'deadbeef'
class TestCreateHeaders(object):
@@ -258,7 +258,7 @@ class TestCreateHeaders(object):
(b'server', b'version')])
protocol = HTTP2Protocol(self.c)
- protocol.http2_settings[frame.SettingsFrame.MAX_FRAME_SIZE] = 8
+ protocol.http2_settings[hyperframe.frame.SettingsFrame.MAX_FRAME_SIZE] = 8
bytes = protocol._create_headers(headers, 1, end_stream=True)
assert len(bytes) == 3
assert bytes[0] == codecs.decode('000008010100000001828487408294e783', 'hex_codec')
@@ -281,7 +281,7 @@ class TestCreateBody(object):
def test_create_body_multiple_frames(self):
protocol = HTTP2Protocol(self.c)
- protocol.http2_settings[frame.SettingsFrame.MAX_FRAME_SIZE] = 5
+ protocol.http2_settings[hyperframe.frame.SettingsFrame.MAX_FRAME_SIZE] = 5
bytes = protocol._create_body(b'foobarmehm42', 1)
assert len(bytes) == 3
assert bytes[0] == codecs.decode('000005000000000001666f6f6261', 'hex_codec')
diff --git a/test/netlib/http/test_headers.py b/test/netlib/http/test_headers.py
index cd2ca9d1..e12bceaf 100644
--- a/test/netlib/http/test_headers.py
+++ b/test/netlib/http/test_headers.py
@@ -1,4 +1,5 @@
from netlib.http import Headers
+from netlib.http.headers import parse_content_type
from netlib.tutils import raises
@@ -72,3 +73,12 @@ class TestHeaders(object):
replacements = headers.replace(r"Host: ", "X-Host ")
assert replacements == 0
assert headers["Host"] == "example.com"
+
+
+def test_parse_content_type():
+ p = parse_content_type
+ assert p("text/html") == ("text", "html", {})
+ assert p("text") is None
+
+ v = p("text/html; charset=UTF-8")
+ assert v == ('text', 'html', {'charset': 'UTF-8'})
diff --git a/test/netlib/http/test_multipart.py b/test/netlib/http/test_multipart.py
new file mode 100644
index 00000000..1d7e0062
--- /dev/null
+++ b/test/netlib/http/test_multipart.py
@@ -0,0 +1,24 @@
+from netlib.http import Headers
+from netlib.http import multipart
+
+
+def test_decode():
+ boundary = 'somefancyboundary'
+ headers = Headers(
+ content_type='multipart/form-data; boundary=' + boundary
+ )
+ content = (
+ "--{0}\n"
+ "Content-Disposition: form-data; name=\"field1\"\n\n"
+ "value1\n"
+ "--{0}\n"
+ "Content-Disposition: form-data; name=\"field2\"\n\n"
+ "value2\n"
+ "--{0}--".format(boundary).encode()
+ )
+
+ form = multipart.decode(headers, content)
+
+ assert len(form) == 2
+ assert form[0] == (b"field1", b"value1")
+ assert form[1] == (b"field2", b"value2")
diff --git a/test/netlib/http/test_response.py b/test/netlib/http/test_response.py
index 1faef7ec..b3c2f736 100644
--- a/test/netlib/http/test_response.py
+++ b/test/netlib/http/test_response.py
@@ -24,7 +24,7 @@ class TestResponseCore(object):
"""
def test_repr(self):
response = tresp()
- assert repr(response) == "Response(200 OK, unknown content type, 7B)"
+ assert repr(response) == "Response(200 OK, unknown content type, 7b)"
response.content = None
assert repr(response) == "Response(200 OK, no content)"
diff --git a/test/netlib/http/test_url.py b/test/netlib/http/test_url.py
new file mode 100644
index 00000000..26b37230
--- /dev/null
+++ b/test/netlib/http/test_url.py
@@ -0,0 +1,66 @@
+from netlib import tutils
+from netlib.http import url
+
+
+def test_parse():
+ with tutils.raises(ValueError):
+ url.parse("")
+
+ s, h, po, pa = url.parse(b"http://foo.com:8888/test")
+ assert s == b"http"
+ assert h == b"foo.com"
+ assert po == 8888
+ assert pa == b"/test"
+
+ s, h, po, pa = url.parse("http://foo/bar")
+ assert s == b"http"
+ assert h == b"foo"
+ assert po == 80
+ assert pa == b"/bar"
+
+ s, h, po, pa = url.parse(b"http://user:pass@foo/bar")
+ assert s == b"http"
+ assert h == b"foo"
+ assert po == 80
+ assert pa == b"/bar"
+
+ s, h, po, pa = url.parse(b"http://foo")
+ assert pa == b"/"
+
+ s, h, po, pa = url.parse(b"https://foo")
+ assert po == 443
+
+ with tutils.raises(ValueError):
+ url.parse(b"https://foo:bar")
+
+ # Invalid IDNA
+ with tutils.raises(ValueError):
+ url.parse("http://\xfafoo")
+ # Invalid PATH
+ with tutils.raises(ValueError):
+ url.parse("http:/\xc6/localhost:56121")
+ # Null byte in host
+ with tutils.raises(ValueError):
+ url.parse("http://foo\0")
+ # Port out of range
+ _, _, port, _ = url.parse("http://foo:999999")
+ assert port == 80
+ # Invalid IPv6 URL - see http://www.ietf.org/rfc/rfc2732.txt
+ with tutils.raises(ValueError):
+ url.parse('http://lo[calhost')
+
+
+def test_unparse():
+ assert url.unparse("http", "foo.com", 99, "") == "http://foo.com:99"
+ assert url.unparse("http", "foo.com", 80, "/bar") == "http://foo.com/bar"
+ assert url.unparse("https", "foo.com", 80, "") == "https://foo.com:80"
+ assert url.unparse("https", "foo.com", 443, "") == "https://foo.com"
+
+
+def test_urlencode():
+ assert url.encode([('foo', 'bar')])
+
+
+def test_urldecode():
+ s = "one=two&three=four"
+ assert len(url.decode(s)) == 2
generated by cgit v1.2.3 (git 2.25.1) at 2025-11-04 13:59:02 +0000