aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/__init__.py (renamed from test/h2/__init__.py)0
-rw-r--r--test/h2/example.py18
-rw-r--r--test/http2/__init__.py0
-rw-r--r--test/http2/test_frames.py (renamed from test/h2/test_frames.py)417
-rw-r--r--test/http2/test_http2_protocol.py217
-rw-r--r--test/test_tcp.py81
6 files changed, 487 insertions, 246 deletions
diff --git a/test/h2/__init__.py b/test/__init__.py
index e69de29b..e69de29b 100644
--- a/test/h2/__init__.py
+++ b/test/__init__.py
diff --git a/test/h2/example.py b/test/h2/example.py
deleted file mode 100644
index fc4f2f10..00000000
--- a/test/h2/example.py
+++ /dev/null
@@ -1,18 +0,0 @@
-from netlib.h2.frame import *
-from netlib.h2.h2 import *
-
-c = H2Client(("127.0.0.1", 443))
-c.connect()
-
-c.send_frame(HeadersFrame(
- flags=(Frame.FLAG_END_HEADERS | Frame.FLAG_END_STREAM),
- stream_id=0x1,
- headers=[
- (b':method', 'GET'),
- (b':path', b'/index.html'),
- (b':scheme', b'https'),
- (b':authority', b'localhost'),
- ]))
-
-while True:
- print c.read_frame().human_readable()
diff --git a/test/http2/__init__.py b/test/http2/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/test/http2/__init__.py
diff --git a/test/h2/test_frames.py b/test/http2/test_frames.py
index 313ef405..76a4b712 100644
--- a/test/h2/test_frames.py
+++ b/test/http2/test_frames.py
@@ -1,50 +1,75 @@
-from netlib.h2.frame import *
-import tutils
-
+import cStringIO
+from test import tutils
from nose.tools import assert_equal
+from netlib import tcp
+from netlib.http2.frame import *
+
-# TODO test stream association if valid or not
+def hex_to_file(data):
+ data = data.decode('hex')
+ return tcp.Reader(cStringIO.StringIO(data))
def test_invalid_flags():
tutils.raises(
ValueError,
DataFrame,
- ContinuationFrame.FLAG_END_HEADERS,
- 0x1234567,
- 'foobar')
+ flags=ContinuationFrame.FLAG_END_HEADERS,
+ stream_id=0x1234567,
+ payload='foobar')
def test_frame_equality():
- a = DataFrame(6, Frame.FLAG_END_STREAM, 0x1234567, 'foobar')
- b = DataFrame(6, Frame.FLAG_END_STREAM, 0x1234567, 'foobar')
+ a = DataFrame(
+ length=6,
+ flags=Frame.FLAG_END_STREAM,
+ stream_id=0x1234567,
+ payload='foobar')
+ b = DataFrame(
+ length=6,
+ flags=Frame.FLAG_END_STREAM,
+ stream_id=0x1234567,
+ payload='foobar')
assert_equal(a, b)
def test_too_large_frames():
- DataFrame(6, Frame.FLAG_END_STREAM, 0x1234567)
+ f = DataFrame(
+ length=9000,
+ flags=Frame.FLAG_END_STREAM,
+ stream_id=0x1234567,
+ payload='foobar' * 3000)
+ tutils.raises(FrameSizeError, f.to_bytes)
def test_data_frame_to_bytes():
- f = DataFrame(6, Frame.FLAG_END_STREAM, 0x1234567, 'foobar')
+ f = DataFrame(
+ length=6,
+ flags=Frame.FLAG_END_STREAM,
+ stream_id=0x1234567,
+ payload='foobar')
assert_equal(f.to_bytes().encode('hex'), '000006000101234567666f6f626172')
f = DataFrame(
- 11,
- Frame.FLAG_END_STREAM | Frame.FLAG_PADDED,
- 0x1234567,
- 'foobar',
+ length=11,
+ flags=(Frame.FLAG_END_STREAM | Frame.FLAG_PADDED),
+ stream_id=0x1234567,
+ payload='foobar',
pad_length=3)
assert_equal(
f.to_bytes().encode('hex'),
'00000a00090123456703666f6f626172000000')
- f = DataFrame(6, Frame.FLAG_NO_FLAGS, 0x0, 'foobar')
+ f = DataFrame(
+ length=6,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
+ payload='foobar')
tutils.raises(ValueError, f.to_bytes)
def test_data_frame_from_bytes():
- f = Frame.from_bytes('000006000101234567666f6f626172'.decode('hex'))
+ f = Frame.from_file(hex_to_file('000006000101234567666f6f626172'))
assert isinstance(f, DataFrame)
assert_equal(f.length, 6)
assert_equal(f.TYPE, DataFrame.TYPE)
@@ -52,7 +77,7 @@ def test_data_frame_from_bytes():
assert_equal(f.stream_id, 0x1234567)
assert_equal(f.payload, 'foobar')
- f = Frame.from_bytes('00000a00090123456703666f6f626172000000'.decode('hex'))
+ f = Frame.from_file(hex_to_file('00000a00090123456703666f6f626172000000'))
assert isinstance(f, DataFrame)
assert_equal(f.length, 10)
assert_equal(f.TYPE, DataFrame.TYPE)
@@ -63,37 +88,37 @@ def test_data_frame_from_bytes():
def test_data_frame_human_readable():
f = DataFrame(
- 11,
- Frame.FLAG_END_STREAM | Frame.FLAG_PADDED,
- 0x1234567,
- 'foobar',
+ length=11,
+ flags=(Frame.FLAG_END_STREAM | Frame.FLAG_PADDED),
+ stream_id=0x1234567,
+ payload='foobar',
pad_length=3)
assert f.human_readable()
def test_headers_frame_to_bytes():
f = HeadersFrame(
- 6,
- Frame.FLAG_NO_FLAGS,
- 0x1234567,
- headers=[('host', 'foo.bar')])
+ length=6,
+ flags=(Frame.FLAG_NO_FLAGS),
+ stream_id=0x1234567,
+ header_block_fragment='668594e75e31d9'.decode('hex'))
assert_equal(f.to_bytes().encode('hex'), '000007010001234567668594e75e31d9')
f = HeadersFrame(
- 10,
- HeadersFrame.FLAG_PADDED,
- 0x1234567,
- headers=[('host', 'foo.bar')],
+ length=10,
+ flags=(HeadersFrame.FLAG_PADDED),
+ stream_id=0x1234567,
+ header_block_fragment='668594e75e31d9'.decode('hex'),
pad_length=3)
assert_equal(
f.to_bytes().encode('hex'),
'00000b01080123456703668594e75e31d9000000')
f = HeadersFrame(
- 10,
- HeadersFrame.FLAG_PRIORITY,
- 0x1234567,
- headers=[('host', 'foo.bar')],
+ length=10,
+ flags=(HeadersFrame.FLAG_PRIORITY),
+ stream_id=0x1234567,
+ header_block_fragment='668594e75e31d9'.decode('hex'),
exclusive=True,
stream_dependency=0x7654321,
weight=42)
@@ -102,10 +127,10 @@ def test_headers_frame_to_bytes():
'00000c012001234567876543212a668594e75e31d9')
f = HeadersFrame(
- 14,
- HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY,
- 0x1234567,
- headers=[('host', 'foo.bar')],
+ length=14,
+ flags=(HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY),
+ stream_id=0x1234567,
+ header_block_fragment='668594e75e31d9'.decode('hex'),
pad_length=3,
exclusive=True,
stream_dependency=0x7654321,
@@ -115,10 +140,10 @@ def test_headers_frame_to_bytes():
'00001001280123456703876543212a668594e75e31d9000000')
f = HeadersFrame(
- 14,
- HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY,
- 0x1234567,
- headers=[('host', 'foo.bar')],
+ length=14,
+ flags=(HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY),
+ stream_id=0x1234567,
+ header_block_fragment='668594e75e31d9'.decode('hex'),
pad_length=3,
exclusive=False,
stream_dependency=0x7654321,
@@ -127,60 +152,65 @@ def test_headers_frame_to_bytes():
f.to_bytes().encode('hex'),
'00001001280123456703076543212a668594e75e31d9000000')
- f = HeadersFrame(6, Frame.FLAG_NO_FLAGS, 0x0, 'foobar')
+ f = HeadersFrame(
+ length=6,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
+ header_block_fragment='668594e75e31d9'.decode('hex'))
tutils.raises(ValueError, f.to_bytes)
def test_headers_frame_from_bytes():
- f = Frame.from_bytes('000007010001234567668594e75e31d9'.decode('hex'))
+ f = Frame.from_file(hex_to_file(
+ '000007010001234567668594e75e31d9'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 7)
assert_equal(f.TYPE, HeadersFrame.TYPE)
assert_equal(f.flags, Frame.FLAG_NO_FLAGS)
assert_equal(f.stream_id, 0x1234567)
- assert_equal(f.headers, [('host', 'foo.bar')])
+ assert_equal(f.header_block_fragment, '668594e75e31d9'.decode('hex'))
- f = Frame.from_bytes(
- '00000b01080123456703668594e75e31d9000000'.decode('hex'))
+ f = Frame.from_file(hex_to_file(
+ '00000b01080123456703668594e75e31d9000000'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 11)
assert_equal(f.TYPE, HeadersFrame.TYPE)
assert_equal(f.flags, HeadersFrame.FLAG_PADDED)
assert_equal(f.stream_id, 0x1234567)
- assert_equal(f.headers, [('host', 'foo.bar')])
+ assert_equal(f.header_block_fragment, '668594e75e31d9'.decode('hex'))
- f = Frame.from_bytes(
- '00000c012001234567876543212a668594e75e31d9'.decode('hex'))
+ f = Frame.from_file(hex_to_file(
+ '00000c012001234567876543212a668594e75e31d9'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 12)
assert_equal(f.TYPE, HeadersFrame.TYPE)
assert_equal(f.flags, HeadersFrame.FLAG_PRIORITY)
assert_equal(f.stream_id, 0x1234567)
- assert_equal(f.headers, [('host', 'foo.bar')])
+ assert_equal(f.header_block_fragment, '668594e75e31d9'.decode('hex'))
assert_equal(f.exclusive, True)
assert_equal(f.stream_dependency, 0x7654321)
assert_equal(f.weight, 42)
- f = Frame.from_bytes(
- '00001001280123456703876543212a668594e75e31d9000000'.decode('hex'))
+ f = Frame.from_file(hex_to_file(
+ '00001001280123456703876543212a668594e75e31d9000000'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 16)
assert_equal(f.TYPE, HeadersFrame.TYPE)
assert_equal(f.flags, HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY)
assert_equal(f.stream_id, 0x1234567)
- assert_equal(f.headers, [('host', 'foo.bar')])
+ assert_equal(f.header_block_fragment, '668594e75e31d9'.decode('hex'))
assert_equal(f.exclusive, True)
assert_equal(f.stream_dependency, 0x7654321)
assert_equal(f.weight, 42)
- f = Frame.from_bytes(
- '00001001280123456703076543212a668594e75e31d9000000'.decode('hex'))
+ f = Frame.from_file(hex_to_file(
+ '00001001280123456703076543212a668594e75e31d9000000'))
assert isinstance(f, HeadersFrame)
assert_equal(f.length, 16)
assert_equal(f.TYPE, HeadersFrame.TYPE)
assert_equal(f.flags, HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY)
assert_equal(f.stream_id, 0x1234567)
- assert_equal(f.headers, [('host', 'foo.bar')])
+ assert_equal(f.header_block_fragment, '668594e75e31d9'.decode('hex'))
assert_equal(f.exclusive, False)
assert_equal(f.stream_dependency, 0x7654321)
assert_equal(f.weight, 42)
@@ -188,10 +218,10 @@ def test_headers_frame_from_bytes():
def test_headers_frame_human_readable():
f = HeadersFrame(
- 7,
- HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY,
- 0x1234567,
- headers=[],
+ length=7,
+ flags=(HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY),
+ stream_id=0x1234567,
+ header_block_fragment=b'',
pad_length=3,
exclusive=False,
stream_dependency=0x7654321,
@@ -199,10 +229,10 @@ def test_headers_frame_human_readable():
assert f.human_readable()
f = HeadersFrame(
- 14,
- HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY,
- 0x1234567,
- headers=[('host', 'foo.bar')],
+ length=14,
+ flags=(HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY),
+ stream_id=0x1234567,
+ header_block_fragment='668594e75e31d9'.decode('hex'),
pad_length=3,
exclusive=False,
stream_dependency=0x7654321,
@@ -212,32 +242,40 @@ def test_headers_frame_human_readable():
def test_priority_frame_to_bytes():
f = PriorityFrame(
- 5,
- Frame.FLAG_NO_FLAGS,
- 0x1234567,
+ length=5,
+ flags=(Frame.FLAG_NO_FLAGS),
+ stream_id=0x1234567,
exclusive=True,
stream_dependency=0x7654321,
weight=42)
assert_equal(f.to_bytes().encode('hex'), '000005020001234567876543212a')
f = PriorityFrame(
- 5,
- Frame.FLAG_NO_FLAGS,
- 0x1234567,
+ length=5,
+ flags=(Frame.FLAG_NO_FLAGS),
+ stream_id=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)
+ f = PriorityFrame(
+ length=5,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
+ stream_dependency=0x1234567)
tutils.raises(ValueError, f.to_bytes)
- f = PriorityFrame(5, Frame.FLAG_NO_FLAGS, 0x1234567, stream_dependency=0x0)
+ f = PriorityFrame(
+ length=5,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
+ stream_dependency=0x0)
tutils.raises(ValueError, f.to_bytes)
def test_priority_frame_from_bytes():
- f = Frame.from_bytes('000005020001234567876543212a'.decode('hex'))
+ f = Frame.from_file(hex_to_file('000005020001234567876543212a'))
assert isinstance(f, PriorityFrame)
assert_equal(f.length, 5)
assert_equal(f.TYPE, PriorityFrame.TYPE)
@@ -247,7 +285,7 @@ def test_priority_frame_from_bytes():
assert_equal(f.stream_dependency, 0x7654321)
assert_equal(f.weight, 42)
- f = Frame.from_bytes('0000050200012345670765432115'.decode('hex'))
+ f = Frame.from_file(hex_to_file('0000050200012345670765432115'))
assert isinstance(f, PriorityFrame)
assert_equal(f.length, 5)
assert_equal(f.TYPE, PriorityFrame.TYPE)
@@ -260,9 +298,9 @@ def test_priority_frame_from_bytes():
def test_priority_frame_human_readable():
f = PriorityFrame(
- 5,
- Frame.FLAG_NO_FLAGS,
- 0x1234567,
+ length=5,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
exclusive=False,
stream_dependency=0x7654321,
weight=21)
@@ -270,15 +308,22 @@ def test_priority_frame_human_readable():
def test_rst_stream_frame_to_bytes():
- f = RstStreamFrame(4, Frame.FLAG_NO_FLAGS, 0x1234567, error_code=0x7654321)
+ f = RstStreamFrame(
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
+ error_code=0x7654321)
assert_equal(f.to_bytes().encode('hex'), '00000403000123456707654321')
- f = RstStreamFrame(4, Frame.FLAG_NO_FLAGS, 0x0)
+ f = RstStreamFrame(
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0)
tutils.raises(ValueError, f.to_bytes)
def test_rst_stream_frame_from_bytes():
- f = Frame.from_bytes('00000403000123456707654321'.decode('hex'))
+ f = Frame.from_file(hex_to_file('00000403000123456707654321'))
assert isinstance(f, RstStreamFrame)
assert_equal(f.length, 4)
assert_equal(f.TYPE, RstStreamFrame.TYPE)
@@ -288,28 +333,39 @@ def test_rst_stream_frame_from_bytes():
def test_rst_stream_frame_human_readable():
- f = RstStreamFrame(4, Frame.FLAG_NO_FLAGS, 0x1234567, error_code=0x7654321)
+ f = RstStreamFrame(
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
+ error_code=0x7654321)
assert f.human_readable()
def test_settings_frame_to_bytes():
- f = SettingsFrame(0, Frame.FLAG_NO_FLAGS, 0x0)
+ f = SettingsFrame(
+ length=0,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0)
assert_equal(f.to_bytes().encode('hex'), '000000040000000000')
- f = SettingsFrame(0, SettingsFrame.FLAG_ACK, 0x0)
+ f = SettingsFrame(
+ length=0,
+ flags=SettingsFrame.FLAG_ACK,
+ stream_id=0x0)
assert_equal(f.to_bytes().encode('hex'), '000000040100000000')
f = SettingsFrame(
- 6,
- SettingsFrame.FLAG_ACK, 0x0,
+ length=6,
+ flags=SettingsFrame.FLAG_ACK,
+ stream_id=0x0,
settings={
SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH: 1})
assert_equal(f.to_bytes().encode('hex'), '000006040100000000000200000001')
f = SettingsFrame(
- 12,
- Frame.FLAG_NO_FLAGS,
- 0x0,
+ length=12,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
settings={
SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH: 1,
SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS: 0x12345678})
@@ -317,26 +373,29 @@ def test_settings_frame_to_bytes():
f.to_bytes().encode('hex'),
'00000c040000000000000200000001000312345678')
- f = SettingsFrame(0, Frame.FLAG_NO_FLAGS, 0x1234567)
+ f = SettingsFrame(
+ length=0,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567)
tutils.raises(ValueError, f.to_bytes)
def test_settings_frame_from_bytes():
- f = Frame.from_bytes('000000040000000000'.decode('hex'))
+ f = Frame.from_file(hex_to_file('000000040000000000'))
assert isinstance(f, SettingsFrame)
assert_equal(f.length, 0)
assert_equal(f.TYPE, SettingsFrame.TYPE)
assert_equal(f.flags, Frame.FLAG_NO_FLAGS)
assert_equal(f.stream_id, 0x0)
- f = Frame.from_bytes('000000040100000000'.decode('hex'))
+ f = Frame.from_file(hex_to_file('000000040100000000'))
assert isinstance(f, SettingsFrame)
assert_equal(f.length, 0)
assert_equal(f.TYPE, SettingsFrame.TYPE)
assert_equal(f.flags, SettingsFrame.FLAG_ACK)
assert_equal(f.stream_id, 0x0)
- f = Frame.from_bytes('000006040100000000000200000001'.decode('hex'))
+ f = Frame.from_file(hex_to_file('000006040100000000000200000001'))
assert isinstance(f, SettingsFrame)
assert_equal(f.length, 6)
assert_equal(f.TYPE, SettingsFrame.TYPE)
@@ -345,8 +404,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_file(hex_to_file(
+ '00000c040000000000000200000001000312345678'))
assert isinstance(f, SettingsFrame)
assert_equal(f.length, 12)
assert_equal(f.TYPE, SettingsFrame.TYPE)
@@ -361,13 +420,17 @@ def test_settings_frame_from_bytes():
def test_settings_frame_human_readable():
- f = SettingsFrame(12, Frame.FLAG_NO_FLAGS, 0x0, settings={})
+ f = SettingsFrame(
+ length=12,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
+ settings={})
assert f.human_readable()
f = SettingsFrame(
- 12,
- Frame.FLAG_NO_FLAGS,
- 0x0,
+ length=12,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
settings={
SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH: 1,
SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS: 0x12345678})
@@ -376,35 +439,43 @@ def test_settings_frame_human_readable():
def test_push_promise_frame_to_bytes():
f = PushPromiseFrame(
- 10,
- Frame.FLAG_NO_FLAGS,
- 0x1234567,
- 0x7654321,
- 'foobar')
+ length=10,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
+ promised_stream=0x7654321,
+ header_block_fragment='foobar')
assert_equal(
f.to_bytes().encode('hex'),
'00000a05000123456707654321666f6f626172')
f = PushPromiseFrame(
- 14,
- HeadersFrame.FLAG_PADDED,
- 0x1234567,
- 0x7654321,
- 'foobar',
+ length=14,
+ flags=HeadersFrame.FLAG_PADDED,
+ stream_id=0x1234567,
+ promised_stream=0x7654321,
+ header_block_fragment='foobar',
pad_length=3)
assert_equal(
f.to_bytes().encode('hex'),
'00000e0508012345670307654321666f6f626172000000')
- f = PushPromiseFrame(4, Frame.FLAG_NO_FLAGS, 0x0, 0x1234567)
+ f = PushPromiseFrame(
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
+ promised_stream=0x1234567)
tutils.raises(ValueError, f.to_bytes)
- f = PushPromiseFrame(4, Frame.FLAG_NO_FLAGS, 0x1234567, 0x0)
+ f = PushPromiseFrame(
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
+ promised_stream=0x0)
tutils.raises(ValueError, f.to_bytes)
def test_push_promise_frame_from_bytes():
- f = Frame.from_bytes('00000a05000123456707654321666f6f626172'.decode('hex'))
+ f = Frame.from_file(hex_to_file('00000a05000123456707654321666f6f626172'))
assert isinstance(f, PushPromiseFrame)
assert_equal(f.length, 10)
assert_equal(f.TYPE, PushPromiseFrame.TYPE)
@@ -412,8 +483,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_file(hex_to_file(
+ '00000e0508012345670307654321666f6f626172000000'))
assert isinstance(f, PushPromiseFrame)
assert_equal(f.length, 14)
assert_equal(f.TYPE, PushPromiseFrame.TYPE)
@@ -424,32 +495,43 @@ def test_push_promise_frame_from_bytes():
def test_push_promise_frame_human_readable():
f = PushPromiseFrame(
- 14,
- HeadersFrame.FLAG_PADDED,
- 0x1234567,
- 0x7654321,
- 'foobar',
+ length=14,
+ flags=HeadersFrame.FLAG_PADDED,
+ stream_id=0x1234567,
+ promised_stream=0x7654321,
+ header_block_fragment='foobar',
pad_length=3)
assert f.human_readable()
def test_ping_frame_to_bytes():
- f = PingFrame(8, PingFrame.FLAG_ACK, 0x0, payload=b'foobar')
+ f = PingFrame(
+ length=8,
+ flags=PingFrame.FLAG_ACK,
+ stream_id=0x0,
+ payload=b'foobar')
assert_equal(
f.to_bytes().encode('hex'),
'000008060100000000666f6f6261720000')
- f = PingFrame(8, Frame.FLAG_NO_FLAGS, 0x0, payload=b'foobardeadbeef')
+ f = PingFrame(
+ length=8,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
+ payload=b'foobardeadbeef')
assert_equal(
f.to_bytes().encode('hex'),
'000008060000000000666f6f6261726465')
- f = PingFrame(8, Frame.FLAG_NO_FLAGS, 0x1234567)
+ f = PingFrame(
+ length=8,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567)
tutils.raises(ValueError, f.to_bytes)
def test_ping_frame_from_bytes():
- f = Frame.from_bytes('000008060100000000666f6f6261720000'.decode('hex'))
+ f = Frame.from_file(hex_to_file('000008060100000000666f6f6261720000'))
assert isinstance(f, PingFrame)
assert_equal(f.length, 8)
assert_equal(f.TYPE, PingFrame.TYPE)
@@ -457,7 +539,7 @@ def test_ping_frame_from_bytes():
assert_equal(f.stream_id, 0x0)
assert_equal(f.payload, b'foobar\0\0')
- f = Frame.from_bytes('000008060000000000666f6f6261726465'.decode('hex'))
+ f = Frame.from_file(hex_to_file('000008060000000000666f6f6261726465'))
assert isinstance(f, PingFrame)
assert_equal(f.length, 8)
assert_equal(f.TYPE, PingFrame.TYPE)
@@ -467,15 +549,19 @@ def test_ping_frame_from_bytes():
def test_ping_frame_human_readable():
- f = PingFrame(8, PingFrame.FLAG_ACK, 0x0, payload=b'foobar')
+ f = PingFrame(
+ length=8,
+ flags=PingFrame.FLAG_ACK,
+ stream_id=0x0,
+ payload=b'foobar')
assert f.human_readable()
def test_goaway_frame_to_bytes():
f = GoAwayFrame(
- 8,
- Frame.FLAG_NO_FLAGS,
- 0x0,
+ length=8,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
last_stream=0x1234567,
error_code=0x87654321,
data=b'')
@@ -484,9 +570,9 @@ def test_goaway_frame_to_bytes():
'0000080700000000000123456787654321')
f = GoAwayFrame(
- 14,
- Frame.FLAG_NO_FLAGS,
- 0x0,
+ length=14,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
last_stream=0x1234567,
error_code=0x87654321,
data=b'foobar')
@@ -495,16 +581,17 @@ def test_goaway_frame_to_bytes():
'00000e0700000000000123456787654321666f6f626172')
f = GoAwayFrame(
- 8,
- Frame.FLAG_NO_FLAGS,
- 0x1234567,
+ length=8,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
last_stream=0x1234567,
error_code=0x87654321)
tutils.raises(ValueError, f.to_bytes)
def test_goaway_frame_from_bytes():
- f = Frame.from_bytes('0000080700000000000123456787654321'.decode('hex'))
+ f = Frame.from_file(hex_to_file(
+ '0000080700000000000123456787654321'))
assert isinstance(f, GoAwayFrame)
assert_equal(f.length, 8)
assert_equal(f.TYPE, GoAwayFrame.TYPE)
@@ -514,8 +601,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_file(hex_to_file(
+ '00000e0700000000000123456787654321666f6f626172'))
assert isinstance(f, GoAwayFrame)
assert_equal(f.length, 14)
assert_equal(f.TYPE, GoAwayFrame.TYPE)
@@ -528,9 +615,9 @@ def test_goaway_frame_from_bytes():
def test_go_away_frame_human_readable():
f = GoAwayFrame(
- 14,
- Frame.FLAG_NO_FLAGS,
- 0x0,
+ length=14,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
last_stream=0x1234567,
error_code=0x87654321,
data=b'foobar')
@@ -539,23 +626,23 @@ def test_go_away_frame_human_readable():
def test_window_update_frame_to_bytes():
f = WindowUpdateFrame(
- 4,
- Frame.FLAG_NO_FLAGS,
- 0x0,
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
window_size_increment=0x1234567)
assert_equal(f.to_bytes().encode('hex'), '00000408000000000001234567')
f = WindowUpdateFrame(
- 4,
- Frame.FLAG_NO_FLAGS,
- 0x1234567,
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
window_size_increment=0x7654321)
assert_equal(f.to_bytes().encode('hex'), '00000408000123456707654321')
f = WindowUpdateFrame(
- 4,
- Frame.FLAG_NO_FLAGS,
- 0x0,
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x0,
window_size_increment=0xdeadbeef)
tutils.raises(ValueError, f.to_bytes)
@@ -564,7 +651,7 @@ def test_window_update_frame_to_bytes():
def test_window_update_frame_from_bytes():
- f = Frame.from_bytes('00000408000000000001234567'.decode('hex'))
+ f = Frame.from_file(hex_to_file('00000408000000000001234567'))
assert isinstance(f, WindowUpdateFrame)
assert_equal(f.length, 4)
assert_equal(f.TYPE, WindowUpdateFrame.TYPE)
@@ -575,27 +662,31 @@ def test_window_update_frame_from_bytes():
def test_window_update_frame_human_readable():
f = WindowUpdateFrame(
- 4,
- Frame.FLAG_NO_FLAGS,
- 0x1234567,
+ length=4,
+ flags=Frame.FLAG_NO_FLAGS,
+ stream_id=0x1234567,
window_size_increment=0x7654321)
assert f.human_readable()
def test_continuation_frame_to_bytes():
f = ContinuationFrame(
- 6,
- ContinuationFrame.FLAG_END_HEADERS,
- 0x1234567,
- 'foobar')
+ length=6,
+ flags=ContinuationFrame.FLAG_END_HEADERS,
+ stream_id=0x1234567,
+ header_block_fragment='foobar')
assert_equal(f.to_bytes().encode('hex'), '000006090401234567666f6f626172')
- f = ContinuationFrame(6, ContinuationFrame.FLAG_END_HEADERS, 0x0, 'foobar')
+ f = ContinuationFrame(
+ length=6,
+ flags=ContinuationFrame.FLAG_END_HEADERS,
+ stream_id=0x0,
+ header_block_fragment='foobar')
tutils.raises(ValueError, f.to_bytes)
def test_continuation_frame_from_bytes():
- f = Frame.from_bytes('000006090401234567666f6f626172'.decode('hex'))
+ f = Frame.from_file(hex_to_file('000006090401234567666f6f626172'))
assert isinstance(f, ContinuationFrame)
assert_equal(f.length, 6)
assert_equal(f.TYPE, ContinuationFrame.TYPE)
@@ -606,8 +697,8 @@ def test_continuation_frame_from_bytes():
def test_continuation_frame_human_readable():
f = ContinuationFrame(
- 6,
- ContinuationFrame.FLAG_END_HEADERS,
- 0x1234567,
- 'foobar')
+ length=6,
+ flags=ContinuationFrame.FLAG_END_HEADERS,
+ stream_id=0x1234567,
+ header_block_fragment='foobar')
assert f.human_readable()
diff --git a/test/http2/test_http2_protocol.py b/test/http2/test_http2_protocol.py
new file mode 100644
index 00000000..cb46bc68
--- /dev/null
+++ b/test/http2/test_http2_protocol.py
@@ -0,0 +1,217 @@
+
+import OpenSSL
+
+from netlib import http2
+from netlib import tcp
+from netlib import test
+from netlib.http2.frame import *
+from test import tutils
+
+
+class EchoHandler(tcp.BaseHandler):
+ sni = None
+
+ def handle(self):
+ while True:
+ v = self.rfile.safe_read(1)
+ self.wfile.write(v)
+ self.wfile.flush()
+
+
+class TestCheckALPNMatch(test.ServerTestBase):
+ handler = EchoHandler
+ ssl = dict(
+ alpn_select=http2.HTTP2Protocol.ALPN_PROTO_H2,
+ )
+
+ if OpenSSL._util.lib.Cryptography_HAS_ALPN:
+
+ def test_check_alpn(self):
+ c = tcp.TCPClient(("127.0.0.1", self.port))
+ c.connect()
+ c.convert_to_ssl(alpn_protos=[http2.HTTP2Protocol.ALPN_PROTO_H2])
+ protocol = http2.HTTP2Protocol(c)
+ assert protocol.check_alpn()
+
+
+class TestCheckALPNMismatch(test.ServerTestBase):
+ handler = EchoHandler
+ ssl = dict(
+ alpn_select=None,
+ )
+
+ if OpenSSL._util.lib.Cryptography_HAS_ALPN:
+
+ def test_check_alpn(self):
+ c = tcp.TCPClient(("127.0.0.1", self.port))
+ c.connect()
+ c.convert_to_ssl(alpn_protos=[http2.HTTP2Protocol.ALPN_PROTO_H2])
+ protocol = http2.HTTP2Protocol(c)
+ tutils.raises(NotImplementedError, protocol.check_alpn)
+
+
+class TestPerformConnectionPreface(test.ServerTestBase):
+ class handler(tcp.BaseHandler):
+
+ def handle(self):
+ # check magic
+ assert self.rfile.read(24) ==\
+ '505249202a20485454502f322e300d0a0d0a534d0d0a0d0a'.decode('hex')
+
+ # check empty settings frame
+ assert self.rfile.read(9) ==\
+ '000000040000000000'.decode('hex')
+
+ # send empty settings frame
+ self.wfile.write('000000040000000000'.decode('hex'))
+ self.wfile.flush()
+
+ # check settings acknowledgement
+ assert self.rfile.read(9) == \
+ '000000040100000000'.decode('hex')
+
+ # send settings acknowledgement
+ self.wfile.write('000000040100000000'.decode('hex'))
+ self.wfile.flush()
+
+ ssl = True
+
+ def test_perform_connection_preface(self):
+ c = tcp.TCPClient(("127.0.0.1", self.port))
+ c.connect()
+ c.convert_to_ssl()
+ protocol = http2.HTTP2Protocol(c)
+ protocol.perform_connection_preface()
+
+
+class TestStreamIds():
+ c = tcp.TCPClient(("127.0.0.1", 0))
+ protocol = http2.HTTP2Protocol(c)
+
+ def test_stream_ids(self):
+ assert self.protocol.current_stream_id is None
+ assert self.protocol.next_stream_id() == 1
+ assert self.protocol.current_stream_id == 1
+ assert self.protocol.next_stream_id() == 3
+ assert self.protocol.current_stream_id == 3
+ assert self.protocol.next_stream_id() == 5
+ assert self.protocol.current_stream_id == 5
+
+
+class TestApplySettings(test.ServerTestBase):
+ class handler(tcp.BaseHandler):
+
+ def handle(self):
+ # check settings acknowledgement
+ assert self.rfile.read(9) == '000000040100000000'.decode('hex')
+ self.wfile.write("OK")
+ self.wfile.flush()
+
+ ssl = True
+
+ def test_apply_settings(self):
+ c = tcp.TCPClient(("127.0.0.1", self.port))
+ c.connect()
+ c.convert_to_ssl()
+ protocol = http2.HTTP2Protocol(c)
+
+ protocol._apply_settings({
+ SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH: 'foo',
+ SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS: 'bar',
+ SettingsFrame.SETTINGS.SETTINGS_INITIAL_WINDOW_SIZE: 'deadbeef',
+ })
+
+ assert c.rfile.safe_read(2) == "OK"
+
+ assert protocol.http2_settings[
+ SettingsFrame.SETTINGS.SETTINGS_ENABLE_PUSH] == 'foo'
+ assert protocol.http2_settings[
+ SettingsFrame.SETTINGS.SETTINGS_MAX_CONCURRENT_STREAMS] == 'bar'
+ assert protocol.http2_settings[
+ SettingsFrame.SETTINGS.SETTINGS_INITIAL_WINDOW_SIZE] == 'deadbeef'
+
+
+class TestCreateHeaders():
+ c = tcp.TCPClient(("127.0.0.1", 0))
+
+ def test_create_headers(self):
+ headers = [
+ (b':method', b'GET'),
+ (b':path', b'index.html'),
+ (b':scheme', b'https'),
+ (b'foo', b'bar')]
+
+ bytes = http2.HTTP2Protocol(self.c)._create_headers(
+ headers, 1, end_stream=True)
+ assert b''.join(bytes) ==\
+ '000014010500000001824488355217caf3a69a3f87408294e7838c767f'\
+ .decode('hex')
+
+ bytes = http2.HTTP2Protocol(self.c)._create_headers(
+ headers, 1, end_stream=False)
+ assert b''.join(bytes) ==\
+ '000014010400000001824488355217caf3a69a3f87408294e7838c767f'\
+ .decode('hex')
+
+ # TODO: add test for too large header_block_fragments
+
+
+class TestCreateBody():
+ c = tcp.TCPClient(("127.0.0.1", 0))
+ protocol = http2.HTTP2Protocol(c)
+
+ def test_create_body_empty(self):
+ bytes = self.protocol._create_body(b'', 1)
+ assert b''.join(bytes) == ''.decode('hex')
+
+ def test_create_body_single_frame(self):
+ bytes = self.protocol._create_body('foobar', 1)
+ assert b''.join(bytes) == '000006000100000001666f6f626172'.decode('hex')
+
+ def test_create_body_multiple_frames(self):
+ pass
+ # bytes = self.protocol._create_body('foobar' * 3000, 1)
+ # TODO: add test for too large frames
+
+
+class TestCreateRequest():
+ c = tcp.TCPClient(("127.0.0.1", 0))
+
+ def test_create_request_simple(self):
+ bytes = http2.HTTP2Protocol(self.c).create_request('GET', '/')
+ assert len(bytes) == 1
+ assert bytes[0] == '000003010500000001828487'.decode('hex')
+
+ def test_create_request_with_body(self):
+ bytes = http2.HTTP2Protocol(self.c).create_request(
+ 'GET', '/', [(b'foo', b'bar')], 'foobar')
+ assert len(bytes) == 2
+ assert bytes[0] ==\
+ '00000b010400000001828487408294e7838c767f'.decode('hex')
+ assert bytes[1] ==\
+ '000006000100000001666f6f626172'.decode('hex')
+
+
+class TestReadResponse(test.ServerTestBase):
+ class handler(tcp.BaseHandler):
+
+ def handle(self):
+ self.wfile.write(
+ b'00000801040000000188628594e78c767f'.decode('hex'))
+ self.wfile.write(
+ b'000006000100000001666f6f626172'.decode('hex'))
+ self.wfile.flush()
+
+ ssl = True
+
+ def test_read_response(self):
+ c = tcp.TCPClient(("127.0.0.1", self.port))
+ c.connect()
+ c.convert_to_ssl()
+ protocol = http2.HTTP2Protocol(c)
+
+ status, headers, body = protocol.read_response()
+
+ assert headers == {':status': '200', 'etag': 'foobar'}
+ assert status == '200'
+ assert body == b'foobar'
diff --git a/test/test_tcp.py b/test/test_tcp.py
index 14ba555d..f8fc6a28 100644
--- a/test/test_tcp.py
+++ b/test/test_tcp.py
@@ -135,10 +135,6 @@ class TestFinishFail(test.ServerTestBase):
class TestServerSSL(test.ServerTestBase):
handler = EchoHandler
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False,
cipher_list="AES256-SHA",
chain_file=tutils.test_data.path("data/server.crt")
)
@@ -165,8 +161,6 @@ class TestServerSSL(test.ServerTestBase):
class TestSSLv3Only(test.ServerTestBase):
handler = EchoHandler
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
request_client_cert=False,
v3_only=True
)
@@ -188,9 +182,8 @@ class TestSSLClientCert(test.ServerTestBase):
def handle(self):
self.wfile.write("%s\n" % self.clientcert.serial)
self.wfile.flush()
+
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
request_client_cert=True,
v3_only=False
)
@@ -224,12 +217,7 @@ class TestSNI(test.ServerTestBase):
self.wfile.write(self.sni)
self.wfile.flush()
- ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False
- )
+ ssl = True
def test_echo(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
@@ -242,10 +230,6 @@ class TestSNI(test.ServerTestBase):
class TestServerCipherList(test.ServerTestBase):
handler = ClientCipherListHandler
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False,
cipher_list='RC4-SHA'
)
@@ -264,11 +248,8 @@ class TestServerCurrentCipher(test.ServerTestBase):
def handle(self):
self.wfile.write("%s" % str(self.get_current_cipher()))
self.wfile.flush()
+
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False,
cipher_list='RC4-SHA'
)
@@ -282,10 +263,6 @@ class TestServerCurrentCipher(test.ServerTestBase):
class TestServerCipherListError(test.ServerTestBase):
handler = ClientCipherListHandler
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False,
cipher_list='bogus'
)
@@ -298,10 +275,6 @@ class TestServerCipherListError(test.ServerTestBase):
class TestClientCipherListError(test.ServerTestBase):
handler = ClientCipherListHandler
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False,
cipher_list='RC4-SHA'
)
@@ -321,12 +294,8 @@ class TestSSLDisconnect(test.ServerTestBase):
def handle(self):
self.finish()
- ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False
- )
+
+ ssl = True
def test_echo(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
@@ -341,12 +310,7 @@ class TestSSLDisconnect(test.ServerTestBase):
class TestSSLHardDisconnect(test.ServerTestBase):
handler = HardDisconnectHandler
- ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False
- )
+ ssl = True
def test_echo(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
@@ -400,13 +364,9 @@ class TestTimeOut(test.ServerTestBase):
class TestALPN(test.ServerTestBase):
- handler = HangHandler
+ handler = EchoHandler
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False,
- alpn_select="h2"
+ alpn_select="foobar"
)
if OpenSSL._util.lib.Cryptography_HAS_ALPN:
@@ -414,19 +374,13 @@ class TestALPN(test.ServerTestBase):
def test_alpn(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
- c.convert_to_ssl(alpn_protos=["h2"])
- print "ALPN: %s" % c.get_alpn_proto_negotiated()
- assert c.get_alpn_proto_negotiated() == "h2"
+ c.convert_to_ssl(alpn_protos=["foobar"])
+ assert c.get_alpn_proto_negotiated() == "foobar"
class TestSSLTimeOut(test.ServerTestBase):
handler = HangHandler
- ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False
- )
+ ssl = True
def test_timeout_client(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
@@ -439,10 +393,6 @@ class TestSSLTimeOut(test.ServerTestBase):
class TestDHParams(test.ServerTestBase):
handler = HangHandler
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False,
dhparams=certutils.CertStore.load_dhparam(
tutils.test_data.path("data/dhparam.pem"),
),
@@ -625,6 +575,11 @@ class TestFileLike:
s = tcp.Reader(o)
tutils.raises(tcp.NetLibDisconnect, s.readline, 10)
+ def test_reader_incomplete_error(self):
+ s = cStringIO.StringIO("foobar")
+ s = tcp.Reader(s)
+ tutils.raises(tcp.NetLibIncomplete, s.safe_read, 10)
+
class TestAddress:
@@ -643,10 +598,6 @@ class TestAddress:
class TestSSLKeyLogger(test.ServerTestBase):
handler = EchoHandler
ssl = dict(
- cert=tutils.test_data.path("data/server.crt"),
- key=tutils.test_data.path("data/server.key"),
- request_client_cert=False,
- v3_only=False,
cipher_list="AES256-SHA"
)
generated by cgit v1.2.3 (git 2.25.1) at 2025-05-01 10:48:35 +0000