<commit_before>#!/usr/bin/env python
#
# Copyright 2012, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
#     * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#     * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
#     * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.


"""Tests for msgutil module."""


import array
import Queue
import random
import struct
import unittest
import zlib

import set_sys_path  # Update sys.path to locate mod_pywebsocket module.

from mod_pywebsocket import common
from mod_pywebsocket.extensions import DeflateFrameExtensionProcessor
from mod_pywebsocket.extensions import PerFrameCompressionExtensionProcessor
from mod_pywebsocket.extensions import PerMessageCompressionExtensionProcessor
from mod_pywebsocket import msgutil
from mod_pywebsocket.stream import InvalidUTF8Exception
from mod_pywebsocket.stream import Stream
from mod_pywebsocket.stream import StreamHixie75
from mod_pywebsocket.stream import StreamOptions
from mod_pywebsocket import util
from test import mock


# We use one fixed nonce for testing instead of cryptographically secure PRNG.
_MASKING_NONCE = 'ABCD'


def _mask_hybi(frame):
    frame_key = map(ord, _MASKING_NONCE)
    frame_key_len = len(frame_key)
    result = array.array('B')
    result.fromstring(frame)
    count = 0
    for i in xrange(len(result)):
        result[i] ^= frame_key[count]
        count = (count + 1) % frame_key_len
    return _MASKING_NONCE + result.tostring()


def _install_extension_processor(processor, request, stream_options):
    response = processor.get_extension_response()
    if response is not None:
        processor.setup_stream_options(stream_options)
        request.ws_extension_processors.append(processor)


def _create_request_from_rawdata(
    read_data, deflate_stream=False, deflate_frame_request=None,
    perframe_compression_request=None, permessage_compression_request=None):
    req = mock.MockRequest(connection=mock.MockConn(''.join(read_data)))
    req.ws_version = common.VERSION_HYBI_LATEST
    stream_options = StreamOptions()
    stream_options.deflate_stream = deflate_stream
    req.ws_extension_processors = []
    if deflate_frame_request is not None:
        processor = DeflateFrameExtensionProcessor(deflate_frame_request)
        _install_extension_processor(processor, req, stream_options)
    elif perframe_compression_request is not None:
        processor = PerFrameCompressionExtensionProcessor(
            perframe_compression_request)
        _install_extension_processor(processor, req, stream_options)
    elif permessage_compression_request is not None:
        processor = PerMessageCompressionExtensionProcessor(
            permessage_compression_request)
        _install_extension_processor(processor, req, stream_options)

    req.ws_stream = Stream(req, stream_options)
    return req


def _create_request(*frames):
    """Creates MockRequest using data given as frames.

    frames will be returned on calling request.connection.read() where request
    is MockRequest returned by this function.
    """

    read_data = []
    for (header, body) in frames:
        read_data.append(header + _mask_hybi(body))

    return _create_request_from_rawdata(read_data)


def _create_blocking_request():
    """Creates MockRequest.

    Data written to a MockRequest can be read out by calling
    request.connection.written_data().
    """

    req = mock.MockRequest(connection=mock.MockBlockingConn())
    req.ws_version = common.VERSION_HYBI_LATEST
    stream_options = StreamOptions()
    req.ws_stream = Stream(req, stream_options)
    return req


def _create_request_hixie75(read_data=''):
    req = mock.MockRequest(connection=mock.MockConn(read_data))
    req.ws_stream = StreamHixie75(req)
    return req


def _create_blocking_request_hixie75():
    req = mock.MockRequest(connection=mock.MockBlockingConn())
    req.ws_stream = StreamHixie75(req)
    return req


