aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/h2/test_frames.py213
-rw-r--r--test/test_certutils.py18
-rw-r--r--test/test_http_auth.py5
-rw-r--r--test/test_socks.py7
-rw-r--r--test/test_tcp.py17
-rw-r--r--test/test_websockets.py3
6 files changed, 211 insertions, 52 deletions
diff --git a/test/h2/test_frames.py b/test/h2/test_frames.py
index eb470dd4..313ef405 100644
--- a/test/h2/test_frames.py
+++ b/test/h2/test_frames.py
@@ -7,7 +7,12 @@ from nose.tools import assert_equal
def test_invalid_flags():
- tutils.raises(ValueError, DataFrame, ContinuationFrame.FLAG_END_HEADERS, 0x1234567, 'foobar')
+ tutils.raises(
+ ValueError,
+ DataFrame,
+ ContinuationFrame.FLAG_END_HEADERS,
+ 0x1234567,
+ 'foobar')
def test_frame_equality():
@@ -24,8 +29,15 @@ def test_data_frame_to_bytes():
f = DataFrame(6, Frame.FLAG_END_STREAM, 0x1234567, 'foobar')
assert_equal(f.to_bytes().encode('hex'), '000006000101234567666f6f626172')
- f = DataFrame(11, Frame.FLAG_END_STREAM | Frame.FLAG_PADDED, 0x1234567, 'foobar', pad_length=3)
- assert_equal(f.to_bytes().encode('hex'), '00000a00090123456703666f6f626172000000')
+ f = DataFrame(
+ 11,
+ Frame.FLAG_END_STREAM | Frame.FLAG_PADDED,
+ 0x1234567,
+ 'foobar',
+ pad_length=3)
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00000a00090123456703666f6f626172000000')
f = DataFrame(6, Frame.FLAG_NO_FLAGS, 0x0, 'foobar')
tutils.raises(ValueError, f.to_bytes)
@@ -50,7 +62,12 @@ def test_data_frame_from_bytes():
def test_data_frame_human_readable():
- f = DataFrame(11, Frame.FLAG_END_STREAM | Frame.FLAG_PADDED, 0x1234567, 'foobar', pad_length=3)
+ f = DataFrame(
+ 11,
+ Frame.FLAG_END_STREAM | Frame.FLAG_PADDED,
+ 0x1234567,
+ 'foobar',
+ pad_length=3)
assert f.human_readable()
@@ -68,7 +85,9 @@ def test_headers_frame_to_bytes():
0x1234567,
headers=[('host', 'foo.bar')],
pad_length=3)
- assert_equal(f.to_bytes().encode('hex'), '00000b01080123456703668594e75e31d9000000')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00000b01080123456703668594e75e31d9000000')
f = HeadersFrame(
10,
@@ -78,7 +97,9 @@ def test_headers_frame_to_bytes():
exclusive=True,
stream_dependency=0x7654321,
weight=42)
- assert_equal(f.to_bytes().encode('hex'), '00000c012001234567876543212a668594e75e31d9')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00000c012001234567876543212a668594e75e31d9')
f = HeadersFrame(
14,
@@ -89,7 +110,9 @@ def test_headers_frame_to_bytes():
exclusive=True,
stream_dependency=0x7654321,
weight=42)
- assert_equal(f.to_bytes().encode('hex'), '00001001280123456703876543212a668594e75e31d9000000')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00001001280123456703876543212a668594e75e31d9000000')
f = HeadersFrame(
14,
@@ -100,7 +123,9 @@ def test_headers_frame_to_bytes():
exclusive=False,
stream_dependency=0x7654321,
weight=42)
- assert_equal(f.to_bytes().encode('hex'), '00001001280123456703076543212a668594e75e31d9000000')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00001001280123456703076543212a668594e75e31d9000000')
f = HeadersFrame(6, Frame.FLAG_NO_FLAGS, 0x0, 'foobar')
tutils.raises(ValueError, f.to_bytes)
@@ -115,7 +140,8 @@ def test_headers_frame_from_bytes():
assert_equal(f.stream_id, 0x1234567)
assert_equal(f.headers, [('host', 'foo.bar')])
- f = Frame.from_bytes('00000b01080123456703668594e75e31d9000000'.decode('hex'))
+ f = Frame.from_bytes(
+ '00000b01080123456703668594e75e31d9000000'.decode('hex'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 11)
assert_equal(f.TYPE, HeadersFrame.TYPE)
@@ -123,7 +149,8 @@ def test_headers_frame_from_bytes():
assert_equal(f.stream_id, 0x1234567)
assert_equal(f.headers, [('host', 'foo.bar')])
- f = Frame.from_bytes('00000c012001234567876543212a668594e75e31d9'.decode('hex'))
+ f = Frame.from_bytes(
+ '00000c012001234567876543212a668594e75e31d9'.decode('hex'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 12)
assert_equal(f.TYPE, HeadersFrame.TYPE)
@@ -134,7 +161,8 @@ def test_headers_frame_from_bytes():
assert_equal(f.stream_dependency, 0x7654321)
assert_equal(f.weight, 42)
- f = Frame.from_bytes('00001001280123456703876543212a668594e75e31d9000000'.decode('hex'))
+ f = Frame.from_bytes(
+ '00001001280123456703876543212a668594e75e31d9000000'.decode('hex'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 16)
assert_equal(f.TYPE, HeadersFrame.TYPE)
@@ -145,7 +173,8 @@ def test_headers_frame_from_bytes():
assert_equal(f.stream_dependency, 0x7654321)
assert_equal(f.weight, 42)
- f = Frame.from_bytes('00001001280123456703076543212a668594e75e31d9000000'.decode('hex'))
+ f = Frame.from_bytes(
+ '00001001280123456703076543212a668594e75e31d9000000'.decode('hex'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 16)
assert_equal(f.TYPE, HeadersFrame.TYPE)
@@ -182,10 +211,22 @@ def test_headers_frame_human_readable():
def test_priority_frame_to_bytes():
- f = PriorityFrame(5, Frame.FLAG_NO_FLAGS, 0x1234567, exclusive=True, stream_dependency=0x7654321, weight=42)
+ f = PriorityFrame(
+ 5,
+ Frame.FLAG_NO_FLAGS,
+ 0x1234567,
+ exclusive=True,
+ stream_dependency=0x7654321,
+ weight=42)
assert_equal(f.to_bytes().encode('hex'), '000005020001234567876543212a')
- f = PriorityFrame(5, Frame.FLAG_NO_FLAGS, 0x1234567, exclusive=False, stream_dependency=0x7654321, weight=21)
+ f = PriorityFrame(
+ 5,
+ Frame.FLAG_NO_FLAGS,
+ 0x1234567,
+ exclusive=False,
+ stream_dependency=0x7654321,
+ weight=21)
assert_equal(f.to_bytes().encode('hex'), '0000050200012345670765432115')
f = PriorityFrame(5, Frame.FLAG_NO_FLAGS, 0x0, stream_dependency=0x1234567)
@@ -218,7 +259,13 @@ def test_priority_frame_from_bytes():
def test_priority_frame_human_readable():
- f = PriorityFrame(5, Frame.FLAG_NO_FLAGS, 0x1234567, exclusive=False, stream_dependency=0x7654321, weight=21)
+ f = PriorityFrame(
+ 5,
+ Frame.FLAG_NO_FLAGS,
+ 0x1234567,
+ exclusive=False,
+ stream_dependency=0x7654321,
+ weight=21)
assert f.human_readable()
@@ -266,7 +313,9 @@ def test_settings_frame_to_bytes():
settings={
SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH: 1,
SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS: 0x12345678})
- assert_equal(f.to_bytes().encode('hex'), '00000c040000000000000200000001000312345678')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00000c040000000000000200000001000312345678')
f = SettingsFrame(0, Frame.FLAG_NO_FLAGS, 0x1234567)
tutils.raises(ValueError, f.to_bytes)
@@ -296,7 +345,8 @@ def test_settings_frame_from_bytes():
assert_equal(len(f.settings), 1)
assert_equal(f.settings[SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH], 1)
- f = Frame.from_bytes('00000c040000000000000200000001000312345678'.decode('hex'))
+ f = Frame.from_bytes(
+ '00000c040000000000000200000001000312345678'.decode('hex'))
assert isinstance(f, SettingsFrame)
assert_equal(f.length, 12)
assert_equal(f.TYPE, SettingsFrame.TYPE)
@@ -304,7 +354,10 @@ def test_settings_frame_from_bytes():
assert_equal(f.stream_id, 0x0)
assert_equal(len(f.settings), 2)
assert_equal(f.settings[SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH], 1)
- assert_equal(f.settings[SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS], 0x12345678)
+ assert_equal(
+ f.settings[
+ SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS],
+ 0x12345678)
def test_settings_frame_human_readable():
@@ -322,11 +375,26 @@ def test_settings_frame_human_readable():
def test_push_promise_frame_to_bytes():
- f = PushPromiseFrame(10, Frame.FLAG_NO_FLAGS, 0x1234567, 0x7654321, 'foobar')
- assert_equal(f.to_bytes().encode('hex'), '00000a05000123456707654321666f6f626172')
+ f = PushPromiseFrame(
+ 10,
+ Frame.FLAG_NO_FLAGS,
+ 0x1234567,
+ 0x7654321,
+ 'foobar')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00000a05000123456707654321666f6f626172')
- f = PushPromiseFrame(14, HeadersFrame.FLAG_PADDED, 0x1234567, 0x7654321, 'foobar', pad_length=3)
- assert_equal(f.to_bytes().encode('hex'), '00000e0508012345670307654321666f6f626172000000')
+ f = PushPromiseFrame(
+ 14,
+ HeadersFrame.FLAG_PADDED,
+ 0x1234567,
+ 0x7654321,
+ 'foobar',
+ pad_length=3)
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00000e0508012345670307654321666f6f626172000000')
f = PushPromiseFrame(4, Frame.FLAG_NO_FLAGS, 0x0, 0x1234567)
tutils.raises(ValueError, f.to_bytes)
@@ -344,7 +412,8 @@ def test_push_promise_frame_from_bytes():
assert_equal(f.stream_id, 0x1234567)
assert_equal(f.header_block_fragment, 'foobar')
- f = Frame.from_bytes('00000e0508012345670307654321666f6f626172000000'.decode('hex'))
+ f = Frame.from_bytes(
+ '00000e0508012345670307654321666f6f626172000000'.decode('hex'))
assert isinstance(f, PushPromiseFrame)
assert_equal(f.length, 14)
assert_equal(f.TYPE, PushPromiseFrame.TYPE)
@@ -354,16 +423,26 @@ def test_push_promise_frame_from_bytes():
def test_push_promise_frame_human_readable():
- f = PushPromiseFrame(14, HeadersFrame.FLAG_PADDED, 0x1234567, 0x7654321, 'foobar', pad_length=3)
+ f = PushPromiseFrame(
+ 14,
+ HeadersFrame.FLAG_PADDED,
+ 0x1234567,
+ 0x7654321,
+ 'foobar',
+ pad_length=3)
assert f.human_readable()
def test_ping_frame_to_bytes():
f = PingFrame(8, PingFrame.FLAG_ACK, 0x0, payload=b'foobar')
- assert_equal(f.to_bytes().encode('hex'), '000008060100000000666f6f6261720000')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '000008060100000000666f6f6261720000')
f = PingFrame(8, Frame.FLAG_NO_FLAGS, 0x0, payload=b'foobardeadbeef')
- assert_equal(f.to_bytes().encode('hex'), '000008060000000000666f6f6261726465')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '000008060000000000666f6f6261726465')
f = PingFrame(8, Frame.FLAG_NO_FLAGS, 0x1234567)
tutils.raises(ValueError, f.to_bytes)
@@ -393,13 +472,34 @@ def test_ping_frame_human_readable():
def test_goaway_frame_to_bytes():
- f = GoAwayFrame(8, Frame.FLAG_NO_FLAGS, 0x0, last_stream=0x1234567, error_code=0x87654321, data=b'')
- assert_equal(f.to_bytes().encode('hex'), '0000080700000000000123456787654321')
-
- f = GoAwayFrame(14, Frame.FLAG_NO_FLAGS, 0x0, last_stream=0x1234567, error_code=0x87654321, data=b'foobar')
- assert_equal(f.to_bytes().encode('hex'), '00000e0700000000000123456787654321666f6f626172')
-
- f = GoAwayFrame(8, Frame.FLAG_NO_FLAGS, 0x1234567, last_stream=0x1234567, error_code=0x87654321)
+ f = GoAwayFrame(
+ 8,
+ Frame.FLAG_NO_FLAGS,
+ 0x0,
+ last_stream=0x1234567,
+ error_code=0x87654321,
+ data=b'')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '0000080700000000000123456787654321')
+
+ f = GoAwayFrame(
+ 14,
+ Frame.FLAG_NO_FLAGS,
+ 0x0,
+ last_stream=0x1234567,
+ error_code=0x87654321,
+ data=b'foobar')
+ assert_equal(
+ f.to_bytes().encode('hex'),
+ '00000e0700000000000123456787654321666f6f626172')
+
+ f = GoAwayFrame(
+ 8,
+ Frame.FLAG_NO_FLAGS,
+ 0x1234567,
+ last_stream=0x1234567,
+ error_code=0x87654321)
tutils.raises(ValueError, f.to_bytes)
@@ -414,7 +514,8 @@ def test_goaway_frame_from_bytes():
assert_equal(f.error_code, 0x87654321)
assert_equal(f.data, b'')
- f = Frame.from_bytes('00000e0700000000000123456787654321666f6f626172'.decode('hex'))
+ f = Frame.from_bytes(
+ '00000e0700000000000123456787654321666f6f626172'.decode('hex'))
assert isinstance(f, GoAwayFrame)
assert_equal(f.length, 14)
assert_equal(f.TYPE, GoAwayFrame.TYPE)
@@ -426,18 +527,36 @@ def test_goaway_frame_from_bytes():
def test_go_away_frame_human_readable():
- f = GoAwayFrame(14, Frame.FLAG_NO_FLAGS, 0x0, last_stream=0x1234567, error_code=0x87654321, data=b'foobar')
+ f = GoAwayFrame(
+ 14,
+ Frame.FLAG_NO_FLAGS,
+ 0x0,
+ last_stream=0x1234567,
+ error_code=0x87654321,
+ data=b'foobar')
assert f.human_readable()
def test_window_update_frame_to_bytes():
- f = WindowUpdateFrame(4, Frame.FLAG_NO_FLAGS, 0x0, window_size_increment=0x1234567)
+ f = WindowUpdateFrame(
+ 4,
+ Frame.FLAG_NO_FLAGS,
+ 0x0,
+ window_size_increment=0x1234567)
assert_equal(f.to_bytes().encode('hex'), '00000408000000000001234567')
- f = WindowUpdateFrame(4, Frame.FLAG_NO_FLAGS, 0x1234567, window_size_increment=0x7654321)
+ f = WindowUpdateFrame(
+ 4,
+ Frame.FLAG_NO_FLAGS,
+ 0x1234567,
+ window_size_increment=0x7654321)
assert_equal(f.to_bytes().encode('hex'), '00000408000123456707654321')
- f = WindowUpdateFrame(4, Frame.FLAG_NO_FLAGS, 0x0, window_size_increment=0xdeadbeef)
+ f = WindowUpdateFrame(
+ 4,
+ Frame.FLAG_NO_FLAGS,
+ 0x0,
+ window_size_increment=0xdeadbeef)
tutils.raises(ValueError, f.to_bytes)
f = WindowUpdateFrame(4, Frame.FLAG_NO_FLAGS, 0x0, window_size_increment=0)
@@ -455,12 +574,20 @@ def test_window_update_frame_from_bytes():
def test_window_update_frame_human_readable():
- f = WindowUpdateFrame(4, Frame.FLAG_NO_FLAGS, 0x1234567, window_size_increment=0x7654321)
+ f = WindowUpdateFrame(
+ 4,
+ Frame.FLAG_NO_FLAGS,
+ 0x1234567,
+ window_size_increment=0x7654321)
assert f.human_readable()
def test_continuation_frame_to_bytes():
- f = ContinuationFrame(6, ContinuationFrame.FLAG_END_HEADERS, 0x1234567, 'foobar')
+ f = ContinuationFrame(
+ 6,
+ ContinuationFrame.FLAG_END_HEADERS,
+ 0x1234567,
+ 'foobar')
assert_equal(f.to_bytes().encode('hex'), '000006090401234567666f6f626172')
f = ContinuationFrame(6, ContinuationFrame.FLAG_END_HEADERS, 0x0, 'foobar')
@@ -478,5 +605,9 @@ def test_continuation_frame_from_bytes():
def test_continuation_frame_human_readable():
- f = ContinuationFrame(6, ContinuationFrame.FLAG_END_HEADERS, 0x1234567, 'foobar')
+ f = ContinuationFrame(
+ 6,
+ ContinuationFrame.FLAG_END_HEADERS,
+ 0x1234567,
+ 'foobar')
assert f.human_readable()
diff --git a/test/test_certutils.py b/test/test_certutils.py
index 115cac4d..e079ec40 100644
--- a/test/test_certutils.py
+++ b/test/test_certutils.py
@@ -42,7 +42,8 @@ class TestCertStore:
ca2 = certutils.CertStore.from_store(d, "test")
assert ca2.get_cert("foo", [])
- assert ca.default_ca.get_serial_number() == ca2.default_ca.get_serial_number()
+ assert ca.default_ca.get_serial_number(
+ ) == ca2.default_ca.get_serial_number()
def test_create_tmp(self):
with tutils.tmpdir() as d:
@@ -78,7 +79,8 @@ class TestCertStore:
with tutils.tmpdir() as d:
ca1 = certutils.CertStore.from_store(os.path.join(d, "ca1"), "test")
ca2 = certutils.CertStore.from_store(os.path.join(d, "ca2"), "test")
- assert not ca1.default_ca.get_serial_number() == ca2.default_ca.get_serial_number()
+ assert not ca1.default_ca.get_serial_number(
+ ) == ca2.default_ca.get_serial_number()
dc = ca2.get_cert("foo.com", ["sans.example.com"])
dcp = os.path.join(d, "dc")
@@ -93,8 +95,16 @@ class TestCertStore:
def test_gen_pkey(self):
try:
with tutils.tmpdir() as d:
- ca1 = certutils.CertStore.from_store(os.path.join(d, "ca1"), "test")
- ca2 = certutils.CertStore.from_store(os.path.join(d, "ca2"), "test")
+ ca1 = certutils.CertStore.from_store(
+ os.path.join(
+ d,
+ "ca1"),
+ "test")
+ ca2 = certutils.CertStore.from_store(
+ os.path.join(
+ d,
+ "ca2"),
+ "test")
cert = ca1.get_cert("foo.com", [])
assert certffi.get_flags(ca2.gen_pkey(cert[0])) == 1
finally:
diff --git a/test/test_http_auth.py b/test/test_http_auth.py
index 045fb13e..c842925b 100644
--- a/test/test_http_auth.py
+++ b/test/test_http_auth.py
@@ -13,7 +13,10 @@ class TestPassManNonAnon:
class TestPassManHtpasswd:
def test_file_errors(self):
- tutils.raises("malformed htpasswd file", http_auth.PassManHtpasswd, tutils.test_data.path("data/server.crt"))
+ tutils.raises(
+ "malformed htpasswd file",
+ http_auth.PassManHtpasswd,
+ tutils.test_data.path("data/server.crt"))
def test_simple(self):
pm = http_auth.PassManHtpasswd(tutils.test_data.path("data/htpasswd"))
diff --git a/test/test_socks.py b/test/test_socks.py
index a596dedf..a9db4706 100644
--- a/test/test_socks.py
+++ b/test/test_socks.py
@@ -61,7 +61,12 @@ def test_message_ipv6():
# Test ATYP=0x04 (IPV6)
ipv6_addr = "2001:db8:85a3:8d3:1319:8a2e:370:7344"
- raw = tutils.treader("\x05\x01\x00\x04" + socket.inet_pton(socket.AF_INET6, ipv6_addr) + "\xDE\xAD\xBE\xEF")
+ raw = tutils.treader(
+ "\x05\x01\x00\x04" +
+ socket.inet_pton(
+ socket.AF_INET6,
+ ipv6_addr) +
+ "\xDE\xAD\xBE\xEF")
out = StringIO()
msg = socks.Message.from_file(raw)
assert raw.read(2) == "\xBE\xEF"
diff --git a/test/test_tcp.py b/test/test_tcp.py
index 62617707..14ba555d 100644
--- a/test/test_tcp.py
+++ b/test/test_tcp.py
@@ -75,7 +75,9 @@ class TestServerBind(test.ServerTestBase):
for i in range(20):
random_port = random.randrange(1024, 65535)
try:
- c = tcp.TCPClient(("127.0.0.1", self.port), source_address=("127.0.0.1", random_port))
+ c = tcp.TCPClient(
+ ("127.0.0.1", self.port), source_address=(
+ "127.0.0.1", random_port))
c.connect()
assert c.rfile.readline() == str(("127.0.0.1", random_port))
return
@@ -196,7 +198,8 @@ class TestSSLClientCert(test.ServerTestBase):
def test_clientcert(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
- c.convert_to_ssl(cert=tutils.test_data.path("data/clientcert/client.pem"))
+ c.convert_to_ssl(
+ cert=tutils.test_data.path("data/clientcert/client.pem"))
assert c.rfile.readline().strip() == "1"
def test_clientcert_err(self):
@@ -305,7 +308,11 @@ class TestClientCipherListError(test.ServerTestBase):
def test_echo(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
- tutils.raises("cipher specification", c.convert_to_ssl, sni="foo.com", cipher_list="bogus")
+ tutils.raises(
+ "cipher specification",
+ c.convert_to_ssl,
+ sni="foo.com",
+ cipher_list="bogus")
class TestSSLDisconnect(test.ServerTestBase):
@@ -666,5 +673,7 @@ class TestSSLKeyLogger(test.ServerTestBase):
tcp.log_ssl_key = _logfun
def test_create_logfun(self):
- assert isinstance(tcp.SSLKeyLogger.create_logfun("test"), tcp.SSLKeyLogger)
+ assert isinstance(
+ tcp.SSLKeyLogger.create_logfun("test"),
+ tcp.SSLKeyLogger)
assert not tcp.SSLKeyLogger.create_logfun(False)
diff --git a/test/test_websockets.py b/test/test_websockets.py
index 38947295..8ed14708 100644
--- a/test/test_websockets.py
+++ b/test/test_websockets.py
@@ -63,7 +63,8 @@ class WebSocketsClient(tcp.TCPClient):
resp = http.read_response(self.rfile, "get", None)
server_nonce = websockets.check_server_handshake(resp.headers)
- if not server_nonce == websockets.create_server_nonce(self.client_nonce):
+ if not server_nonce == websockets.create_server_nonce(
+ self.client_nonce):
self.close()
def read_next_message(self):
generated by cgit v1.2.3 (git 2.25.1) at 2025-05-04 00:31:11 +0000