あらすじ
あるディレクトリを監視してファイルに変更が加えられたら何かしらの処理を非同期で実行する時にテストどうする
状況説明
version は下記バージョン
% python --version Python 2.7.3
サンプルコード
import pyinotify wm = pyinotify.WatchManager() # Watch Manager mask = pyinotify.IN_MODIFY # watched events class EventHandler(pyinotify.ProcessEvent): def process_IN_MODIFY(self, event): print event print "Modify:", event.pathname handler = EventHandler() notifier = pyinotify.Notifier(wm, handler) wdd = wm.add_watch('/tmp', mask, rec=True) notifier.loop()
このスクリプトを実行している間にファイルに変更を加える
% echo 'hoge' >> /tmp/notify_test
そうすると以下のように process_IN_MODIFY が実行されます。
:!python notify.py <Event dir=False mask=0x2 maskname=IN_MODIFY name=output path=/tmp pathname=/tmp/output wd=1 > Modify: /tmp/output
ただこのままだと process_IN_MODIFY テスト書く時にどうすればいいんだ問題が発生します(帰り値がない)。どうしたものか。
callback を与えてみた
import pyinotify wm = pyinotify.WatchManager() # Watch Manager mask = pyinotify.IN_MODIFY # watched events class EventHandler(pyinotify.ProcessEvent): def __init__(self, modify_callback=None): self.modify_callback = modify_callback super(EventHandler, self).__init__() def process_IN_MODIFY(self, event): print event print "Modify:", event.pathname if self.modify_callback: self.modify_callback('ok') def modify_ok(result): print result == 'ok' handler = EventHandler(modify_callback=modify_ok) notifier = pyinotify.Notifier(wm, handler) wdd = wm.add_watch('/tmp', mask, rec=True) notifier.loop()
このスクリプトを起動して /tmp 以下のファイルに変更を加えると以下の通り
:!python notify.py <Event dir=False mask=0x2 maskname=IN_MODIFY name=output path=/tmp pathname=/tmp/output wd=1 > Modify: /tmp/output True
とりあえずテスト可能になった。