aboutsummaryrefslogtreecommitdiffstats
path: root/test/netlib
diff options
context:
space:
mode:
Diffstat (limited to 'test/netlib')
-rw-r--r--test/netlib/http/http1/test_read.py13
-rw-r--r--test/netlib/http/http2/test_connections.py31
-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
-rw-r--r--test/netlib/test_basetypes.py28
-rw-r--r--test/netlib/test_human.py36
-rw-r--r--test/netlib/test_utils.py141
9 files changed, 194 insertions, 157 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 ff462ba6..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')
@@ -312,7 +312,10 @@ class TestReadRequest(tservers.ServerTestBase):
req = protocol.read_request(NotImplemented)
assert req.stream_id
- assert req.headers.fields == ((b':method', b'GET'), (b':path', b'/'), (b':scheme', b'https'))
+ assert req.headers.fields == ()
+ assert req.method == "GET"
+ assert req.path == "/"
+ assert req.scheme == "https"
assert req.content == b'foobar'
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
diff --git a/test/netlib/test_basetypes.py b/test/netlib/test_basetypes.py
new file mode 100644
index 00000000..aa415784
--- /dev/null
+++ b/test/netlib/test_basetypes.py
@@ -0,0 +1,28 @@
+from netlib import basetypes
+
+
+class SerializableDummy(basetypes.Serializable):
+ def __init__(self, i):
+ self.i = i
+
+ def get_state(self):
+ return self.i
+
+ def set_state(self, i):
+ self.i = i
+
+ def from_state(self, state):
+ return type(self)(state)
+
+
+class TestSerializable:
+
+ def test_copy(self):
+ a = SerializableDummy(42)
+ assert a.i == 42
+ b = a.copy()
+ assert b.i == 42
+
+ a.set_state(1)
+ assert a.i == 1
+ assert b.i == 42
diff --git a/test/netlib/test_human.py b/test/netlib/test_human.py
new file mode 100644
index 00000000..2a5c2a85
--- /dev/null
+++ b/test/netlib/test_human.py
@@ -0,0 +1,36 @@
+from netlib import human, tutils
+
+
+def test_parse_size():
+ assert human.parse_size("0") == 0
+ assert human.parse_size("0b") == 0
+ assert human.parse_size("1") == 1
+ assert human.parse_size("1k") == 1024
+ assert human.parse_size("1m") == 1024**2
+ assert human.parse_size("1g") == 1024**3
+ tutils.raises(ValueError, human.parse_size, "1f")
+ tutils.raises(ValueError, human.parse_size, "ak")
+
+
+def test_pretty_size():
+ assert human.pretty_size(0) == "0b"
+ assert human.pretty_size(100) == "100b"
+ assert human.pretty_size(1024) == "1k"
+ assert human.pretty_size(1024 + (1024 / 2.0)) == "1.5k"
+ assert human.pretty_size(1024 * 1024) == "1m"
+ assert human.pretty_size(10 * 1024 * 1024) == "10m"
+
+
+def test_pretty_duration():
+ assert human.pretty_duration(0.00001) == "0ms"
+ assert human.pretty_duration(0.0001) == "0ms"
+ assert human.pretty_duration(0.001) == "1ms"
+ assert human.pretty_duration(0.01) == "10ms"
+ assert human.pretty_duration(0.1) == "100ms"
+ assert human.pretty_duration(1) == "1.00s"
+ assert human.pretty_duration(10) == "10.0s"
+ assert human.pretty_duration(100) == "100s"
+ assert human.pretty_duration(1000) == "1000s"
+ assert human.pretty_duration(10000) == "10000s"
+ assert human.pretty_duration(1.123) == "1.12s"
+ assert human.pretty_duration(0.123) == "123ms"
diff --git a/test/netlib/test_utils.py b/test/netlib/test_utils.py
index e4c81a48..e13029cb 100644
--- a/test/netlib/test_utils.py
+++ b/test/netlib/test_utils.py
@@ -1,7 +1,6 @@
# coding=utf-8
from netlib import utils, tutils
-from netlib.http import Headers
def test_bidi():
@@ -31,146 +30,6 @@ def test_clean_bin():
assert utils.clean_bin(u"\u2605") == u"\u2605"
-def test_pretty_size():
- assert utils.pretty_size(100) == "100B"
- assert utils.pretty_size(1024) == "1kB"
- assert utils.pretty_size(1024 + (1024 / 2.0)) == "1.5kB"
- assert utils.pretty_size(1024 * 1024) == "1MB"
-
-
-def test_parse_url():
- with tutils.raises(ValueError):
- utils.parse_url("")
-
- s, h, po, pa = utils.parse_url(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 = utils.parse_url("http://foo/bar")
- assert s == b"http"
- assert h == b"foo"
- assert po == 80
- assert pa == b"/bar"
-
- s, h, po, pa = utils.parse_url(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 = utils.parse_url(b"http://foo")
- assert pa == b"/"
-
- s, h, po, pa = utils.parse_url(b"https://foo")
- assert po == 443
-
- with tutils.raises(ValueError):
- utils.parse_url(b"https://foo:bar")
-
- # Invalid IDNA
- with tutils.raises(ValueError):
- utils.parse_url("http://\xfafoo")
- # Invalid PATH
- with tutils.raises(ValueError):
- utils.parse_url("http:/\xc6/localhost:56121")
- # Null byte in host
- with tutils.raises(ValueError):
- utils.parse_url("http://foo\0")
- # Port out of range
- _, _, port, _ = utils.parse_url("http://foo:999999")
- assert port == 80
- # Invalid IPv6 URL - see http://www.ietf.org/rfc/rfc2732.txt
- with tutils.raises(ValueError):
- utils.parse_url('http://lo[calhost')
-
-
-def test_unparse_url():
- assert utils.unparse_url("http", "foo.com", 99, "") == "http://foo.com:99"
- assert utils.unparse_url("http", "foo.com", 80, "/bar") == "http://foo.com/bar"
- assert utils.unparse_url("https", "foo.com", 80, "") == "https://foo.com:80"
- assert utils.unparse_url("https", "foo.com", 443, "") == "https://foo.com"
-
-
-def test_urlencode():
- assert utils.urlencode([('foo', 'bar')])
-
-
-def test_urldecode():
- s = "one=two&three=four"
- assert len(utils.urldecode(s)) == 2
-
-
-def test_get_header_tokens():
- headers = Headers()
- assert utils.get_header_tokens(headers, "foo") == []
- headers["foo"] = "bar"
- assert utils.get_header_tokens(headers, "foo") == ["bar"]
- headers["foo"] = "bar, voing"
- assert utils.get_header_tokens(headers, "foo") == ["bar", "voing"]
- headers.set_all("foo", ["bar, voing", "oink"])
- assert utils.get_header_tokens(headers, "foo") == ["bar", "voing", "oink"]
-
-
-def test_multipartdecode():
- 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 = utils.multipartdecode(headers, content)
-
- assert len(form) == 2
- assert form[0] == (b"field1", b"value1")
- assert form[1] == (b"field2", b"value2")
-
-
-def test_parse_content_type():
- p = utils.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'})
-
-
-class SerializableDummy(utils.Serializable):
- def __init__(self, i):
- self.i = i
-
- def get_state(self):
- return self.i
-
- def set_state(self, i):
- self.i = i
-
- def from_state(self, state):
- return type(self)(state)
-
-
-class TestSerializable:
-
- def test_copy(self):
- a = SerializableDummy(42)
- assert a.i == 42
- b = a.copy()
- assert b.i == 42
-
- a.set_state(1)
- assert a.i == 1
- assert b.i == 42
-
-
def test_safe_subn():
assert utils.safe_subn("foo", u"bar", "\xc2foo")
generated by cgit v1.2.3 (git 2.25.1) at 2024-08-09 20:00:39 +0000