Add tests for libuv-based watcher

Four possible results of running tests (first three are errors):

- “The change to inotify/file2 was not detected” on line 84: most common
- “Spurious change detected” at line 82 (uncommon, usually fixed by sleeping
  before running test in do_test_for_change)
- “The change to inotify was not detected” in tree watcher test (e.g. from line
  131) (very rare)
- All OK.
This commit is contained in:
ZyX 2014-06-29 11:17:23 +04:00
parent ea3cd2c1c7
commit 0ee5293e1a

View File

@ -2,6 +2,7 @@
from __future__ import absolute_import, unicode_literals, print_function, division from __future__ import absolute_import, unicode_literals, print_function, division
from powerline.lib.watcher import create_file_watcher, create_tree_watcher, INotifyError from powerline.lib.watcher import create_file_watcher, create_tree_watcher, INotifyError
from powerline.lib.watcher.uv import UvNotFound
from powerline import get_fallback_logger from powerline import get_fallback_logger
from powerline.lib.monotonic import monotonic from powerline.lib.monotonic import monotonic
@ -16,6 +17,14 @@ from tests import TestCase, SkipTest
INOTIFY_DIR = 'inotify' + os.environ.get('PYTHON', '') INOTIFY_DIR = 'inotify' + os.environ.get('PYTHON', '')
def clear_dir(dir):
for root, dirs, files in list(os.walk(dir, topdown=False)):
for f in files:
os.remove(os.path.join(root, f))
for d in dirs:
os.rmdir(os.path.join(root, d))
class TestFilesystemWatchers(TestCase): class TestFilesystemWatchers(TestCase):
def do_test_for_change(self, watcher, path): def do_test_for_change(self, watcher, path):
st = monotonic() st = monotonic()
@ -30,6 +39,10 @@ class TestFilesystemWatchers(TestCase):
w = create_file_watcher(pl=get_fallback_logger(), watcher_type='inotify') w = create_file_watcher(pl=get_fallback_logger(), watcher_type='inotify')
except INotifyError: except INotifyError:
raise SkipTest('This test is not suitable for a stat based file watcher') 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):
try:
f1, f2, f3 = map(lambda x: os.path.join(INOTIFY_DIR, 'file%d' % x), (1, 2, 3)) f1, f2, f3 = map(lambda x: os.path.join(INOTIFY_DIR, 'file%d' % x), (1, 2, 3))
with open(f1, 'wb'): with open(f1, 'wb'):
with open(f2, 'wb'): with open(f2, 'wb'):
@ -69,13 +82,29 @@ class TestFilesystemWatchers(TestCase):
self.assertFalse(w(f2), 'Spurious change detected') self.assertFalse(w(f2), 'Spurious change detected')
os.utime(f2, None) os.utime(f2, None)
self.do_test_for_change(w, f2) self.do_test_for_change(w, f2)
finally:
clear_dir(INOTIFY_DIR)
def test_uv_file_watcher(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(w)
def test_tree_watcher(self): def test_tree_watcher(self):
tw = create_tree_watcher(get_fallback_logger()) 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') subdir = os.path.join(INOTIFY_DIR, 'subdir')
os.mkdir(subdir) os.mkdir(subdir)
try:
if tw.watch(INOTIFY_DIR).is_dummy: if tw.watch(INOTIFY_DIR).is_dummy:
raise SkipTest('No tree watcher available') raise SkipTest('No tree watcher available')
except UvNotFound:
raise SkipTest('Pyuv is not available')
self.assertTrue(tw(INOTIFY_DIR)) self.assertTrue(tw(INOTIFY_DIR))
self.assertFalse(tw(INOTIFY_DIR)) self.assertFalse(tw(INOTIFY_DIR))
changed = partial(self.do_test_for_change, tw, INOTIFY_DIR) changed = partial(self.do_test_for_change, tw, INOTIFY_DIR)
@ -102,6 +131,12 @@ class TestFilesystemWatchers(TestCase):
changed() changed()
os.rename(f, f + '1') os.rename(f, f + '1')
changed() changed()
finally:
clear_dir(INOTIFY_DIR)
def test_uv_tree_watcher(self):
tw = create_tree_watcher(get_fallback_logger(), 'uv')
return self.do_test_tree_watcher(tw)
old_cwd = None old_cwd = None
@ -116,11 +151,7 @@ def setUpModule():
def tearDownModule(): def tearDownModule():
for d in [INOTIFY_DIR]: for d in [INOTIFY_DIR]:
for root, dirs, files in list(os.walk(d, topdown=False)): clear_dir(d)
for file in files:
os.remove(os.path.join(root, file))
for dir in dirs:
os.rmdir(os.path.join(root, dir))
os.rmdir(d) os.rmdir(d)
os.chdir(old_cwd) os.chdir(old_cwd)