1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108
|
import os
import time
import threading
import pytest
from doit.filewatch import FileModifyWatcher, get_platform_system
def testUnsuportedPlatform(monkeypatch):
monkeypatch.setattr(FileModifyWatcher, 'supported_platforms', ())
pytest.raises(Exception, FileModifyWatcher, [])
platform = get_platform_system()
@pytest.mark.skipif('platform not in FileModifyWatcher.supported_platforms')
class TestFileWatcher(object):
def testInit(self, restore_cwd, tmpdir):
dir1 = 'data3'
files = ('data/w1.txt', 'data/w2.txt')
tmpdir.mkdir('data')
for fname in files:
tmpdir.join(fname).open('a').close()
os.chdir(tmpdir.strpath)
fw = FileModifyWatcher((files[0], files[1], dir1))
# file_list contains absolute paths
assert 2 == len(fw.file_list)
assert os.path.abspath(files[0]) in fw.file_list
assert os.path.abspath(files[1]) in fw.file_list
# watch_dirs
assert 2 == len(fw.watch_dirs)
assert tmpdir.join('data') in fw.watch_dirs
assert tmpdir.join('data3') in fw.watch_dirs
# notify_dirs
assert 1 == len(fw.notify_dirs)
assert tmpdir.join('data3') in fw.notify_dirs
def testHandleEventNotSubclassed(self):
fw = FileModifyWatcher([])
pytest.raises(NotImplementedError, fw.handle_event, None)
def testLoop(self, restore_cwd, tmpdir):
files = ['data/w1.txt', 'data/w2.txt', 'data/w3.txt']
stop_file = 'data/stop'
tmpdir.mkdir('data')
for fname in files + [stop_file]:
tmpdir.join(fname).open('a').close()
os.chdir(tmpdir.strpath)
fw = FileModifyWatcher((files[0], files[1], stop_file))
events = []
should_stop = []
started = []
def handle_event(event):
events.append(event.pathname)
if event.pathname.endswith("stop"):
should_stop.append(True)
fw.handle_event = handle_event
def loop_callback(notifier):
started.append(True)
# force loop to stop
if should_stop:
raise KeyboardInterrupt
loop_thread = threading.Thread(target=fw.loop, args=(loop_callback,))
loop_thread.daemon = True
loop_thread.start()
# wait watcher to be ready
while not started: # pragma: no cover
time.sleep(0.01)
assert loop_thread.isAlive()
# write in watched file
fd = open(files[0], 'w')
fd.write("hi")
fd.close()
# write in non-watched file
fd = open(files[2], 'w')
fd.write("hi")
fd.close()
# write in another watched file
fd = open(files[1], 'w')
fd.write("hi")
fd.close()
# tricky to stop watching
fd = open(stop_file, 'w')
fd.write("hi")
fd.close()
time.sleep(0.1)
loop_thread.join(1)
if loop_thread.isAlive(): # pragma: no cover
# this test is very flaky so we give it one more chance...
# write on file to terminate thread
fd = open(stop_file, 'w')
fd.write("hi")
fd.close()
loop_thread.join(1)
if loop_thread.is_alive(): # pragma: no cover
raise Exception("thread not terminated")
assert os.path.abspath(files[0]) == events[0]
assert os.path.abspath(files[1]) == events[1]
|