<s> """<STR_LIT>""" <EOL> import random <EOL> import zipfile <EOL> from hashlib import md5 <EOL> from twisted . python import zipstream , filepath <EOL> from twisted . trial import unittest <EOL> class FileEntryMixin : <EOL> """<STR_LIT>""" <EOL> def getFileEntry ( self , contents ) : <EOL> """<STR_LIT>""" <EOL> filename = self . mktemp ( ) <EOL> z = zipfile . ZipFile ( filename , '<STR_LIT:w>' , self . compression ) <EOL> z . writestr ( '<STR_LIT:content>' , contents ) <EOL> z . close ( ) <EOL> z = zipstream . ChunkingZipFile ( filename , '<STR_LIT:r>' ) <EOL> return z . readfile ( '<STR_LIT:content>' ) <EOL> def test_isatty ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . getFileEntry ( '<STR_LIT>' ) . isatty ( ) , False ) <EOL> def test_closed ( self ) : <EOL> """<STR_LIT>""" <EOL> fileEntry = self . getFileEntry ( '<STR_LIT>' ) <EOL> self . assertEqual ( fileEntry . closed , False ) <EOL> fileEntry . close ( ) <EOL> self . assertEqual ( fileEntry . closed , True ) <EOL> def test_readline ( self ) : <EOL> """<STR_LIT>""" <EOL> fileEntry = self . getFileEntry ( '<STR_LIT>' ) <EOL> self . assertEqual ( fileEntry . readline ( ) , '<STR_LIT>' ) <EOL> self . assertEqual ( fileEntry . readline ( ) , '<STR_LIT>' ) <EOL> self . assertEqual ( fileEntry . readline ( ) , '<STR_LIT>' ) <EOL> def test_next ( self ) : <EOL> """<STR_LIT>""" <EOL> fileEntry = self . getFileEntry ( '<STR_LIT>' ) <EOL> self . assertEqual ( fileEntry . next ( ) , '<STR_LIT>' ) <EOL> self . assertEqual ( fileEntry . next ( ) , '<STR_LIT>' ) <EOL> self . assertRaises ( StopIteration , fileEntry . next ) <EOL> def test_readlines ( self ) : <EOL> """<STR_LIT>""" <EOL> fileEntry = self . getFileEntry ( '<STR_LIT>' ) <EOL> self . assertEqual ( fileEntry . readlines ( ) , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_iteration ( self ) : <EOL> """<STR_LIT>""" <EOL> fileEntry = self . getFileEntry ( '<STR_LIT>' ) <EOL> self . assertIdentical ( iter ( fileEntry ) , fileEntry ) <EOL> self . assertIdentical ( fileEntry . xreadlines ( ) , fileEntry ) <EOL> def test_readWhole ( self ) : <EOL> """<STR_LIT>""" <EOL> contents = "<STR_LIT>" <EOL> entry = self . getFileEntry ( contents ) <EOL> self . assertEqual ( entry . read ( ) , contents ) <EOL> def test_readPartial ( self ) : <EOL> """<STR_LIT>""" <EOL> contents = "<STR_LIT>" <EOL> entry = self . getFileEntry ( contents ) <EOL> one = entry . read ( <NUM_LIT:4> ) <EOL> two = entry . read ( <NUM_LIT:200> ) <EOL> self . assertEqual ( one , "<STR_LIT>" ) <EOL> self . assertEqual ( two , "<STR_LIT>" ) <EOL> def test_tell ( self ) : <EOL> """<STR_LIT>""" <EOL> contents = "<STR_LIT:x>" * <NUM_LIT:100> <EOL> entry = self . getFileEntry ( contents ) <EOL> entry . read ( <NUM_LIT:2> ) <EOL> self . assertEqual ( entry . tell ( ) , <NUM_LIT:2> ) <EOL> entry . read ( <NUM_LIT:4> ) <EOL> self . assertEqual ( entry . tell ( ) , <NUM_LIT:6> ) <EOL> class DeflatedZipFileEntryTests ( FileEntryMixin , unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> compression = zipfile . ZIP_DEFLATED <EOL> class ZipFileEntryTests ( FileEntryMixin , unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> compression = zipfile . ZIP_STORED <EOL> class ZipstreamTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . testdir = filepath . FilePath ( self . mktemp ( ) ) <EOL> self . testdir . makedirs ( ) <EOL> self . unzipdir = self . testdir . child ( '<STR_LIT>' ) <EOL> self . unzipdir . makedirs ( ) <EOL> def makeZipFile ( self , contents , directory = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> zpfilename = self . testdir . child ( '<STR_LIT>' ) . path <EOL> zpfile = zipfile . ZipFile ( zpfilename , '<STR_LIT:w>' ) <EOL> for i , content in enumerate ( contents ) : <EOL> filename = str ( i ) <EOL> if directory : <EOL> filename = directory + "<STR_LIT:/>" + filename <EOL> zpfile . writestr ( filename , content ) <EOL> zpfile . close ( ) <EOL> return zpfilename <EOL> def test_invalidMode ( self ) : <EOL> """<STR_LIT>""" <EOL> czf = zipstream . ChunkingZipFile ( self . mktemp ( ) , "<STR_LIT:w>" ) <EOL> self . assertRaises ( RuntimeError , czf . readfile , "<STR_LIT>" ) <EOL> def test_closedArchive ( self ) : <EOL> """<STR_LIT>""" <EOL> czf = zipstream . ChunkingZipFile ( self . makeZipFile ( [ "<STR_LIT>" ] ) , "<STR_LIT:r>" ) <EOL> czf . close ( ) <EOL> self . assertRaises ( RuntimeError , czf . readfile , "<STR_LIT>" ) <EOL> def test_invalidHeader ( self ) : <EOL> """<STR_LIT>""" <EOL> fn = self . makeZipFile ( [ "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> zf = zipfile . ZipFile ( fn , "<STR_LIT:r>" ) <EOL> zeroOffset = zf . getinfo ( "<STR_LIT:0>" ) . header_offset <EOL> zf . close ( ) <EOL> scribble = file ( fn , "<STR_LIT>" ) <EOL> scribble . seek ( zeroOffset , <NUM_LIT:0> ) <EOL> scribble . write ( chr ( <NUM_LIT:0> ) * <NUM_LIT:4> ) <EOL> scribble . close ( ) <EOL> czf = zipstream . ChunkingZipFile ( fn ) <EOL> self . assertRaises ( zipfile . BadZipfile , czf . readfile , "<STR_LIT:0>" ) <EOL> self . assertEqual ( czf . readfile ( "<STR_LIT:1>" ) . read ( ) , "<STR_LIT>" ) <EOL> def test_filenameMismatch ( self ) : <EOL> """<STR_LIT>""" <EOL> fn = self . makeZipFile ( [ "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> zf = zipfile . ZipFile ( fn , "<STR_LIT:r>" ) <EOL> info = zf . getinfo ( "<STR_LIT:0>" ) <EOL> info . filename = "<STR_LIT>" <EOL> zf . close ( ) <EOL> scribble = file ( fn , "<STR_LIT>" ) <EOL> scribble . seek ( info . header_offset , <NUM_LIT:0> ) <EOL> scribble . write ( info . FileHeader ( ) ) <EOL> scribble . close ( ) <EOL> czf = zipstream . ChunkingZipFile ( fn ) <EOL> self . assertRaises ( zipfile . BadZipfile , czf . readfile , "<STR_LIT:0>" ) <EOL> self . assertEqual ( czf . readfile ( "<STR_LIT:1>" ) . read ( ) , "<STR_LIT>" ) <EOL> def test_unsupportedCompression ( self ) : <EOL> """<STR_LIT>""" <EOL> fn = self . mktemp ( ) <EOL> zf = zipfile . ZipFile ( fn , "<STR_LIT:w>" ) <EOL> zi = zipfile . ZipInfo ( "<STR_LIT:0>" ) <EOL> zf . writestr ( zi , "<STR_LIT>" ) <EOL> zi . compress_type = <NUM_LIT> <EOL> zf . close ( ) <EOL> czf = zipstream . ChunkingZipFile ( fn ) <EOL> self . assertRaises ( zipfile . BadZipfile , czf . readfile , "<STR_LIT:0>" ) <EOL> def test_extraData ( self ) : <EOL> """<STR_LIT>""" <EOL> fn = self . mktemp ( ) <EOL> zf = zipfile . ZipFile ( fn , '<STR_LIT:w>' ) <EOL> zi = zipfile . ZipInfo ( "<STR_LIT:0>" ) <EOL> zi . extra = "<STR_LIT>" <EOL> zf . writestr ( zi , "<STR_LIT>" ) <EOL> zf . close ( ) <EOL> czf = zipstream . ChunkingZipFile ( fn ) <EOL> self . assertEqual ( czf . readfile ( "<STR_LIT:0>" ) . read ( ) , "<STR_LIT>" ) <EOL> def test_unzipIterChunky ( self ) : <EOL> """<STR_LIT>""" <EOL> numfiles = <NUM_LIT:10> <EOL> contents = [ '<STR_LIT>' % i for i in range ( numfiles ) ] <EOL> zpfilename = self . makeZipFile ( contents ) <EOL> list ( zipstream . unzipIterChunky ( zpfilename , self . unzipdir . path ) ) <EOL> self . assertEqual ( <EOL> set ( self . unzipdir . listdir ( ) ) , <EOL> set ( map ( str , range ( numfiles ) ) ) ) <EOL> for child in self . unzipdir . children ( ) : <EOL> num = int ( child . basename ( ) ) <EOL> self . assertEqual ( child . getContent ( ) , contents [ num ] ) <EOL> def test_unzipIterChunkyDirectory ( self ) : <EOL> """<STR_LIT>""" <EOL> numfiles = <NUM_LIT:10> <EOL> contents = [ '<STR_LIT>' % i for i in range ( numfiles ) ] <EOL> zpfilename = self . makeZipFile ( contents , '<STR_LIT:foo>' ) <EOL> list ( zipstream . unzipIterChunky ( zpfilename , self . unzipdir . path ) ) <EOL> self . assertEqual ( <EOL> set ( self . unzipdir . child ( '<STR_LIT:foo>' ) . listdir ( ) ) , <EOL> set ( map ( str , range ( numfiles ) ) ) ) <EOL> for child in self . unzipdir . child ( '<STR_LIT:foo>' ) . children ( ) : <EOL> num = int ( child . basename ( ) ) <EOL> self . assertEqual ( child . getContent ( ) , contents [ num ] ) <EOL> def _unzipIterChunkyTest ( self , compression , chunksize , lower , upper ) : <EOL> """<STR_LIT>""" <EOL> junk = '<STR_LIT:U+0020>' . join ( [ str ( random . random ( ) ) for n in xrange ( <NUM_LIT:1000> ) ] ) <EOL> junkmd5 = md5 ( junk ) . hexdigest ( ) <EOL> tempdir = filepath . FilePath ( self . mktemp ( ) ) <EOL> tempdir . makedirs ( ) <EOL> zfpath = tempdir . child ( '<STR_LIT>' ) . path <EOL> self . _makebigfile ( zfpath , compression , junk ) <EOL> uziter = zipstream . unzipIterChunky ( zfpath , tempdir . path , <EOL> chunksize = chunksize ) <EOL> r = uziter . next ( ) <EOL> approx = lower < r < upper <EOL> self . assertTrue ( approx ) <EOL> for r in uziter : <EOL> pass <EOL> self . assertEqual ( r , <NUM_LIT:0> ) <EOL> newmd5 = md5 ( <EOL> tempdir . child ( "<STR_LIT>" ) . open ( ) . read ( ) ) . hexdigest ( ) <EOL> self . assertEqual ( newmd5 , junkmd5 ) <EOL> def test_unzipIterChunkyStored ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _unzipIterChunkyTest ( zipfile . ZIP_STORED , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> def test_chunkyDeflated ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _unzipIterChunkyTest ( zipfile . ZIP_DEFLATED , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> def _makebigfile ( self , filename , compression , junk ) : <EOL> """<STR_LIT>""" <EOL> zf = zipfile . ZipFile ( filename , '<STR_LIT:w>' , compression ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> fn = '<STR_LIT>' % i <EOL> zf . writestr ( fn , "<STR_LIT>" ) <EOL> zf . writestr ( '<STR_LIT>' , junk ) <EOL> zf . close ( ) </s>
