Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_inbound_compressed_control_frame(self) -> None:
payload = b"x" * 23
ext = wpext.PerMessageDeflate()
ext._enabled = True
proto = fp.FrameProtocol(client=True, extensions=[ext])
result = ext.frame_inbound_header(
proto, fp.Opcode.PING, fp.RsvBits(True, False, False), len(payload)
)
assert result == fp.CloseReason.PROTOCOL_ERROR
def test_decompressor_reset(self, client: bool, no_context_takeover: bool) -> None:
if client:
args = {"server_no_context_takeover": no_context_takeover}
else:
args = {"client_no_context_takeover": no_context_takeover}
ext = wpext.PerMessageDeflate(**args)
ext._enabled = True
proto = fp.FrameProtocol(client=client, extensions=[ext])
result = ext.frame_inbound_header(
proto, fp.Opcode.BINARY, fp.RsvBits(True, False, False), 0
)
assert isinstance(result, fp.RsvBits)
assert result.rsv1
assert ext._decompressor is not None
result2 = ext.frame_inbound_complete(proto, True)
assert not isinstance(result2, fp.CloseReason)
if no_context_takeover:
assert ext._decompressor is None
else:
assert ext._decompressor is not None
result3 = ext.frame_inbound_header(
proto, fp.Opcode.BINARY, fp.RsvBits(True, False, False), 0
def test_inbound_compressed_continuation_frame(self) -> None:
payload = b"x" * 23
ext = wpext.PerMessageDeflate()
ext._enabled = True
proto = fp.FrameProtocol(client=True, extensions=[ext])
result = ext.frame_inbound_header(
proto, fp.Opcode.CONTINUATION, fp.RsvBits(True, False, False), len(payload)
)
assert result == fp.CloseReason.PROTOCOL_ERROR
def test_outbound_uncompressible_opcode(self) -> None:
ext = wpext.PerMessageDeflate()
ext._enabled = True
proto = fp.FrameProtocol(client=True, extensions=[ext])
rsv = fp.RsvBits(False, False, False)
payload = b"x" * 23
rsv, data = ext.frame_outbound(proto, fp.Opcode.PING, rsv, payload, True)
assert rsv.rsv1 is False
assert data == payload