borg/attic/testsuite/__init__.py

118 lines
3.9 KiB
Python
Raw Normal View History

2013-07-26 11:18:57 +00:00
import filecmp
import os
2013-07-29 11:57:43 +00:00
import posix
2013-07-26 11:18:57 +00:00
import sys
2013-07-29 11:57:43 +00:00
import sysconfig
2013-07-26 11:18:57 +00:00
import time
2013-06-24 20:41:05 +00:00
import unittest
2013-07-29 11:57:43 +00:00
from attic.helpers import st_mtime_ns
2013-07-27 20:33:43 +00:00
from attic.xattr import get_all
try:
import llfuse
# Does this version of llfuse support ns precision?
have_fuse_mtime_ns = hasattr(llfuse.EntryAttributes, 'st_mtime_ns')
except ImportError:
have_fuse_mtime_ns = False
2013-07-29 11:57:43 +00:00
# The mtime get/set precison varies on different OS and Python versions
if 'HAVE_FUTIMENS' in getattr(posix, '_have_functions', []):
2013-07-29 11:57:43 +00:00
st_mtime_ns_round = 0
elif 'HAVE_UTIMES' in sysconfig.get_config_vars():
2013-07-29 19:09:31 +00:00
st_mtime_ns_round = -6
2013-07-29 11:57:43 +00:00
else:
st_mtime_ns_round = -9
2013-07-26 11:18:57 +00:00
has_mtime_ns = sys.version >= '3.3'
utime_supports_fd = os.utime in getattr(os, 'supports_fd', {})
2013-06-24 20:41:05 +00:00
2013-07-08 21:38:27 +00:00
class AtticTestCase(unittest.TestCase):
2013-06-24 20:41:05 +00:00
"""
"""
assert_in = unittest.TestCase.assertIn
assert_not_in = unittest.TestCase.assertNotIn
2013-06-24 20:41:05 +00:00
assert_equal = unittest.TestCase.assertEqual
assert_not_equal = unittest.TestCase.assertNotEqual
assert_raises = unittest.TestCase.assertRaises
2013-07-26 11:18:57 +00:00
def _get_xattrs(self, path):
try:
2013-07-27 20:33:43 +00:00
return get_all(path, follow_symlinks=False)
2013-07-26 11:18:57 +00:00
except EnvironmentError:
return {}
def assert_dirs_equal(self, dir1, dir2):
2013-07-26 11:18:57 +00:00
diff = filecmp.dircmp(dir1, dir2)
self._assert_dirs_equal_cmp(diff)
2013-07-26 11:18:57 +00:00
def _assert_dirs_equal_cmp(self, diff):
2013-07-26 11:18:57 +00:00
self.assert_equal(diff.left_only, [])
self.assert_equal(diff.right_only, [])
self.assert_equal(diff.diff_files, [])
self.assert_equal(diff.funny_files, [])
for filename in diff.common:
path1 = os.path.join(diff.left, filename)
path2 = os.path.join(diff.right, filename)
s1 = os.lstat(path1)
s2 = os.lstat(path2)
# Assume path2 is on FUSE if st_dev is different
fuse = s1.st_dev != s2.st_dev
2013-07-26 11:18:57 +00:00
attrs = ['st_mode', 'st_uid', 'st_gid', 'st_rdev']
if not fuse or not os.path.isdir(path1):
# dir nlink is always 1 on our fuse fileystem
attrs.append('st_nlink')
d1 = [filename] + [getattr(s1, a) for a in attrs]
d2 = [filename] + [getattr(s2, a) for a in attrs]
2013-07-29 11:57:43 +00:00
if not os.path.islink(path1) or utime_supports_fd:
# Older versions of llfuse does not support ns precision properly
if fuse and not have_fuse_mtime_ns:
2013-07-29 11:57:43 +00:00
d1.append(round(st_mtime_ns(s1), -4))
d2.append(round(st_mtime_ns(s2), -4))
d1.append(round(st_mtime_ns(s1), st_mtime_ns_round))
d2.append(round(st_mtime_ns(s2), st_mtime_ns_round))
2013-07-26 11:18:57 +00:00
d1.append(self._get_xattrs(path1))
d2.append(self._get_xattrs(path2))
self.assert_equal(d1, d2)
for sub_diff in diff.subdirs.values():
self._assert_dirs_equal_cmp(sub_diff)
2013-07-26 11:18:57 +00:00
def wait_for_mount(self, path, timeout=5):
"""Wait until a filesystem is mounted on `path`
"""
timeout += time.time()
while timeout > time.time():
2013-07-26 11:23:33 +00:00
if os.path.ismount(path):
2013-07-26 11:18:57 +00:00
return
time.sleep(.1)
raise Exception('wait_for_mount(%s) timeout' % path)
2013-06-24 20:41:05 +00:00
def get_tests(suite):
"""Generates a sequence of tests from a test suite
"""
for item in suite:
try:
# TODO: This could be "yield from..." with Python 3.3+
for i in get_tests(item):
yield i
except TypeError:
yield item
class TestLoader(unittest.TestLoader):
"""A customzied test loader that properly detects and filters our test cases
"""
2013-07-27 20:33:43 +00:00
2013-06-24 20:41:05 +00:00
def loadTestsFromName(self, pattern, module=None):
2013-07-08 21:38:27 +00:00
suite = self.discover('attic.testsuite', '*.py')
2013-06-24 20:41:05 +00:00
tests = unittest.TestSuite()
for test in get_tests(suite):
if pattern.lower() in test.id().lower():
tests.addTest(test)
return tests