def test_import_gzip_reader():
 data = file(__file__).read()
 data_gz_sio = cStringIO.StringIO()
 gz = gzip.GzipFile(fileobj=data_gz_sio, mode='wb')
 gz.write(data)
 gz.close()
 data_gz = data_gz_sio.getvalue()
 old_peek_size = beeswax.create_table.IMPORT_PEEK_SIZE
 beeswax.create_table.IMPORT_PEEK_SIZE = (len(data_gz) - 1024)
 try:
  reader = beeswax.create_table.GzipFileReader
  lines = reader.readlines(data_gz_sio, 'utf-8')
  assert_true((lines is not None))
  lines_joined = '\n'.join(lines)
  assert_equal(data[:len(lines_joined)], lines_joined)
 finally:
  beeswax.create_table.IMPORT_PEEK_SIZE = old_peek_size

def test_import_gzip_reader ( ) : data = file ( __file__ ) . read ( ) data_gz_sio = cStringIO . StringIO ( ) gz = gzip . GzipFile ( fileobj = data_gz_sio , mode = 'wb' ) gz . write ( data ) gz . close ( ) data_gz = data_gz_sio . getvalue ( ) old_peek_size = beeswax . create_table . IMPORT_PEEK_SIZE beeswax . create_table . IMPORT_PEEK_SIZE = ( len ( data_gz ) - 1024 ) try : reader = beeswax . create_table . GzipFileReader lines = reader . readlines ( data_gz_sio , 'utf-8' ) assert_true ( ( lines is not None ) ) lines_joined = '\n' . join ( lines ) assert_equal ( data [ : len ( lines_joined ) ] , lines_joined ) finally : beeswax . create_table . IMPORT_PEEK_SIZE = old_peek_size
test the gzip reader in create table .