Run watcher tests with bytes() arguments
This commit is contained in:
ZyX
parent
7dd65a000b
commit
4e91f18908
|
|
@ -27,6 +27,67 @@ def clear_dir(dir):
|
|||
os.rmdir(os.path.join(root, d))
|
||||
|
||||
|
||||
def set_watcher_tests(l):
|
||||
byte_tests = (('bytes', True), ('unicode', False))
|
||||
|
||||
for btn, use_bytes in byte_tests:
|
||||
def test_inotify_file_watcher(self, use_bytes=use_bytes):
|
||||
try:
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='inotify')
|
||||
except INotifyError:
|
||||
raise SkipTest('This test is not suitable for a stat based file watcher')
|
||||
self.do_test_file_watcher(w, use_bytes)
|
||||
|
||||
def test_uv_file_watcher(self, use_bytes=use_bytes):
|
||||
raise SkipTest('Uv watcher tests are not stable')
|
||||
try:
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='uv')
|
||||
except UvNotFound:
|
||||
raise SkipTest('Pyuv is not available')
|
||||
self.do_test_file_watcher(w, use_bytes)
|
||||
|
||||
def test_inotify_tree_watcher(self, use_bytes=use_bytes):
|
||||
try:
|
||||
tw = create_tree_watcher(get_fallback_logger(), watcher_type='inotify')
|
||||
except INotifyError:
|
||||
raise SkipTest('INotify is not available')
|
||||
self.do_test_tree_watcher(tw, use_bytes)
|
||||
|
||||
def test_uv_tree_watcher(self, use_bytes=use_bytes):
|
||||
raise SkipTest('Uv watcher tests are not stable')
|
||||
try:
|
||||
tw = create_tree_watcher(get_fallback_logger(), 'uv')
|
||||
except UvNotFound:
|
||||
raise SkipTest('Pyuv is not available')
|
||||
self.do_test_tree_watcher(tw, use_bytes)
|
||||
|
||||
def test_inotify_file_watcher_is_watching(self, use_bytes=use_bytes):
|
||||
try:
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='inotify')
|
||||
except INotifyError:
|
||||
raise SkipTest('INotify is not available')
|
||||
self.do_test_file_watcher_is_watching(w, use_bytes)
|
||||
|
||||
def test_stat_file_watcher_is_watching(self, use_bytes=use_bytes):
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='stat')
|
||||
self.do_test_file_watcher_is_watching(w, use_bytes)
|
||||
|
||||
def test_uv_file_watcher_is_watching(self, use_bytes=use_bytes):
|
||||
try:
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='uv')
|
||||
except UvNotFound:
|
||||
raise SkipTest('Pyuv is not available')
|
||||
self.do_test_file_watcher_is_watching(w, use_bytes)
|
||||
|
||||
for wt in ('uv', 'inotify'):
|
||||
l['test_{0}_file_watcher_{1}'.format(wt, btn)] = locals()['test_{0}_file_watcher'.format(wt)]
|
||||
l['test_{0}_tree_watcher_{1}'.format(wt, btn)] = locals()['test_{0}_tree_watcher'.format(wt)]
|
||||
l['test_{0}_file_watcher_is_watching_{1}'.format(wt, btn)] = (
|
||||
locals()['test_{0}_file_watcher_is_watching'.format(wt)])
|
||||
l['test_{0}_file_watcher_is_watching_{1}'.format('stat', btn)] = (
|
||||
locals()['test_{0}_file_watcher_is_watching'.format('stat')])
|
||||
|
||||
|
||||
class TestFilesystemWatchers(TestCase):
|
||||
def do_test_for_change(self, watcher, path):
|
||||
st = monotonic()
|
||||
|
|
@ -36,21 +97,19 @@ class TestFilesystemWatchers(TestCase):
|
|||
sleep(0.1)
|
||||
self.fail('The change to {0} was not detected'.format(path))
|
||||
|
||||
def test_file_watcher(self):
|
||||
try:
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='inotify')
|
||||
except INotifyError:
|
||||
raise SkipTest('This test is not suitable for a stat based file watcher')
|
||||
return self.do_test_file_watcher(w)
|
||||
|
||||
def do_test_file_watcher(self, w):
|
||||
def do_test_file_watcher(self, w, use_bytes=False):
|
||||
try:
|
||||
f1, f2, f3 = map(lambda x: os.path.join(INOTIFY_DIR, 'file%d' % x), (1, 2, 3))
|
||||
ne = os.path.join(INOTIFY_DIR, 'notexists')
|
||||
if use_bytes:
|
||||
f1 = f1.encode('utf-8')
|
||||
f2 = f2.encode('utf-8')
|
||||
f3 = f3.encode('utf-8')
|
||||
ne = ne.encode('utf-8')
|
||||
with open(f1, 'wb'):
|
||||
with open(f2, 'wb'):
|
||||
with open(f3, 'wb'):
|
||||
pass
|
||||
ne = os.path.join(INOTIFY_DIR, 'notexists')
|
||||
self.assertRaises(OSError, w, ne)
|
||||
self.assertTrue(w(f1))
|
||||
self.assertTrue(w(f2))
|
||||
|
|
@ -87,87 +146,68 @@ class TestFilesystemWatchers(TestCase):
|
|||
finally:
|
||||
clear_dir(INOTIFY_DIR)
|
||||
|
||||
def test_uv_file_watcher(self):
|
||||
raise SkipTest('Uv watcher tests are not stable')
|
||||
def do_test_tree_watcher(self, tw, use_bytes=False):
|
||||
try:
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='uv')
|
||||
except UvNotFound:
|
||||
raise SkipTest('Pyuv is not available')
|
||||
return self.do_test_file_watcher(w)
|
||||
|
||||
def test_tree_watcher(self):
|
||||
tw = create_tree_watcher(get_fallback_logger())
|
||||
return self.do_test_tree_watcher(tw)
|
||||
|
||||
def do_test_tree_watcher(self, tw):
|
||||
try:
|
||||
subdir = os.path.join(INOTIFY_DIR, 'subdir')
|
||||
inotify_dir = INOTIFY_DIR
|
||||
subdir = os.path.join(inotify_dir, 'subdir')
|
||||
t1 = os.path.join(inotify_dir, 'tree1')
|
||||
ts1 = os.path.join(subdir, 'tree1')
|
||||
suffix = '1'
|
||||
f = os.path.join(subdir, 'f')
|
||||
if use_bytes:
|
||||
inotify_dir = inotify_dir.encode('utf-8')
|
||||
subdir = subdir.encode('utf-8')
|
||||
t1 = t1.encode('utf-8')
|
||||
ts1 = ts1.encode('utf-8')
|
||||
suffix = suffix.encode('utf-8')
|
||||
f = f.encode('utf-8')
|
||||
os.mkdir(subdir)
|
||||
try:
|
||||
if tw.watch(INOTIFY_DIR).is_dummy:
|
||||
if tw.watch(inotify_dir).is_dummy:
|
||||
raise SkipTest('No tree watcher available')
|
||||
except UvNotFound:
|
||||
raise SkipTest('Pyuv is not available')
|
||||
self.assertTrue(tw(INOTIFY_DIR))
|
||||
self.assertFalse(tw(INOTIFY_DIR))
|
||||
changed = partial(self.do_test_for_change, tw, INOTIFY_DIR)
|
||||
open(os.path.join(INOTIFY_DIR, 'tree1'), 'w').close()
|
||||
self.assertTrue(tw(inotify_dir))
|
||||
self.assertFalse(tw(inotify_dir))
|
||||
changed = partial(self.do_test_for_change, tw, inotify_dir)
|
||||
open(t1, 'w').close()
|
||||
changed()
|
||||
open(os.path.join(subdir, 'tree1'), 'w').close()
|
||||
open(ts1, 'w').close()
|
||||
changed()
|
||||
os.unlink(os.path.join(subdir, 'tree1'))
|
||||
os.unlink(ts1)
|
||||
changed()
|
||||
os.rmdir(subdir)
|
||||
changed()
|
||||
os.mkdir(subdir)
|
||||
changed()
|
||||
os.rename(subdir, subdir + '1')
|
||||
os.rename(subdir, subdir + suffix)
|
||||
changed()
|
||||
shutil.rmtree(subdir + '1')
|
||||
shutil.rmtree(subdir + suffix)
|
||||
changed()
|
||||
os.mkdir(subdir)
|
||||
f = os.path.join(subdir, 'f')
|
||||
open(f, 'w').close()
|
||||
changed()
|
||||
with open(f, 'a') as s:
|
||||
s.write(' ')
|
||||
changed()
|
||||
os.rename(f, f + '1')
|
||||
os.rename(f, f + suffix)
|
||||
changed()
|
||||
finally:
|
||||
clear_dir(INOTIFY_DIR)
|
||||
clear_dir(inotify_dir)
|
||||
|
||||
def test_uv_tree_watcher(self):
|
||||
raise SkipTest('Uv watcher tests are not stable')
|
||||
tw = create_tree_watcher(get_fallback_logger(), 'uv')
|
||||
return self.do_test_tree_watcher(tw)
|
||||
|
||||
def test_inotify_file_watcher_is_watching(self):
|
||||
try:
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='inotify')
|
||||
except INotifyError:
|
||||
raise SkipTest('INotify is not available')
|
||||
return self.do_test_file_watcher_is_watching(w)
|
||||
|
||||
def test_stat_file_watcher_is_watching(self):
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='stat')
|
||||
return self.do_test_file_watcher_is_watching(w)
|
||||
|
||||
def test_uv_file_watcher_is_watching(self):
|
||||
try:
|
||||
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='uv')
|
||||
except UvNotFound:
|
||||
raise SkipTest('Pyuv is not available')
|
||||
return self.do_test_file_watcher_is_watching(w)
|
||||
|
||||
def do_test_file_watcher_is_watching(self, w):
|
||||
def do_test_file_watcher_is_watching(self, w, use_bytes=False):
|
||||
try:
|
||||
f1, f2, f3 = map(lambda x: os.path.join(INOTIFY_DIR, 'file%d' % x), (1, 2, 3))
|
||||
ne = os.path.join(INOTIFY_DIR, 'notexists')
|
||||
if use_bytes:
|
||||
f1 = f1.encode('utf-8')
|
||||
f2 = f2.encode('utf-8')
|
||||
f3 = f3.encode('utf-8')
|
||||
ne = ne.encode('utf-8')
|
||||
with open(f1, 'wb'):
|
||||
with open(f2, 'wb'):
|
||||
with open(f3, 'wb'):
|
||||
pass
|
||||
ne = os.path.join(INOTIFY_DIR, 'notexists')
|
||||
self.assertRaises(OSError, w, ne)
|
||||
try:
|
||||
w(ne)
|
||||
|
|
@ -180,6 +220,8 @@ class TestFilesystemWatchers(TestCase):
|
|||
finally:
|
||||
clear_dir(INOTIFY_DIR)
|
||||
|
||||
set_watcher_tests(locals())
|
||||
|
||||
|
||||
old_cwd = None
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue