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

Make Inotify testing faster

Browse Source
This commit is contained in:
R. Richard 2019-06-19 15:26:04 +02:00 committed by Kristóf Tóth
parent 8bf18113b2
commit c6e0b54930
1 changed files with 66 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

145
lib/tfw/components/inotify/test_inotify.py
View File

@ -9,6 +9,7 @@ from os import chdir, mkdir, rename, walk, listdir
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
import pytest import pytest
import watchdog
from inotify import InotifyObserver from inotify import InotifyObserver
from inotify import ( from inotify import (
@ -17,11 +18,15 @@ from inotify import (
InotifyDirMovedEvent, InotifyDirDeletedEvent InotifyDirMovedEvent, InotifyDirDeletedEvent
) )
watchdog.observers.inotify_buffer.InotifyBuffer.delay = 0
class InotifyContext: class InotifyContext:
def __init__(self, folder, observer): def __init__(self, workdir, subdir, subfile, observer):
self.missing_events = 0 self.missing_events = 0
self.folder = folder self.workdir = workdir
self.subdir = subdir
self.subfile = subfile
self.observer = observer self.observer = observer
self.event_to_queue = { self.event_to_queue = {
@ -46,18 +51,18 @@ class InotifyContext:
return dirname return dirname
def join(self, path): def join(self, path):
return join(self.folder, path) return join(self.workdir, path)
def check_event(self, event_type, path): def check_event(self, event_type, path):
self.missing_events += 1 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 isinstance(event, event_type)
assert event.src_path == path assert event.src_path == path
return event return event
def check_empty(self, event_type): def check_empty(self, event_type):
with pytest.raises(Empty): 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): def check_any(self):
attrs = self.observer.__dict__.values() attrs = self.observer.__dict__.values()
@ -92,95 +97,77 @@ def generate_name():
@pytest.fixture() @pytest.fixture()
def context(): def context():
with TemporaryDirectory() as workdir: with TemporaryDirectory() as workdir:
chdir(workdir) subdir = join(workdir, generate_name())
subfile = join(subdir, generate_name()+'.txt')
for _ in range(5): mkdir(subdir)
newdir = join(workdir, generate_name()) Path(subfile).touch()
mkdir(newdir)
Path(join(newdir, generate_name()+'.txt')).touch()
monitor = InotifyTestObserver(workdir, recursive=True) monitor = InotifyTestObserver(workdir, recursive=True)
monitor.start() monitor.start()
yield InotifyContext(workdir, monitor) yield InotifyContext(workdir, subdir, subfile, monitor)
def test_create(context): def test_create(context):
for _, dirs, _ in list(walk(context.folder)): newfile = context.create_random_file(context.workdir, '.txt')
for name in dirs: context.check_event(InotifyFileCreatedEvent, newfile)
newfile = context.create_random_file(name, '.txt') newdir = context.create_random_folder(context.workdir)
context.check_event(InotifyFileCreatedEvent, newfile) context.check_event(InotifyDirCreatedEvent, newdir)
newdir = context.create_random_folder(name) assert context.check_any()
context.check_event(InotifyDirCreatedEvent, newdir)
assert context.check_any()
def test_modify(context): def test_modify(context):
for root, _, files in list(walk(context.folder)): with open(context.subfile, 'w') as ofile:
for name in files: ofile.write('text')
oldfile = join(root, name) context.check_event(InotifyFileModifiedEvent, context.subfile)
with open(oldfile, 'w') as ofile: while True:
ofile.write('text') try:
context.check_event(InotifyFileModifiedEvent, oldfile) context.observer.modify_queue.get(timeout=0.1)
rename(oldfile, oldfile+'_new') context.missing_events += 1
context.check_event(InotifyDirModifiedEvent, root) except Empty:
assert context.check_any() break
rename(context.subfile, context.subfile+'_new')
context.check_event(InotifyDirModifiedEvent, context.subdir)
assert context.check_any()
def test_move(context): def test_move(context):
for root, dirs, _ in list(walk(context.folder)): rename(context.subdir, context.subdir+'_new')
for name in dirs: context.check_event(InotifyDirMovedEvent, context.subdir)
olddir = join(root, name) context.check_event(InotifyFileMovedEvent, context.subfile)
oldfile = join(olddir, list(listdir(olddir))[0]) assert context.check_any()
rename(olddir, olddir+'_new')
context.check_event(InotifyDirMovedEvent, olddir)
context.check_event(InotifyFileMovedEvent, oldfile)
assert context.check_any()
def test_delete(context): def test_delete(context):
for root, dirs, _ in list(walk(context.folder)): rmtree(context.subdir)
for name in dirs: context.check_event(InotifyFileDeletedEvent, context.subfile)
olddir = join(root, name) context.check_event(InotifyDirDeletedEvent, context.subdir)
oldfile = join(olddir, list(listdir(olddir))[0]) assert context.check_any()
rmtree(olddir)
context.check_event(InotifyFileDeletedEvent, oldfile)
context.check_event(InotifyDirDeletedEvent, olddir)
assert context.check_any()
def test_path(context): def test_path(context):
for _, dirs, _ in list(walk(context.folder)): context.observer.path = context.subdir
for name in dirs: context.create_random_folder(context.workdir)
context.observer.path = context.join(name) newfile = context.create_random_file(context.subdir, '.txt')
context.create_random_folder('.') context.check_event(InotifyFileCreatedEvent, newfile)
newfile = context.create_random_file(name, '.txt') context.observer.path = context.subdir
context.check_event(InotifyFileCreatedEvent, newfile) assert context.check_any()
context.observer.path = context.folder
assert context.check_any()
def test_patterns(context): def test_patterns(context):
for _, dirs, _ in list(walk(context.folder)): context.observer.patterns = ['*.txt']
for name in dirs: context.create_random_file(context.subdir, '.bin')
context.observer.patterns = ["*.txt"] newfile = context.create_random_file(context.subdir, '.txt')
context.create_random_file(name, '.bin') context.check_event(InotifyFileCreatedEvent, newfile)
newfile = context.create_random_file(name, '.txt') context.check_empty(InotifyFileCreatedEvent)
context.check_event(InotifyFileCreatedEvent, newfile) assert context.check_any()
context.check_empty(InotifyFileCreatedEvent) context.observer.patterns = None
assert context.check_any()
context.observer.patterns = None
def test_exclude(context): def test_exclude(context):
for _, dirs, _ in list(walk(context.folder)): context.observer.exclude = ['*.txt']
for name in dirs: context.create_random_file(context.subdir, '.txt')
context.observer.exclude = ["*.txt"] newfile = context.create_random_file(context.subdir, '.bin')
context.create_random_file(name, '.txt') context.check_event(InotifyFileCreatedEvent, newfile)
newfile = context.create_random_file(name, '.bin') context.check_empty(InotifyFileCreatedEvent)
context.check_event(InotifyFileCreatedEvent, newfile) assert context.check_any()
context.check_empty(InotifyFileCreatedEvent) context.observer.exclude = None
assert context.check_any()
context.observer.exclude = None
def test_stress(context): def test_stress(context):
for _, dirs, _ in list(walk(context.folder)): newfile = []
for name in dirs: for i in range(1024):
newfile = [] newfile.append(context.create_random_file(context.subdir, '.txt'))
for i in range(1024): for i in range(1024):
newfile.append(context.create_random_file(name, '.txt')) context.check_event(InotifyFileCreatedEvent, newfile[i])
for i in range(1024): assert context.check_any()
context.check_event(InotifyFileCreatedEvent, newfile[i])
assert context.check_any()