class MessageTest(unittest.TestCase):
    # Tests for Stream

    def test_send_message(self):
        request = _create_request()
        msgutil.send_message(request, 'Hello')
        self.assertEqual('\x81\x05Hello', request.connection.written_data())

        payload = 'a' * 125
        request = _create_request()
        msgutil.send_message(request, payload)
        self.assertEqual('\x81\x7d' + payload,
                         request.connection.written_data())

    def test_send_medium_message(self):
        payload = 'a' * 126
        request = _create_request()
        msgutil.send_message(request, payload)
        self.assertEqual('\x81\x7e\x00\x7e' + payload,
                         request.connection.written_data())

        payload = 'a' * ((1 << 16) - 1)
        request = _create_request()
        msgutil.send_message(request, payload)
        self.assertEqual('\x81\x7e\xff\xff' + payload,
                         request.connection.written_data())

    def test_send_large_message(self):
        payload = 'a' * (1 << 16)
        request = _create_request()
        msgutil.send_message(request, payload)
        self.assertEqual('\x81\x7f\x00\x00\x00\x00\x00\x01\x00\x00' + payload,
                         request.connection.written_data())

    def test_send_message_unicode(self):
        request = _create_request()
        msgutil.send_message(request, u'\u65e5')
        # U+65e5 is encoded as e6,97,a5 in UTF-8
        self.assertEqual('\x81\x03\xe6\x97\xa5',
                         request.connection.written_data())

    def test_send_message_fragments(self):
        request = _create_request()
        msgutil.send_message(request, 'Hello', False)
        msgutil.send_message(request, ' ', False)
        msgutil.send_message(request, 'World', False)
        msgutil.send_message(request, '!', True)
        self.assertEqual('\x01\x05Hello\x00\x01 \x00\x05World\x80\x01!',
                         request.connection.written_data())

    def test_send_fragments_immediate_zero_termination(self):
        request = _create_request()
        msgutil.send_message(request, 'Hello World!', False)
        msgutil.send_message(request, '', True)
        self.assertEqual('\x01\x0cHello World!\x80\x00',
                         request.connection.written_data())

    def test_send_message_deflate_stream(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)

        request = _create_request_from_rawdata('', deflate_stream=True)
        msgutil.send_message(request, 'Hello')
        expected = compress.compress('\x81\x05Hello')
        expected += compress.flush(zlib.Z_SYNC_FLUSH)
        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_deflate_frame(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        request = _create_request_from_rawdata(
            '', deflate_frame_request=extension)
        msgutil.send_message(request, 'Hello')
        msgutil.send_message(request, 'World')

        expected = ''

        compressed_hello = compress.compress('Hello')
        compressed_hello += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_hello = compressed_hello[:-4]
        expected += '\xc1%c' % len(compressed_hello)
        expected += compressed_hello

        compressed_world = compress.compress('World')
        compressed_world += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_world = compressed_world[:-4]
        expected += '\xc1%c' % len(compressed_world)
        expected += compressed_world

        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_deflate_frame_bfinal(self):
        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        request = _create_request_from_rawdata(
            '', deflate_frame_request=extension)
        self.assertEquals(1, len(request.ws_extension_processors))
        deflate_frame_processor = request.ws_extension_processors[0]
        deflate_frame_processor.set_bfinal(True)
        msgutil.send_message(request, 'Hello')
        msgutil.send_message(request, 'World')

        expected = ''

        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)
        compressed_hello = compress.compress('Hello')
        compressed_hello += compress.flush(zlib.Z_FINISH)
        compressed_hello = compressed_hello + chr(0)
        expected += '\xc1%c' % len(compressed_hello)
        expected += compressed_hello

        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)
        compressed_world = compress.compress('World')
        compressed_world += compress.flush(zlib.Z_FINISH)
        compressed_world = compressed_world + chr(0)
        expected += '\xc1%c' % len(compressed_world)
        expected += compressed_world

        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_deflate_frame_comp_bit(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        request = _create_request_from_rawdata(
            '', deflate_frame_request=extension)
        self.assertEquals(1, len(request.ws_extension_processors))
        deflate_frame_processor = request.ws_extension_processors[0]
        msgutil.send_message(request, 'Hello')
        deflate_frame_processor.disable_outgoing_compression()
        msgutil.send_message(request, 'Hello')
        deflate_frame_processor.enable_outgoing_compression()
        msgutil.send_message(request, 'Hello')

        expected = ''

        compressed_hello = compress.compress('Hello')
        compressed_hello += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_hello = compressed_hello[:-4]
        expected += '\xc1%c' % len(compressed_hello)
        expected += compressed_hello

        expected += '\x81\x05Hello'

        compressed_2nd_hello = compress.compress('Hello')
        compressed_2nd_hello += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_2nd_hello = compressed_2nd_hello[:-4]
        expected += '\xc1%c' % len(compressed_2nd_hello)
        expected += compressed_2nd_hello

        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_deflate_frame_no_context_takeover_parameter(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        extension.add_parameter('no_context_takeover', None)
        request = _create_request_from_rawdata(
            '', deflate_frame_request=extension)
        for i in xrange(3):
            msgutil.send_message(request, 'Hello')

        compressed_message = compress.compress('Hello')
        compressed_message += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_message = compressed_message[:-4]
        expected = '\xc1%c' % len(compressed_message)
        expected += compressed_message

        self.assertEqual(
            expected + expected + expected, request.connection.written_data())

    def test_deflate_frame_bad_request_parameters(self):
        """Tests that if there's anything wrong with deflate-frame extension
        request, deflate-frame is rejected.
        """

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        # max_window_bits less than 8 is illegal.
        extension.add_parameter('max_window_bits', '7')
        processor = DeflateFrameExtensionProcessor(extension)
        self.assertEqual(None, processor.get_extension_response())

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        # max_window_bits greater than 15 is illegal.
        extension.add_parameter('max_window_bits', '16')
        processor = DeflateFrameExtensionProcessor(extension)
        self.assertEqual(None, processor.get_extension_response())

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        # Non integer max_window_bits is illegal.
        extension.add_parameter('max_window_bits', 'foobar')
        processor = DeflateFrameExtensionProcessor(extension)
        self.assertEqual(None, processor.get_extension_response())

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        # no_context_takeover must not have any value.
        extension.add_parameter('no_context_takeover', 'foobar')
        processor = DeflateFrameExtensionProcessor(extension)
        self.assertEqual(None, processor.get_extension_response())

    def test_deflate_frame_response_parameters(self):
        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        processor = DeflateFrameExtensionProcessor(extension)
        processor.set_response_window_bits(8)
        response = processor.get_extension_response()
        self.assertTrue(response.has_parameter('max_window_bits'))
        self.assertEqual('8', response.get_parameter_value('max_window_bits'))

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        processor = DeflateFrameExtensionProcessor(extension)
        processor.set_response_no_context_takeover(True)
        response = processor.get_extension_response()
        self.assertTrue(response.has_parameter('no_context_takeover'))
        self.assertTrue(
            response.get_parameter_value('no_context_takeover') is None)

    def test_permessage_compress_deflate_response_parameters(self):
        extension = common.ExtensionParameter(
            common.PERMESSAGE_COMPRESSION_EXTENSION)
        extension.add_parameter('method', 'deflate')
        processor = PerMessageCompressionExtensionProcessor(extension)
        response = processor.get_extension_response()
        self.assertEqual('deflate',
                         response.get_parameter_value('method'))

        extension = common.ExtensionParameter(
            common.PERMESSAGE_COMPRESSION_EXTENSION)
        extension.add_parameter('method', 'deflate')
        processor = PerMessageCompressionExtensionProcessor(extension)
        def _compression_processor_hook(compression_processor):
            compression_processor.set_c2s_max_window_bits(8)
            compression_processor.set_c2s_no_context_takeover(True)
        processor.set_compression_processor_hook(
            _compression_processor_hook)
        response = processor.get_extension_response()
        self.assertEqual(
            'deflate; c2s_max_window_bits=8; c2s_no_context_takeover',
            response.get_parameter_value('method'))

    def test_send_message_perframe_compress_deflate(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)
        extension = common.ExtensionParameter(
            common.PERFRAME_COMPRESSION_EXTENSION)
        extension.add_parameter('method', 'deflate')
        request = _create_request_from_rawdata(
                      '', perframe_compression_request=extension)
        msgutil.send_message(request, 'Hello')
        msgutil.send_message(request, 'World')

        expected = ''

        compressed_hello = compress.compress('Hello')
        compressed_hello += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_hello = compressed_hello[:-4]
        expected += '\xc1%c' % len(compressed_hello)
        expected += compressed_hello

        compressed_world = compress.compress('World')
        compressed_world += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_world = compressed_world[:-4]
        expected += '\xc1%c' % len(compressed_world)
        expected += compressed_world

        self.assertEqual(expected, request.connection.written_data())
<commit_msg>Add two simple tests using golden data for permessage deflate Review URL: https://codereview.appspot.com/6819057<commit_after>#!/usr/bin/env python
#
# Copyright 2012, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
#     * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#     * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
#     * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.


"""Tests for msgutil module."""


import array
import Queue
import random
import struct
import unittest
import zlib

import set_sys_path  # Update sys.path to locate mod_pywebsocket module.

from mod_pywebsocket import common
from mod_pywebsocket.extensions import DeflateFrameExtensionProcessor
from mod_pywebsocket.extensions import PerFrameCompressionExtensionProcessor
from mod_pywebsocket.extensions import PerMessageCompressionExtensionProcessor
from mod_pywebsocket import msgutil
from mod_pywebsocket.stream import InvalidUTF8Exception
from mod_pywebsocket.stream import Stream
from mod_pywebsocket.stream import StreamHixie75
from mod_pywebsocket.stream import StreamOptions
from mod_pywebsocket import util
from test import mock


# We use one fixed nonce for testing instead of cryptographically secure PRNG.
_MASKING_NONCE = 'ABCD'


def _mask_hybi(frame):
    frame_key = map(ord, _MASKING_NONCE)
    frame_key_len = len(frame_key)
    result = array.array('B')
    result.fromstring(frame)
    count = 0
    for i in xrange(len(result)):
        result[i] ^= frame_key[count]
        count = (count + 1) % frame_key_len
    return _MASKING_NONCE + result.tostring()


def _install_extension_processor(processor, request, stream_options):
    response = processor.get_extension_response()
    if response is not None:
        processor.setup_stream_options(stream_options)
        request.ws_extension_processors.append(processor)


def _create_request_from_rawdata(
    read_data, deflate_stream=False, deflate_frame_request=None,
    perframe_compression_request=None, permessage_compression_request=None):
    req = mock.MockRequest(connection=mock.MockConn(''.join(read_data)))
    req.ws_version = common.VERSION_HYBI_LATEST
    stream_options = StreamOptions()
    stream_options.deflate_stream = deflate_stream
    req.ws_extension_processors = []
    if deflate_frame_request is not None:
        processor = DeflateFrameExtensionProcessor(deflate_frame_request)
        _install_extension_processor(processor, req, stream_options)
    elif perframe_compression_request is not None:
        processor = PerFrameCompressionExtensionProcessor(
            perframe_compression_request)
        _install_extension_processor(processor, req, stream_options)
    elif permessage_compression_request is not None:
        processor = PerMessageCompressionExtensionProcessor(
            permessage_compression_request)
        _install_extension_processor(processor, req, stream_options)

    req.ws_stream = Stream(req, stream_options)
    return req


def _create_request(*frames):
    """Creates MockRequest using data given as frames.

    frames will be returned on calling request.connection.read() where request
    is MockRequest returned by this function.
    """

    read_data = []
    for (header, body) in frames:
        read_data.append(header + _mask_hybi(body))

    return _create_request_from_rawdata(read_data)


def _create_blocking_request():
    """Creates MockRequest.

    Data written to a MockRequest can be read out by calling
    request.connection.written_data().
    """

    req = mock.MockRequest(connection=mock.MockBlockingConn())
    req.ws_version = common.VERSION_HYBI_LATEST
    stream_options = StreamOptions()
    req.ws_stream = Stream(req, stream_options)
    return req


def _create_request_hixie75(read_data=''):
    req = mock.MockRequest(connection=mock.MockConn(read_data))
    req.ws_stream = StreamHixie75(req)
    return req


def _create_blocking_request_hixie75():
    req = mock.MockRequest(connection=mock.MockBlockingConn())
    req.ws_stream = StreamHixie75(req)
    return req


class MessageTest(unittest.TestCase):
    # Tests for Stream

    def test_send_message(self):
        request = _create_request()
        msgutil.send_message(request, 'Hello')
        self.assertEqual('\x81\x05Hello', request.connection.written_data())

        payload = 'a' * 125
        request = _create_request()
        msgutil.send_message(request, payload)
        self.assertEqual('\x81\x7d' + payload,
                         request.connection.written_data())

    def test_send_medium_message(self):
        payload = 'a' * 126
        request = _create_request()
        msgutil.send_message(request, payload)
        self.assertEqual('\x81\x7e\x00\x7e' + payload,
                         request.connection.written_data())

        payload = 'a' * ((1 << 16) - 1)
        request = _create_request()
        msgutil.send_message(request, payload)
        self.assertEqual('\x81\x7e\xff\xff' + payload,
                         request.connection.written_data())

    def test_send_large_message(self):
        payload = 'a' * (1 << 16)
        request = _create_request()
        msgutil.send_message(request, payload)
        self.assertEqual('\x81\x7f\x00\x00\x00\x00\x00\x01\x00\x00' + payload,
                         request.connection.written_data())

    def test_send_message_unicode(self):
        request = _create_request()
        msgutil.send_message(request, u'\u65e5')
        # U+65e5 is encoded as e6,97,a5 in UTF-8
        self.assertEqual('\x81\x03\xe6\x97\xa5',
                         request.connection.written_data())

    def test_send_message_fragments(self):
        request = _create_request()
        msgutil.send_message(request, 'Hello', False)
        msgutil.send_message(request, ' ', False)
        msgutil.send_message(request, 'World', False)
        msgutil.send_message(request, '!', True)
        self.assertEqual('\x01\x05Hello\x00\x01 \x00\x05World\x80\x01!',
                         request.connection.written_data())

    def test_send_fragments_immediate_zero_termination(self):
        request = _create_request()
        msgutil.send_message(request, 'Hello World!', False)
        msgutil.send_message(request, '', True)
        self.assertEqual('\x01\x0cHello World!\x80\x00',
                         request.connection.written_data())

    def test_send_message_deflate_stream(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)

        request = _create_request_from_rawdata('', deflate_stream=True)
        msgutil.send_message(request, 'Hello')
        expected = compress.compress('\x81\x05Hello')
        expected += compress.flush(zlib.Z_SYNC_FLUSH)
        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_deflate_frame(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        request = _create_request_from_rawdata(
            '', deflate_frame_request=extension)
        msgutil.send_message(request, 'Hello')
        msgutil.send_message(request, 'World')

        expected = ''

        compressed_hello = compress.compress('Hello')
        compressed_hello += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_hello = compressed_hello[:-4]
        expected += '\xc1%c' % len(compressed_hello)
        expected += compressed_hello

        compressed_world = compress.compress('World')
        compressed_world += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_world = compressed_world[:-4]
        expected += '\xc1%c' % len(compressed_world)
        expected += compressed_world

        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_deflate_frame_bfinal(self):
        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        request = _create_request_from_rawdata(
            '', deflate_frame_request=extension)
        self.assertEquals(1, len(request.ws_extension_processors))
        deflate_frame_processor = request.ws_extension_processors[0]
        deflate_frame_processor.set_bfinal(True)
        msgutil.send_message(request, 'Hello')
        msgutil.send_message(request, 'World')

        expected = ''

        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)
        compressed_hello = compress.compress('Hello')
        compressed_hello += compress.flush(zlib.Z_FINISH)
        compressed_hello = compressed_hello + chr(0)
        expected += '\xc1%c' % len(compressed_hello)
        expected += compressed_hello

        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)
        compressed_world = compress.compress('World')
        compressed_world += compress.flush(zlib.Z_FINISH)
        compressed_world = compressed_world + chr(0)
        expected += '\xc1%c' % len(compressed_world)
        expected += compressed_world

        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_deflate_frame_comp_bit(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        request = _create_request_from_rawdata(
            '', deflate_frame_request=extension)
        self.assertEquals(1, len(request.ws_extension_processors))
        deflate_frame_processor = request.ws_extension_processors[0]
        msgutil.send_message(request, 'Hello')
        deflate_frame_processor.disable_outgoing_compression()
        msgutil.send_message(request, 'Hello')
        deflate_frame_processor.enable_outgoing_compression()
        msgutil.send_message(request, 'Hello')

        expected = ''

        compressed_hello = compress.compress('Hello')
        compressed_hello += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_hello = compressed_hello[:-4]
        expected += '\xc1%c' % len(compressed_hello)
        expected += compressed_hello

        expected += '\x81\x05Hello'

        compressed_2nd_hello = compress.compress('Hello')
        compressed_2nd_hello += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_2nd_hello = compressed_2nd_hello[:-4]
        expected += '\xc1%c' % len(compressed_2nd_hello)
        expected += compressed_2nd_hello

        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_deflate_frame_no_context_takeover_parameter(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        extension.add_parameter('no_context_takeover', None)
        request = _create_request_from_rawdata(
            '', deflate_frame_request=extension)
        for i in xrange(3):
            msgutil.send_message(request, 'Hello')

        compressed_message = compress.compress('Hello')
        compressed_message += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_message = compressed_message[:-4]
        expected = '\xc1%c' % len(compressed_message)
        expected += compressed_message

        self.assertEqual(
            expected + expected + expected, request.connection.written_data())

    def test_deflate_frame_bad_request_parameters(self):
        """Tests that if there's anything wrong with deflate-frame extension
        request, deflate-frame is rejected.
        """

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        # max_window_bits less than 8 is illegal.
        extension.add_parameter('max_window_bits', '7')
        processor = DeflateFrameExtensionProcessor(extension)
        self.assertEqual(None, processor.get_extension_response())

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        # max_window_bits greater than 15 is illegal.
        extension.add_parameter('max_window_bits', '16')
        processor = DeflateFrameExtensionProcessor(extension)
        self.assertEqual(None, processor.get_extension_response())

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        # Non integer max_window_bits is illegal.
        extension.add_parameter('max_window_bits', 'foobar')
        processor = DeflateFrameExtensionProcessor(extension)
        self.assertEqual(None, processor.get_extension_response())

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        # no_context_takeover must not have any value.
        extension.add_parameter('no_context_takeover', 'foobar')
        processor = DeflateFrameExtensionProcessor(extension)
        self.assertEqual(None, processor.get_extension_response())

    def test_deflate_frame_response_parameters(self):
        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        processor = DeflateFrameExtensionProcessor(extension)
        processor.set_response_window_bits(8)
        response = processor.get_extension_response()
        self.assertTrue(response.has_parameter('max_window_bits'))
        self.assertEqual('8', response.get_parameter_value('max_window_bits'))

        extension = common.ExtensionParameter(common.DEFLATE_FRAME_EXTENSION)
        processor = DeflateFrameExtensionProcessor(extension)
        processor.set_response_no_context_takeover(True)
        response = processor.get_extension_response()
        self.assertTrue(response.has_parameter('no_context_takeover'))
        self.assertTrue(
            response.get_parameter_value('no_context_takeover') is None)

    def test_permessage_compress_deflate_response_parameters(self):
        extension = common.ExtensionParameter(
            common.PERMESSAGE_COMPRESSION_EXTENSION)
        extension.add_parameter('method', 'deflate')
        processor = PerMessageCompressionExtensionProcessor(extension)
        response = processor.get_extension_response()
        self.assertEqual('deflate',
                         response.get_parameter_value('method'))

        extension = common.ExtensionParameter(
            common.PERMESSAGE_COMPRESSION_EXTENSION)
        extension.add_parameter('method', 'deflate')
        processor = PerMessageCompressionExtensionProcessor(extension)
        def _compression_processor_hook(compression_processor):
            compression_processor.set_c2s_max_window_bits(8)
            compression_processor.set_c2s_no_context_takeover(True)
        processor.set_compression_processor_hook(
            _compression_processor_hook)
        response = processor.get_extension_response()
        self.assertEqual(
            'deflate; c2s_max_window_bits=8; c2s_no_context_takeover',
            response.get_parameter_value('method'))

    def test_send_message_perframe_compress_deflate(self):
        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)
        extension = common.ExtensionParameter(
            common.PERFRAME_COMPRESSION_EXTENSION)
        extension.add_parameter('method', 'deflate')
        request = _create_request_from_rawdata(
                      '', perframe_compression_request=extension)
        msgutil.send_message(request, 'Hello')
        msgutil.send_message(request, 'World')

        expected = ''

        compressed_hello = compress.compress('Hello')
        compressed_hello += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_hello = compressed_hello[:-4]
        expected += '\xc1%c' % len(compressed_hello)
        expected += compressed_hello

        compressed_world = compress.compress('World')
        compressed_world += compress.flush(zlib.Z_SYNC_FLUSH)
        compressed_world = compressed_world[:-4]
        expected += '\xc1%c' % len(compressed_world)
        expected += compressed_world

        self.assertEqual(expected, request.connection.written_data())

    def test_send_message_permessage_compress_deflate_empty(self):
        """Test that an empty message is compressed correctly."""

        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)
        extension = common.ExtensionParameter(
            common.PERMESSAGE_COMPRESSION_EXTENSION)
        extension.add_parameter('method', 'deflate')
        request = _create_request_from_rawdata(
                      '', permessage_compression_request=extension)
        msgutil.send_message(request, '')

        # Payload in binary: 0b00000010 0b00000000
        # From LSB,
        # - 1 bit of BFINAL (0)
        # - 2 bits of BTYPE (01 that means fixed Huffman)
        # - 7 bits of the first code (0000000 that is the code for the
        #   end-of-block)
        # - 1 bit of BFINAL (0)
        # - 2 bits of BTYPE (no compression)
        # - 3 bits of padding
        self.assertEqual('\xc1\x02\x02\x00',
                         request.connection.written_data())

    def test_send_message_permessage_compress_deflate_null_character(self):
        """Test that a simple payload (one null) is framed correctly."""

        compress = zlib.compressobj(
            zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS)
        extension = common.ExtensionParameter(
            common.PERMESSAGE_COMPRESSION_EXTENSION)
        extension.add_parameter('method', 'deflate')
        request = _create_request_from_rawdata(
                      '', permessage_compression_request=extension)
        msgutil.send_message(request, '\x00')

        # Payload in binary: 0b01100010 0b00000000 0b00000000
        # From LSB,
        # - 1 bit of BFINAL (0)
        # - 2 bits of BTYPE (01 that means fixed Huffman)
        # - 8 bits of the first code (00110000 that is the code for the literal
        #   alphabet 0x00)
        # - 7 bits of the second code (0000000 that is the code for the
        #   end-of-block)
        # - 1 bit of BFINAL (0)
        # - 2 bits of BTYPE (no compression)
        # - 2 bits of padding
        self.assertEqual('\xc1\x03\x62\x00\x00',
                         request.connection.written_data())

{'fraction_non_alphanumeric': 0.08452309863742467, 'fraction_numerical': 0.011725747787850823, 'mean_word_length': 3.4044231918708907, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 5, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5860681', 'n_tokens_mistral': 10787, 'n_tokens_neox': 10307, 'n_words': 2076}