あいつの日誌β

働きながら旅しています。

python でファイル操作イベントを検出する処理でテストしたい

あらすじ

あるディレクトリを監視してファイルに変更が加えられたら何かしらの処理を非同期で実行する時にテストどうする

状況説明

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

とりあえずテスト可能になった。