baseimage-tutorial-framework/lib/tfw/components/inotify/test_inotify.py

178 lines
6.2 KiB
Python
Raw Normal View History

2019-05-31 07:36:19 +00:00
# pylint: disable=redefined-outer-name
from queue import Empty, Queue
2019-06-06 11:41:13 +00:00
from secrets import token_urlsafe
2019-05-31 07:36:19 +00:00
from pathlib import Path
from shutil import rmtree
from os.path import join
2019-06-27 15:38:53 +00:00
from os import mkdir, remove, rename
2019-05-31 07:36:19 +00:00
from tempfile import TemporaryDirectory
2019-06-19 13:26:04 +00:00
import watchdog
2019-06-20 14:04:51 +00:00
import pytest
2019-05-31 07:36:19 +00:00
from .inotify import InotifyObserver
from .inotify import (
2019-05-31 07:36:19 +00:00
InotifyFileCreatedEvent, InotifyFileModifiedEvent, InotifyFileMovedEvent,
InotifyFileDeletedEvent, InotifyDirCreatedEvent, InotifyDirModifiedEvent,
InotifyDirMovedEvent, InotifyDirDeletedEvent
)
2019-06-19 13:26:04 +00:00
watchdog.observers.inotify_buffer.InotifyBuffer.delay = 0
2019-05-31 07:36:19 +00:00
class InotifyContext:
2019-06-19 13:26:04 +00:00
def __init__(self, workdir, subdir, subfile, observer):
2019-06-06 11:41:13 +00:00
self.missing_events = 0
2019-06-19 13:26:04 +00:00
self.workdir = workdir
self.subdir = subdir
self.subfile = subfile
2019-05-31 07:36:19 +00:00
self.observer = observer
self.event_to_queue = {
InotifyFileCreatedEvent : self.observer.create_queue,
InotifyFileModifiedEvent : self.observer.modify_queue,
InotifyFileMovedEvent : self.observer.move_queue,
InotifyFileDeletedEvent : self.observer.delete_queue,
InotifyDirCreatedEvent : self.observer.create_queue,
InotifyDirModifiedEvent : self.observer.modify_queue,
InotifyDirMovedEvent : self.observer.move_queue,
InotifyDirDeletedEvent : self.observer.delete_queue
}
2019-06-06 11:41:13 +00:00
def create_random_file(self, dirname, extension):
filename = self.join(f'{dirname}/{generate_name()}{extension}')
2019-05-31 07:36:19 +00:00
Path(filename).touch()
return filename
2019-06-06 11:41:13 +00:00
def create_random_folder(self, basepath):
dirname = self.join(f'{basepath}/{generate_name()}')
2019-05-31 07:36:19 +00:00
mkdir(dirname)
return dirname
def join(self, path):
2019-06-19 13:26:04 +00:00
return join(self.workdir, path)
2019-05-31 07:36:19 +00:00
def check_event(self, event_type, path):
2019-06-06 11:41:13 +00:00
self.missing_events += 1
2019-06-19 13:26:04 +00:00
event = self.event_to_queue[event_type].get(timeout=0.1)
2019-05-31 07:36:19 +00:00
assert isinstance(event, event_type)
assert event.src_path == path
return event
def check_empty(self, event_type):
with pytest.raises(Empty):
2019-06-19 13:26:04 +00:00
self.event_to_queue[event_type].get(timeout=0.1)
2019-05-31 07:36:19 +00:00
2019-06-06 11:41:13 +00:00
def check_any(self):
2019-05-31 07:36:19 +00:00
attrs = self.observer.__dict__.values()
total = sum([q.qsize() for q in attrs if isinstance(q, Queue)])
2019-06-06 11:41:13 +00:00
return total+self.missing_events == len(self.observer.any_list)
2019-05-31 07:36:19 +00:00
class InotifyTestObserver(InotifyObserver):
def __init__(self, paths, patterns=None, exclude=None, recursive=False):
2019-05-31 07:36:19 +00:00
self.any_list = []
2019-06-06 11:41:13 +00:00
self.create_queue, self.modify_queue, self.move_queue, self.delete_queue = [Queue() for _ in range(4)]
super().__init__(paths, patterns, exclude, recursive)
2019-05-31 07:36:19 +00:00
def on_any_event(self, event):
self.any_list.append(event)
def on_created(self, event):
self.create_queue.put(event)
def on_modified(self, event):
self.modify_queue.put(event)
def on_moved(self, event):
self.move_queue.put(event)
def on_deleted(self, event):
self.delete_queue.put(event)
2019-06-06 11:41:13 +00:00
def generate_name():
return token_urlsafe(16)
2019-05-31 07:36:19 +00:00
@pytest.fixture()
def context():
with TemporaryDirectory() as workdir:
2019-06-19 13:26:04 +00:00
subdir = join(workdir, generate_name())
subfile = join(subdir, generate_name()+'.txt')
mkdir(subdir)
Path(subfile).touch()
monitor = InotifyTestObserver(workdir, recursive=True)
2019-05-31 07:36:19 +00:00
monitor.start()
2019-06-19 13:26:04 +00:00
yield InotifyContext(workdir, subdir, subfile, monitor)
2019-05-31 07:36:19 +00:00
def test_create(context):
2019-06-19 13:26:04 +00:00
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()
2019-05-31 07:36:19 +00:00
def test_modify(context):
with open(context.subfile, 'wb', buffering=0) as ofile:
ofile.write(b'text')
2019-06-19 13:26:04 +00:00
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()
2019-05-31 07:36:19 +00:00
def test_move(context):
2019-06-19 13:26:04 +00:00
rename(context.subdir, context.subdir+'_new')
context.check_event(InotifyDirMovedEvent, context.subdir)
context.check_event(InotifyFileMovedEvent, context.subfile)
assert context.check_any()
2019-05-31 07:36:19 +00:00
def test_delete(context):
2019-06-19 13:26:04 +00:00
rmtree(context.subdir)
context.check_event(InotifyFileDeletedEvent, context.subfile)
context.check_event(InotifyDirDeletedEvent, context.subdir)
assert context.check_any()
2019-05-31 07:36:19 +00:00
def test_paths(context):
context.observer.paths = context.subdir
2019-06-27 15:38:53 +00:00
newdir = context.create_random_folder(context.workdir)
2019-06-19 13:26:04 +00:00
newfile = context.create_random_file(context.subdir, '.txt')
2019-06-27 15:38:53 +00:00
context.check_event(InotifyDirModifiedEvent, context.subdir)
2019-06-19 13:26:04 +00:00
context.check_event(InotifyFileCreatedEvent, newfile)
2019-06-27 15:38:53 +00:00
context.observer.paths = [newdir, newfile]
remove(newfile)
context.check_event(InotifyFileDeletedEvent, newfile)
2019-06-19 13:26:04 +00:00
assert context.check_any()
2019-06-27 15:38:53 +00:00
context.observer.paths = context.workdir
2019-05-31 07:36:19 +00:00
def test_patterns(context):
2019-06-19 13:26:04 +00:00
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
2019-05-31 07:36:19 +00:00
def test_exclude(context):
2019-06-19 13:26:04 +00:00
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
2019-05-31 07:36:19 +00:00
def test_stress(context):
2019-06-19 13:26:04 +00:00
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()