diff --git a/tests/test_watcher.py b/tests/test_watcher.py index 0d066150..3622043d 100644 --- a/tests/test_watcher.py +++ b/tests/test_watcher.py @@ -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