164 lines
6.7 KiB
Python
164 lines
6.7 KiB
Python
from tests.testcase_with_tmpdir import TestCase
|
|
|
|
import os.path
|
|
import unittest
|
|
import sys
|
|
import zipfile
|
|
|
|
from resync.dump import Dump, DumpError
|
|
from resync.resource_list import ResourceList
|
|
from resync.change_list import ChangeList
|
|
from resync.resource import Resource
|
|
from resync.resource_dump_manifest import ResourceDumpManifest
|
|
from resync.change_dump_manifest import ChangeDumpManifest
|
|
|
|
class TestDump(TestCase):
|
|
|
|
def test00_dump_zip_resource_list(self):
|
|
rl=ResourceDumpManifest()
|
|
rl.add( Resource('http://ex.org/a', length=7, path='tests/testdata/a') )
|
|
rl.add( Resource('http://ex.org/b', length=21, path='tests/testdata/b') )
|
|
d=Dump()
|
|
zipf=os.path.join(self.tmpdir,"test00_dump.zip")
|
|
d.write_zip(resources=rl,dumpfile=zipf) # named args
|
|
self.assertTrue( os.path.exists(zipf) )
|
|
self.assertTrue( zipfile.is_zipfile(zipf) )
|
|
zo=zipfile.ZipFile(zipf,'r')
|
|
self.assertEqual( len(zo.namelist()), 3 )
|
|
zo.close()
|
|
os.unlink(zipf)
|
|
|
|
def test01_dump_zip_change_list(self):
|
|
cl=ChangeDumpManifest()
|
|
cl.add( Resource('http://ex.org/a', length=7, path='tests/testdata/a', change="updated") )
|
|
cl.add( Resource('http://ex.org/b', length=21, path='tests/testdata/b', change="updated") )
|
|
d=Dump()
|
|
zipf=os.path.join(self.tmpdir,"test01_dump.zip")
|
|
d.write_zip(cl,zipf) # positional args
|
|
self.assertTrue( os.path.exists(zipf) )
|
|
self.assertTrue( zipfile.is_zipfile(zipf) )
|
|
zo=zipfile.ZipFile(zipf,'r')
|
|
self.assertEqual( len(zo.namelist()), 3 )
|
|
zo.close()
|
|
os.unlink(zipf)
|
|
|
|
def test02_dump_check_files(self):
|
|
cl=ChangeList()
|
|
cl.add( Resource('http://ex.org/a', length=7, path='tests/testdata/a', change="updated") )
|
|
cl.add( Resource('http://ex.org/b', length=21, path='tests/testdata/b', change="updated") )
|
|
d=Dump(resources=cl)
|
|
self.assertTrue(d.check_files())
|
|
self.assertEqual(d.total_size, 28)
|
|
|
|
def test03_dump_multi_file_max_size(self):
|
|
rl=ResourceList()
|
|
for letter in map(chr,range(ord('a'),ord('l')+1)):
|
|
uri='http://ex.org/%s' % (letter)
|
|
fname='tests/testdata/a_to_z/%s' % (letter)
|
|
rl.add( Resource(uri, path=fname) )
|
|
self.assertEqual( len(rl), 12 )
|
|
#d=Dump(rl)
|
|
#tmpdir=tempfile.mkdtemp()
|
|
#tmpbase=os.path.join(tmpdir,'base')
|
|
#d.max_size=2000 # start new zip after size exceeds 2000 bytes
|
|
#n=d.write(tmpbase)
|
|
#self.assertEqual( n, 2, 'expect to write 2 dump files' )
|
|
#
|
|
# Now repeat with large size limit but small number of files limit
|
|
d2=Dump(rl)
|
|
tmpbase=os.path.join(self.tmpdir,'test03_')
|
|
d2.max_files=4
|
|
n=d2.write(tmpbase)
|
|
self.assertEqual( n, 3, 'expect to write 3 dump files' )
|
|
self.assertTrue( os.path.isfile(tmpbase+'00000.zip') )
|
|
self.assertTrue( os.path.isfile(tmpbase+'00001.zip') )
|
|
self.assertTrue( os.path.isfile(tmpbase+'00002.zip') )
|
|
# Look at the first file in detail
|
|
zipf=tmpbase+'00000.zip'
|
|
zo=zipfile.ZipFile(zipf,'r')
|
|
self.assertEqual( zo.namelist(), ['manifest.xml','a','b','c','d'] )
|
|
#self.assertEqual( zo.getinfo('manifest.xml').file_size, 470 )
|
|
self.assertEqual( zo.getinfo('a').file_size, 9 )
|
|
self.assertEqual( zo.getinfo('b').file_size, 1116 )
|
|
self.assertEqual( zo.getinfo('c').file_size, 32 )
|
|
self.assertEqual( zo.getinfo('d').file_size, 13 )
|
|
zo.close()
|
|
os.unlink(zipf)
|
|
# Check second and third files have expected contents
|
|
zipf=tmpbase+'00001.zip'
|
|
zo=zipfile.ZipFile(zipf,'r')
|
|
self.assertEqual( zo.namelist(), ['manifest.xml','e','f','g','h'] )
|
|
zo.close()
|
|
os.unlink(zipf)
|
|
zipf=tmpbase+'00002.zip'
|
|
zo=zipfile.ZipFile(zipf,'r')
|
|
self.assertEqual( zo.namelist(), ['manifest.xml','i','j','k','l'] )
|
|
zo.close()
|
|
os.unlink(zipf)
|
|
|
|
def test04_dump_multi_file_max_size(self):
|
|
rl=ResourceList()
|
|
for letter in map(chr,range(ord('a'),ord('l')+1)):
|
|
uri='http://ex.org/%s' % (letter)
|
|
fname='tests/testdata/a_to_z/%s' % (letter)
|
|
rl.add( Resource(uri, path=fname) )
|
|
self.assertEqual( len(rl), 12 )
|
|
d2=Dump(rl)
|
|
tmpbase=os.path.join(self.tmpdir,'test0f_')
|
|
d2.max_size=2000
|
|
n=d2.write(tmpbase)
|
|
self.assertEqual( n, 2, 'expect to write 2 dump files' )
|
|
self.assertTrue( os.path.isfile(tmpbase+'00000.zip') )
|
|
self.assertTrue( os.path.isfile(tmpbase+'00001.zip') )
|
|
# Look at the first file in detail
|
|
zipf=tmpbase+'00000.zip'
|
|
zo=zipfile.ZipFile(zipf,'r')
|
|
self.assertEqual( zo.namelist(), ['manifest.xml','a','b','c','d','e','f'] )
|
|
#self.assertEqual( zo.getinfo('manifest.xml').file_size, 470 )
|
|
self.assertEqual( zo.getinfo('a').file_size, 9 )
|
|
self.assertEqual( zo.getinfo('b').file_size, 1116 )
|
|
self.assertEqual( zo.getinfo('c').file_size, 32 )
|
|
self.assertEqual( zo.getinfo('d').file_size, 13 )
|
|
self.assertEqual( zo.getinfo('e').file_size, 20 )
|
|
self.assertEqual( zo.getinfo('f').file_size, 1625 )
|
|
zo.close()
|
|
os.unlink(zipf)
|
|
# Check second and third files have expected contents
|
|
zipf=tmpbase+'00001.zip'
|
|
zo=zipfile.ZipFile(zipf,'r')
|
|
self.assertEqual( zo.namelist(), ['manifest.xml','g','h','i','j','k','l'] )
|
|
zo.close()
|
|
os.unlink(zipf)
|
|
|
|
def test10_no_path(self):
|
|
rl=ResourceList()
|
|
rl.add( Resource('http://ex.org/a', length=7, path='tests/testdata/a') )
|
|
rl.add( Resource('http://ex.org/b', length=21 ) )
|
|
d=Dump(rl)
|
|
self.assertRaises( DumpError, d.check_files )
|
|
|
|
def test11_bad_size(self):
|
|
rl=ResourceList()
|
|
rl.add( Resource('http://ex.org/a', length=9999, path='tests/testdata/a') )
|
|
d=Dump(rl)
|
|
self.assertTrue( d.check_files(check_length=False) )
|
|
self.assertRaises( DumpError, d.check_files )
|
|
|
|
def test10_no_path(self):
|
|
rl=ResourceList()
|
|
rl.add( Resource('http://ex.org/a', length=7, path='tests/testdata/a') )
|
|
rl.add( Resource('http://ex.org/b', length=21 ) )
|
|
d=Dump(rl)
|
|
self.assertRaises( DumpError, d.check_files )
|
|
|
|
def test11_bad_size(self):
|
|
rl=ResourceList()
|
|
rl.add( Resource('http://ex.org/a', length=9999, path='tests/testdata/a') )
|
|
d=Dump(rl)
|
|
self.assertTrue( d.check_files(check_length=False) )
|
|
self.assertRaises( DumpError, d.check_files )
|
|
|
|
if __name__ == '__main__':
|
|
suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestDump)
|
|
unittest.TextTestRunner(verbosity=2).run(suite)
|
