From ca6797b3544c1fe72f2c002b76cb743c7da689fa Mon Sep 17 00:00:00 2001 From: "R. Richard" Date: Thu, 6 Jun 2019 13:41:13 +0200 Subject: [PATCH] Fixing issues in PR #56 --- lib/tfw/components/inotify/test_inotify.py | 61 +++++++++++----------- 1 file changed, 31 insertions(+), 30 deletions(-) diff --git a/lib/tfw/components/inotify/test_inotify.py b/lib/tfw/components/inotify/test_inotify.py index 1503a34..702e2c7 100644 --- a/lib/tfw/components/inotify/test_inotify.py +++ b/lib/tfw/components/inotify/test_inotify.py @@ -1,7 +1,7 @@ # pylint: disable=redefined-outer-name from queue import Empty, Queue -from secrets import token_urlsafe as random_name +from secrets import token_urlsafe from pathlib import Path from shutil import rmtree from os.path import join @@ -20,7 +20,7 @@ from inotify import ( class InotifyContext: def __init__(self, folder, observer): - self.missing = 0 + self.missing_events = 0 self.folder = folder self.observer = observer @@ -35,13 +35,13 @@ class InotifyContext: InotifyDirDeletedEvent : self.observer.delete_queue } - def random_file(self, dirname, extension): - filename = self.join(f'{dirname}/{random_name(16)}{extension}') + def create_random_file(self, dirname, extension): + filename = self.join(f'{dirname}/{generate_name()}{extension}') Path(filename).touch() return filename - def random_folder(self, basepath): - dirname = self.join(f'{basepath}/{random_name(16)}') + def create_random_folder(self, basepath): + dirname = self.join(f'{basepath}/{generate_name()}') mkdir(dirname) return dirname @@ -49,6 +49,7 @@ class InotifyContext: return join(self.folder, path) def check_event(self, event_type, path): + self.missing_events += 1 event = self.event_to_queue[event_type].get(timeout=1) assert isinstance(event, event_type) assert event.src_path == path @@ -58,18 +59,16 @@ class InotifyContext: with pytest.raises(Empty): self.event_to_queue[event_type].get(timeout=1) - def check_any(self, missing): - self.missing += missing + def check_any(self): attrs = self.observer.__dict__.values() total = sum([q.qsize() for q in attrs if isinstance(q, Queue)]) - return total+self.missing == len(self.observer.any_list) + return total+self.missing_events == len(self.observer.any_list) class InotifyTestObserver(InotifyObserver): def __init__(self, path, patterns=None, exclude=None): self.any_list = [] - (self.create_queue, self.modify_queue, self.move_queue, - self.delete_queue) = [Queue() for _ in range(4)] + self.create_queue, self.modify_queue, self.move_queue, self.delete_queue = [Queue() for _ in range(4)] super().__init__(path, patterns, exclude) def on_any_event(self, event): @@ -87,6 +86,8 @@ class InotifyTestObserver(InotifyObserver): def on_deleted(self, event): self.delete_queue.put(event) +def generate_name(): + return token_urlsafe(16) @pytest.fixture() def context(): @@ -94,9 +95,9 @@ def context(): chdir(workdir) for _ in range(5): - newdir = join(workdir, random_name(16)) + newdir = join(workdir, generate_name()) mkdir(newdir) - Path(join(newdir, random_name(16)+'.txt')).touch() + Path(join(newdir, generate_name()+'.txt')).touch() monitor = InotifyTestObserver(workdir) monitor.start() @@ -105,11 +106,11 @@ def context(): def test_create(context): for _, dirs, _ in list(walk(context.folder)): for name in dirs: - newfile = context.random_file(name, '.txt') + newfile = context.create_random_file(name, '.txt') context.check_event(InotifyFileCreatedEvent, newfile) - newdir = context.random_folder(name) + newdir = context.create_random_folder(name) context.check_event(InotifyDirCreatedEvent, newdir) - assert context.check_any(2) + assert context.check_any() def test_modify(context): for root, _, files in list(walk(context.folder)): @@ -120,7 +121,7 @@ def test_modify(context): context.check_event(InotifyFileModifiedEvent, oldfile) rename(oldfile, oldfile+'_new') context.check_event(InotifyDirModifiedEvent, root) - assert context.check_any(2) + assert context.check_any() def test_move(context): for root, dirs, _ in list(walk(context.folder)): @@ -130,7 +131,7 @@ def test_move(context): rename(olddir, olddir+'_new') context.check_event(InotifyDirMovedEvent, olddir) context.check_event(InotifyFileMovedEvent, oldfile) - assert context.check_any(2) + assert context.check_any() def test_delete(context): for root, dirs, _ in list(walk(context.folder)): @@ -140,38 +141,38 @@ def test_delete(context): rmtree(olddir) context.check_event(InotifyFileDeletedEvent, oldfile) context.check_event(InotifyDirDeletedEvent, olddir) - assert context.check_any(2) + 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.random_folder('.') - newfile = context.random_file(name, '.txt') + 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(1) + assert context.check_any() def test_patterns(context): for _, dirs, _ in list(walk(context.folder)): for name in dirs: context.observer.patterns = ["*.txt"] - context.random_file(name, '.bin') - newfile = context.random_file(name, '.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(1) + 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.random_file(name, '.txt') - newfile = context.random_file(name, '.bin') + 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(1) + assert context.check_any() context.observer.exclude = None def test_stress(context): @@ -179,7 +180,7 @@ def test_stress(context): for name in dirs: newfile = [] for i in range(1024): - newfile.append(context.random_file(name, '.txt')) + newfile.append(context.create_random_file(name, '.txt')) for i in range(1024): context.check_event(InotifyFileCreatedEvent, newfile[i]) - assert context.check_any(1024) + assert context.check_any()