From c6e0b5493067a103975de6a3171113560cff7bb8 Mon Sep 17 00:00:00 2001 From: "R. Richard" Date: Wed, 19 Jun 2019 15:26:04 +0200 Subject: [PATCH] Make Inotify testing faster --- lib/tfw/components/inotify/test_inotify.py | 145 ++++++++++----------- 1 file changed, 66 insertions(+), 79 deletions(-) diff --git a/lib/tfw/components/inotify/test_inotify.py b/lib/tfw/components/inotify/test_inotify.py index 6b4bdb3..b6e659e 100644 --- a/lib/tfw/components/inotify/test_inotify.py +++ b/lib/tfw/components/inotify/test_inotify.py @@ -9,6 +9,7 @@ from os import chdir, mkdir, rename, walk, listdir from tempfile import TemporaryDirectory import pytest +import watchdog from inotify import InotifyObserver from inotify import ( @@ -17,11 +18,15 @@ from inotify import ( InotifyDirMovedEvent, InotifyDirDeletedEvent ) +watchdog.observers.inotify_buffer.InotifyBuffer.delay = 0 + class InotifyContext: - def __init__(self, folder, observer): + def __init__(self, workdir, subdir, subfile, observer): self.missing_events = 0 - self.folder = folder + self.workdir = workdir + self.subdir = subdir + self.subfile = subfile self.observer = observer self.event_to_queue = { @@ -46,18 +51,18 @@ class InotifyContext: return dirname def join(self, path): - return join(self.folder, path) + return join(self.workdir, path) def check_event(self, event_type, path): self.missing_events += 1 - event = self.event_to_queue[event_type].get(timeout=1) + event = self.event_to_queue[event_type].get(timeout=0.1) assert isinstance(event, event_type) assert event.src_path == path return event def check_empty(self, event_type): with pytest.raises(Empty): - self.event_to_queue[event_type].get(timeout=1) + self.event_to_queue[event_type].get(timeout=0.1) def check_any(self): attrs = self.observer.__dict__.values() @@ -92,95 +97,77 @@ def generate_name(): @pytest.fixture() def context(): with TemporaryDirectory() as workdir: - chdir(workdir) - - for _ in range(5): - newdir = join(workdir, generate_name()) - mkdir(newdir) - Path(join(newdir, generate_name()+'.txt')).touch() - + subdir = join(workdir, generate_name()) + subfile = join(subdir, generate_name()+'.txt') + mkdir(subdir) + Path(subfile).touch() monitor = InotifyTestObserver(workdir, recursive=True) monitor.start() - yield InotifyContext(workdir, monitor) + yield InotifyContext(workdir, subdir, subfile, monitor) def test_create(context): - for _, dirs, _ in list(walk(context.folder)): - for name in dirs: - newfile = context.create_random_file(name, '.txt') - context.check_event(InotifyFileCreatedEvent, newfile) - newdir = context.create_random_folder(name) - context.check_event(InotifyDirCreatedEvent, newdir) - assert context.check_any() + newfile = context.create_random_file(context.workdir, '.txt') + context.check_event(InotifyFileCreatedEvent, newfile) + newdir = context.create_random_folder(context.workdir) + context.check_event(InotifyDirCreatedEvent, newdir) + assert context.check_any() def test_modify(context): - for root, _, files in list(walk(context.folder)): - for name in files: - oldfile = join(root, name) - with open(oldfile, 'w') as ofile: - ofile.write('text') - context.check_event(InotifyFileModifiedEvent, oldfile) - rename(oldfile, oldfile+'_new') - context.check_event(InotifyDirModifiedEvent, root) - assert context.check_any() + with open(context.subfile, 'w') as ofile: + ofile.write('text') + context.check_event(InotifyFileModifiedEvent, context.subfile) + while True: + try: + context.observer.modify_queue.get(timeout=0.1) + context.missing_events += 1 + except Empty: + break + rename(context.subfile, context.subfile+'_new') + context.check_event(InotifyDirModifiedEvent, context.subdir) + assert context.check_any() def test_move(context): - for root, dirs, _ in list(walk(context.folder)): - for name in dirs: - olddir = join(root, name) - oldfile = join(olddir, list(listdir(olddir))[0]) - rename(olddir, olddir+'_new') - context.check_event(InotifyDirMovedEvent, olddir) - context.check_event(InotifyFileMovedEvent, oldfile) - assert context.check_any() + rename(context.subdir, context.subdir+'_new') + context.check_event(InotifyDirMovedEvent, context.subdir) + context.check_event(InotifyFileMovedEvent, context.subfile) + assert context.check_any() def test_delete(context): - for root, dirs, _ in list(walk(context.folder)): - for name in dirs: - olddir = join(root, name) - oldfile = join(olddir, list(listdir(olddir))[0]) - rmtree(olddir) - context.check_event(InotifyFileDeletedEvent, oldfile) - context.check_event(InotifyDirDeletedEvent, olddir) - assert context.check_any() + rmtree(context.subdir) + context.check_event(InotifyFileDeletedEvent, context.subfile) + context.check_event(InotifyDirDeletedEvent, context.subdir) + assert context.check_any() def test_path(context): - for _, dirs, _ in list(walk(context.folder)): - for name in dirs: - context.observer.path = context.join(name) - context.create_random_folder('.') - newfile = context.create_random_file(name, '.txt') - context.check_event(InotifyFileCreatedEvent, newfile) - context.observer.path = context.folder - assert context.check_any() + context.observer.path = context.subdir + context.create_random_folder(context.workdir) + newfile = context.create_random_file(context.subdir, '.txt') + context.check_event(InotifyFileCreatedEvent, newfile) + context.observer.path = context.subdir + assert context.check_any() def test_patterns(context): - for _, dirs, _ in list(walk(context.folder)): - for name in dirs: - context.observer.patterns = ["*.txt"] - context.create_random_file(name, '.bin') - newfile = context.create_random_file(name, '.txt') - context.check_event(InotifyFileCreatedEvent, newfile) - context.check_empty(InotifyFileCreatedEvent) - assert context.check_any() - context.observer.patterns = None + context.observer.patterns = ['*.txt'] + context.create_random_file(context.subdir, '.bin') + newfile = context.create_random_file(context.subdir, '.txt') + context.check_event(InotifyFileCreatedEvent, newfile) + context.check_empty(InotifyFileCreatedEvent) + assert context.check_any() + context.observer.patterns = None def test_exclude(context): - for _, dirs, _ in list(walk(context.folder)): - for name in dirs: - context.observer.exclude = ["*.txt"] - context.create_random_file(name, '.txt') - newfile = context.create_random_file(name, '.bin') - context.check_event(InotifyFileCreatedEvent, newfile) - context.check_empty(InotifyFileCreatedEvent) - assert context.check_any() - context.observer.exclude = None + context.observer.exclude = ['*.txt'] + context.create_random_file(context.subdir, '.txt') + newfile = context.create_random_file(context.subdir, '.bin') + context.check_event(InotifyFileCreatedEvent, newfile) + context.check_empty(InotifyFileCreatedEvent) + assert context.check_any() + context.observer.exclude = None def test_stress(context): - for _, dirs, _ in list(walk(context.folder)): - for name in dirs: - newfile = [] - for i in range(1024): - newfile.append(context.create_random_file(name, '.txt')) - for i in range(1024): - context.check_event(InotifyFileCreatedEvent, newfile[i]) - assert context.check_any() + newfile = [] + for i in range(1024): + newfile.append(context.create_random_file(context.subdir, '.txt')) + for i in range(1024): + context.check_event(InotifyFileCreatedEvent, newfile[i]) + assert context.check_any()