191 lines
7.6 KiB
Python
191 lines
7.6 KiB
Python
import os.path
|
|
import unittest
|
|
import tempfile
|
|
import shutil
|
|
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(unittest.TestCase):
|
|
|
|
_tmpdir=None
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
# Create tmp dir to write to and check
|
|
cls._tmpdir=tempfile.mkdtemp()
|
|
if (not os.path.isdir(cls._tmpdir)):
|
|
raise Exception("Failed to create tempdir to use for dump tests")
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
# Cleanup
|
|
if (not os.path.isdir(cls._tmpdir)):
|
|
raise Exception("Ooops, no tempdir (%s) to clean up?" % (tmpdir))
|
|
shutil.rmtree(cls._tmpdir)
|
|
|
|
@property
|
|
def tmpdir(self):
|
|
# read-only access to _tmpdir, just in case... The rmtree scares me
|
|
#
|
|
# FIXME - Hack to work on python2.6 where setUpClass is not called, will
|
|
# FIXME - not have proper tidy as tearDownClass will not be called.
|
|
# FIXME - Remove when 2.6 no longer supported
|
|
if (not self._tmpdir and sys.version_info < (2,7)):
|
|
self.setUpClass()
|
|
return(self._tmpdir)
|
|
|
|
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)
|
