import cStringIO from test import tutils from nose.tools import assert_equal from netlib import tcp from netlib.http2.frame import * def hex_to_file(data): data = data.decode('hex') return tcp.Reader(cStringIO.StringIO(data)) def test_invalid_flags(): tutils.raises( ValueError, DataFrame, flags=ContinuationFrame.FLAG_END_HEADERS, stream_id=0x1234567, payload='foobar') def test_frame_equality(): 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(): 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( length=6, flags=Frame.FLAG_END_STREAM, stream_id=0x1234567, payload='foobar') assert_equal(f.to_bytes().encode('hex'), '000006000101234567666f6f626172') f = DataFrame( 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( 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_file(hex_to_file('000006000101234567666f6f626172')) assert isinstance(f, DataFrame) assert_equal(f.length, 6) assert_equal(f.TYPE, DataFrame.TYPE) assert_equal(f.flags, Frame.FLAG_END_STREAM) assert_equal(f.stream_id, 0x1234567) assert_equal(f.payload, 'foobar') f = Frame.from_file(hex_to_file('00000a00090123456703666f6f626172000000')) assert isinstance(f, DataFrame) assert_equal(f.length, 10) assert_equal(f.TYPE, DataFrame.TYPE) assert_equal(f.flags, Frame.FLAG_END_STREAM | Frame.FLAG_PADDED) assert_equal(f.stream_id, 0x1234567) assert_equal(f.payload, 'foobar') def test_data_frame_human_readable(): f = DataFrame( 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( 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( 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( length=10, flags=(HeadersFrame.FLAG_PRIORITY), stream_id=0x1234567, header_block_fragment='668594e75e31d9'.decode('hex'), exclusive=True, stream_dependency=0x7654321, weight=42) assert_equal( f.to_bytes().encode('hex'), '00000c012001234567876543212a668594e75e31d9') f = HeadersFrame( length=14, flags=(HeadersFrame.FLAG_PADDED | HeadersFrame.FLAG_PRIORITY), stream_id=0x1234567,