<s> import platform <EOL> import pytest <EOL> import unittest2 <EOL> from uuid import uuid4 <EOL> from pykafka . utils import compression <EOL> class CompressionTests ( unittest2 . TestCase ) : <EOL> """<STR_LIT>""" <EOL> text = b"<STR_LIT>" <EOL> def test_gzip ( self ) : <EOL> encoded = compression . encode_gzip ( self . text ) <EOL> self . assertNotEqual ( self . text , encoded ) <EOL> decoded = compression . decode_gzip ( encoded ) <EOL> self . assertEqual ( self . text , decoded ) <EOL> def test_snappy ( self ) : <EOL> encoded = compression . encode_snappy ( self . text ) <EOL> self . assertNotEqual ( self . text , encoded ) <EOL> decoded = compression . decode_snappy ( encoded ) <EOL> self . assertEqual ( self . text , decoded ) <EOL> def test_snappy_xerial ( self ) : <EOL> encoded = compression . encode_snappy ( self . text , xerial_compatible = True ) <EOL> self . assertNotEqual ( self . text , encoded ) <EOL> decoded = compression . decode_snappy ( encoded ) <EOL> self . assertEqual ( self . text , decoded ) <EOL> @ pytest . mark . skipif ( platform . python_implementation ( ) == "<STR_LIT>" , <EOL> reason = "<STR_LIT>" ) <EOL> def test_snappy_large_payload ( self ) : <EOL> payload = b'<STR_LIT>' . join ( [ uuid4 ( ) . bytes for i in range ( <NUM_LIT:10> ) ] ) <EOL> c = compression . encode_snappy ( payload ) <EOL> self . assertEqual ( compression . decode_snappy ( c ) , payload ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest2 . main ( ) </s>
