<commit_before>#!/usr/bin/python
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from __future__ import unicode_literals

import sys

from base64 import b64decode

from kombu.serialization import (registry, register, SerializerNotInstalled,
                                 raw_encode, register_yaml, register_msgpack,
                                 decode, bytes_t, pickle, pickle_protocol,
                                 unregister, register_pickle)

from .utils import TestCase
from .utils import mask_modules, skip_if_not_module

# For content_encoding tests
unicode_string = 'abcdé\u8463'
unicode_string_as_utf8 = unicode_string.encode('utf-8')
latin_string = 'abcdé'
latin_string_as_latin1 = latin_string.encode('latin-1')
latin_string_as_utf8 = latin_string.encode('utf-8')


# For serialization tests
py_data = {
    'string': 'The quick brown fox jumps over the lazy dog',
    'int': 10,
    'float': 3.14159265,
    'unicode': 'Thé quick brown fox jumps over thé lazy dog',
    'list': ['george', 'jerry', 'elaine', 'cosmo'],
}

# JSON serialization tests
json_data = """\
{"int": 10, "float": 3.1415926500000002, \
"list": ["george", "jerry", "elaine", "cosmo"], \
"string": "The quick brown fox jumps over the lazy \
dog", "unicode": "Th\\u00e9 quick brown fox jumps over \
th\\u00e9 lazy dog"}\
"""

# Pickle serialization tests
pickle_data = pickle.dumps(py_data, protocol=pickle_protocol)

# YAML serialization tests
yaml_data = """\
float: 3.1415926500000002
int: 10
list: [george, jerry, elaine, cosmo]
string: The quick brown fox jumps over the lazy dog
unicode: "Th\\xE9 quick brown fox jumps over th\\xE9 lazy dog"
"""


msgpack_py_data = dict(py_data)
# msgpack only supports tuples
#msgpack_py_data['list'] = msgpack_py_data['list']
# Unicode chars are lost in transmit :(
msgpack_py_data['unicode'] = 'Th quick brown fox jumps over th lazy dog'
msgpack_data = b64decode("""\
<KEY>
""")


def say(m):
    sys.stderr.write('%s\n' % (m, ))


registry.register('testS', lambda s: s, lambda s: 'decoded',
                  'application/testS', 'utf-8')


class test_Serialization(TestCase):

    def test_disable(self):
        disabled = registry._disabled_content_types
        try:
            registry.disable('testS')
            self.assertIn('application/testS', disabled)
            disabled.clear()

            registry.disable('application/testS')
            self.assertIn('application/testS', disabled)<commit_msg>[Py3] Fixes test that was broken on Py3.2<commit_after>#!/usr/bin/python
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from __future__ import unicode_literals

import sys

from base64 import b64decode

from kombu.serialization import (
    registry, register, SerializerNotInstalled,
    raw_encode, register_yaml, register_msgpack,
    decode, bytes_t, pickle, pickle_protocol,
    unregister, register_pickle,
)
from kombu.utils.encoding import str_to_bytes

from .utils import TestCase
from .utils import mask_modules, skip_if_not_module

# For content_encoding tests
unicode_string = 'abcdé\u8463'
unicode_string_as_utf8 = unicode_string.encode('utf-8')
latin_string = 'abcdé'
latin_string_as_latin1 = latin_string.encode('latin-1')
latin_string_as_utf8 = latin_string.encode('utf-8')


# For serialization tests
py_data = {
    'string': 'The quick brown fox jumps over the lazy dog',
    'int': 10,
    'float': 3.14159265,
    'unicode': 'Thé quick brown fox jumps over thé lazy dog',
    'list': ['george', 'jerry', 'elaine', 'cosmo'],
}

# JSON serialization tests
json_data = """\
{"int": 10, "float": 3.1415926500000002, \
"list": ["george", "jerry", "elaine", "cosmo"], \
"string": "The quick brown fox jumps over the lazy \
dog", "unicode": "Th\\u00e9 quick brown fox jumps over \
th\\u00e9 lazy dog"}\
"""

# Pickle serialization tests
pickle_data = pickle.dumps(py_data, protocol=pickle_protocol)

# YAML serialization tests
yaml_data = """\
float: 3.1415926500000002
int: 10
list: [george, jerry, elaine, cosmo]
string: The quick brown fox jumps over the lazy dog
unicode: "Th\\xE9 quick brown fox jumps over th\\xE9 lazy dog"
"""


msgpack_py_data = dict(py_data)
# msgpack only supports tuples
#msgpack_py_data['list'] = msgpack_py_data['list']
# Unicode chars are lost in transmit :(
msgpack_py_data['unicode'] = 'Th quick brown fox jumps over th lazy dog'
msgpack_data = b64decode(str_to_bytes("""\
haNpbnQKpWZsb2F0y0AJIftTyNTxpGxpc3SUpmdlb3JnZaVqZXJyeaZlbGFpbmWlY29zbW+mc3Rya\
W5n2gArVGhlIHF1aWNrIGJyb3duIGZveCBqdW1wcyBvdmVyIHRoZSBsYX<KEY>\
ApVGggcXVpY2sgYnJvd24gZm94IGp1bXBzIG92ZXIgdGggbGF6eSBkb2c=\
"""))


def say(m):
    sys.stderr.write('%s\n' % (m, ))


registry.register('testS', lambda s: s, lambda s: 'decoded',
                  'application/testS', 'utf-8')


class test_Serialization(TestCase):

    def test_disable(self):
        disabled = registry._disabled_content_types
        try:
            registry.disable('testS')
            self.assertIn('application/testS', disabled)
            disabled.clear()

            registry.disable('application/testS')
            self.assertIn('application/testS', disabled)
{'fraction_non_alphanumeric': 0.1201557935735151, 'fraction_numerical': 0.034274586173320354, 'mean_word_length': 4.0009737098344695, 'pattern_counts': {'":': 10, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3608099', 'n_tokens_mistral': 1907, 'n_tokens_neox': 1741, 'n_words': 438}