#!/usr/bin/python
#
-# Copyright (C) 2010 Google Inc.
+# Copyright (C) 2010, 2012 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
import unittest
import signal
import os
+import tempfile
+import shutil
try:
- # pylint: disable-msg=E0611
+ # pylint: disable=E0611
from pyinotify import pyinotify
except ImportError:
import pyinotify
# We need one watch manager per notifier, as those contain the file
# descriptor which is monitored by asyncore
self.wms = [pyinotify.WatchManager() for i in self.NOTIFIERS]
- self.cbk = [self.OnInotifyCallback(self, i)
- for i in range(len(self.NOTIFIERS))]
- self.ihandler = [asyncnotifier.SingleFileEventHandler(self.wms[i],
- self.cbk[i],
- self.chk_files[i])
- for i in range(len(self.NOTIFIERS))]
- self.notifiers = [_MyErrorLoggingAsyncNotifier(self.wms[i],
- self.ihandler[i])
- for i in range(len(self.NOTIFIERS))]
+ self.cbk = [self.OnInotifyCallback(self, i) for i in self.NOTIFIERS]
+ self.ihandler = [asyncnotifier.SingleFileEventHandler(wm, cb, cf)
+ for (wm, cb, cf) in
+ zip(self.wms, self.cbk, self.chk_files)]
+ self.notifiers = [_MyErrorLoggingAsyncNotifier(wm, ih)
+ for (wm, ih) in zip(self.wms, self.ihandler)]
# TERM notifier is enabled by default, as we use it to get out of the loop
self.ihandler[self.NOTIFIER_TERM].enable()
+ def tearDown(self):
+ # disable the inotifiers, before removing the files
+ for i in self.ihandler:
+ i.disable()
+ testutils.GanetiTestCase.tearDown(self)
+ # and unregister the fd's being polled
+ for n in self.notifiers:
+ n.del_channel()
+
class OnInotifyCallback:
def __init__(self, testobj, i):
self.testobj = testobj
self.assertEquals(self.notifiers[self.NOTIFIER_TERM].error_count, 0)
+class TestSingleFileEventHandlerError(unittest.TestCase):
+ def setUp(self):
+ self.tmpdir = tempfile.mkdtemp()
+
+ def tearDown(self):
+ shutil.rmtree(self.tmpdir)
+
+ def test(self):
+ wm = pyinotify.WatchManager()
+ handler = asyncnotifier.SingleFileEventHandler(wm, None,
+ utils.PathJoin(self.tmpdir,
+ "nonexist"))
+ self.assertRaises(errors.InotifyError, handler.enable)
+ self.assertRaises(errors.InotifyError, handler.enable)
+ handler.disable()
+ self.assertRaises(errors.InotifyError, handler.enable)
+
+
if __name__ == "__main__":
testutils.GanetiTestProgram()