mrtoth/baseimage-tutorial-framework
1
0
Fork 0
mirror of https://github.com/avatao-content/baseimage-tutorial-framework synced 2024-11-15 02:27:17 +00:00
Code Issues Releases Wiki Activity

Fix event order problem

Browse Source
This commit is contained in:
R. Richard 2019-08-07 09:42:03 +02:00
parent e6d2777520
commit 1210a30c31
1 changed files with 23 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
tfw/internals/inotify/test_inotify.py
View File

@ -55,12 +55,14 @@ class InotifyContext:
def join(self, path): def join(self, path):
return join(self.workdir, path) return join(self.workdir, path)
def check_event(self, event_type, path): def check_event(self, event_type, paths):
self.missing_events += 1 if isinstance(paths, str):
event = self.event_to_queue[event_type].get(timeout=0.1) paths = [paths]
assert isinstance(event, event_type) self.missing_events += len(paths)
assert event.src_path == path for _ in range(len(paths)):
return event event = self.event_to_queue[event_type].get(timeout=0.1)
assert isinstance(event, event_type)
assert event.src_path in paths
def check_empty(self, event_type): def check_empty(self, event_type):
with pytest.raises(Empty): with pytest.raises(Empty):
@ -69,7 +71,7 @@ class InotifyContext:
def check_any(self): def check_any(self):
attrs = self.observer.__dict__.values() attrs = self.observer.__dict__.values()
total = sum([q.qsize() for q in attrs if isinstance(q, Queue)]) total = sum([q.qsize() for q in attrs if isinstance(q, Queue)])
return total+self.missing_events == len(self.observer.any_list) return total + self.missing_events == len(self.observer.any_list)
class InotifyTestObserver(InotifyObserver): class InotifyTestObserver(InotifyObserver):
@ -100,7 +102,7 @@ def generate_name():
def context(): def context():
with TemporaryDirectory() as workdir: with TemporaryDirectory() as workdir:
subdir = join(workdir, generate_name()) subdir = join(workdir, generate_name())
subfile = join(subdir, generate_name()+'.txt') subfile = join(subdir, generate_name() + '.txt')
mkdir(subdir) mkdir(subdir)
Path(subfile).touch() Path(subfile).touch()
monitor = InotifyTestObserver(workdir, recursive=True) monitor = InotifyTestObserver(workdir, recursive=True)
@ -124,16 +126,21 @@ def test_modify(context):
context.missing_events += 1 context.missing_events += 1
except Empty: except Empty:
break break
rename(context.subfile, context.subfile+'_new') rename(context.subfile, context.subfile + '_new')
context.check_event(InotifyDirModifiedEvent, context.subdir) context.check_event(InotifyDirModifiedEvent, context.subdir)
assert context.check_any() assert context.check_any()
def test_move(context): def test_file_move(context):
rename(context.subdir, context.subdir+'_new') rename(context.subfile, context.subfile + '_new')
context.check_event(InotifyDirMovedEvent, context.subdir)
context.check_event(InotifyFileMovedEvent, context.subfile) context.check_event(InotifyFileMovedEvent, context.subfile)
assert context.check_any() assert context.check_any()
def test_folder_move(context):
remove(context.subfile)
rename(context.subdir, context.subdir + '_new')
context.check_event(InotifyDirMovedEvent, context.subdir)
assert context.check_any()
def test_delete(context): def test_delete(context):
rmtree(context.subdir) rmtree(context.subdir)
context.check_event(InotifyFileDeletedEvent, context.subfile) context.check_event(InotifyFileDeletedEvent, context.subfile)
@ -171,9 +178,8 @@ def test_exclude(context):
context.observer.exclude = None context.observer.exclude = None
def test_stress(context): def test_stress(context):
newfile = [] newfiles = []
for i in range(1024): for _ in range(1024):
newfile.append(context.create_random_file(context.subdir, '.txt')) newfiles.append(context.create_random_file(context.subdir, '.txt'))
for i in range(1024): context.check_event(InotifyFileCreatedEvent, newfiles)
context.check_event(InotifyFileCreatedEvent, newfile[i])
assert context.check_any() assert context.check_any()