borg/darc/test.py

119 lines
4.1 KiB
Python
Raw Normal View History

2010-10-31 19:12:32 +00:00
import filecmp
2010-10-16 09:45:36 +00:00
import os
2010-10-30 11:44:25 +00:00
from StringIO import StringIO
import sys
2010-10-16 09:45:36 +00:00
import shutil
import tempfile
import unittest
2010-10-31 19:31:56 +00:00
from xattr import xattr, XATTR_NOFOLLOW
2010-10-19 19:08:42 +00:00
2010-10-26 19:25:25 +00:00
from . import store
2010-10-31 19:12:32 +00:00
from .archiver import Archiver
2010-10-16 09:45:36 +00:00
class Test(unittest.TestCase):
def setUp(self):
self.archiver = Archiver()
self.tmpdir = tempfile.mkdtemp()
self.store_path = os.path.join(self.tmpdir, 'store')
2010-10-31 19:12:32 +00:00
self.input_path = os.path.join(self.tmpdir, 'input')
self.output_path = os.path.join(self.tmpdir, 'output')
os.mkdir(self.input_path)
os.mkdir(self.output_path)
os.chdir(self.tmpdir)
self.keychain = '/tmp/_test_dedupstore.keychain'
if not os.path.exists(self.keychain):
2010-10-31 20:55:09 +00:00
self.darc('init-keychain')
2010-10-27 18:12:40 +00:00
self.darc('init', self.store_path)
2010-10-16 09:45:36 +00:00
def tearDown(self):
shutil.rmtree(self.tmpdir)
2010-10-27 18:12:40 +00:00
def darc(self, *args, **kwargs):
2010-10-16 09:45:36 +00:00
exit_code = kwargs.get('exit_code', 0)
2010-10-25 18:22:20 +00:00
args = ['--keychain', self.keychain] + list(args)
2010-10-30 11:44:25 +00:00
try:
stdout, stderr = sys.stdout, sys.stderr
output = StringIO()
sys.stdout = sys.stderr = output
2010-10-31 19:12:32 +00:00
ret = self.archiver.run(args)
sys.stdout, sys.stderr = stdout, stderr
if ret != exit_code:
print output.getvalue()
self.assertEqual(exit_code, ret)
2010-10-30 11:44:25 +00:00
return output.getvalue()
finally:
sys.stdout, sys.stderr = stdout, stderr
2010-10-16 09:45:36 +00:00
def create_src_archive(self, name):
src_dir = os.path.join(os.getcwd(), os.path.dirname(__file__))
2010-10-27 18:12:40 +00:00
self.darc('create', self.store_path + '::' + name, src_dir)
2010-10-16 09:45:36 +00:00
2010-10-31 19:12:32 +00:00
def create_regual_file(self, name, size=0):
filename = os.path.join(self.input_path, name)
if not os.path.exists(os.path.dirname(filename)):
os.makedirs(os.path.dirname(filename))
with open(filename, 'wb') as fd:
fd.write('X' * size)
2010-10-31 19:31:56 +00:00
def get_xattrs(self, path):
try:
return dict(xattr(path, XATTR_NOFOLLOW))
except IOError:
return {}
2010-10-31 19:12:32 +00:00
def diff_dirs(self, dir1, dir2):
diff = filecmp.dircmp(dir1, dir2)
self.assertEqual(diff.left_only, [])
self.assertEqual(diff.right_only, [])
self.assertEqual(diff.diff_files, [])
for filename in diff.common:
2010-10-31 19:31:56 +00:00
path1 = os.path.join(dir1, filename)
path2 = os.path.join(dir2, filename)
s1 = os.lstat(path1)
s2 = os.lstat(path2)
2010-10-31 19:12:32 +00:00
attrs = ['st_mode', 'st_uid', 'st_gid']
# We can't restore symlink atime/mtime right now
2010-10-31 19:31:56 +00:00
if not os.path.islink(path1):
2010-10-31 19:12:32 +00:00
attrs.append('st_mtime')
d1 = [filename] + [getattr(s1, a) for a in attrs]
d2 = [filename] + [getattr(s2, a) for a in attrs]
2010-10-31 19:31:56 +00:00
d1.append(self.get_xattrs(path1))
d2.append(self.get_xattrs(path2))
2010-10-31 19:12:32 +00:00
self.assertEqual(d1, d2)
2010-10-16 09:45:36 +00:00
def test_basic_functionality(self):
2010-10-31 19:12:32 +00:00
self.create_regual_file('file1', size=1024*80)
self.create_regual_file('dir2/file2', size=1024*80)
2010-10-31 19:31:56 +00:00
x = xattr(os.path.join(self.input_path, 'file1'))
2010-11-01 22:06:56 +00:00
x.set('user.foo', 'bar')
2010-10-31 19:12:32 +00:00
os.symlink('somewhere', os.path.join(self.input_path, 'link1'))
os.mkfifo(os.path.join(self.input_path, 'fifo1'))
self.darc('create', self.store_path + '::test', 'input')
self.darc('extract', self.store_path + '::test', 'output')
self.diff_dirs('input', 'output/input')
2010-10-16 09:45:36 +00:00
def test_corrupted_store(self):
self.create_src_archive('test')
2010-10-27 18:12:40 +00:00
self.darc('verify', self.store_path + '::test')
fd = open(os.path.join(self.tmpdir, 'store', 'bands', '0', '0'), 'r+')
2010-10-31 19:31:56 +00:00
fd.seek(100)
2010-10-16 09:45:36 +00:00
fd.write('X')
fd.close()
2010-10-27 18:12:40 +00:00
self.darc('verify', self.store_path + '::test', exit_code=1)
2010-10-16 09:45:36 +00:00
2010-10-31 20:55:09 +00:00
def test_keychain(self):
keychain = os.path.join(self.tmpdir, 'keychain')
self.darc('-k', keychain, 'init-keychain')
2010-10-19 19:08:42 +00:00
def suite():
suite = unittest.TestSuite()
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(Test))
2010-10-26 19:25:25 +00:00
suite.addTest(store.suite())
2010-10-19 19:08:42 +00:00
return suite
2010-10-16 09:45:36 +00:00
if __name__ == '__main__':
2010-10-19 19:08:42 +00:00
unittest.TextTestRunner(verbosity=2).run(suite())