From 49f986e766fb568347818d7bec945054f7b51b8a Mon Sep 17 00:00:00 2001
From: Guido Trotter <ultrotter@google.com>
Date: Fri, 21 May 2010 15:12:05 +0100
Subject: [PATCH] TestSingleFileEventHandler: abstract notifier type
Rather than hardcode that we have two notifiers, and notifier 0 is the
terminating one, we abstract this with class level constants. This makes
it easier to add more, with different features.
The only real change is that now the callback class takes as input the
whole test object, rather than just the notified array, to have access
to those constants.
The rest is just replacing of hardcoded 0s and 1s with
self.NOTIFIER_TERM and self.NOTIFIER_NORM, and of notifier_count with
len(self.NOTIFIERS).
Signed-off-by: Guido Trotter <ultrotter@google.com>
Reviewed-by: Balazs Lecz <leczb@google.com>
---
test/ganeti.asyncnotifier_unittest.py | 83 ++++++++++++++-------------
1 file changed, 42 insertions(+), 41 deletions(-)
diff --git a/test/ganeti.asyncnotifier_unittest.py b/test/ganeti.asyncnotifier_unittest.py
index f487af772..b379ba1ef 100755
--- a/test/ganeti.asyncnotifier_unittest.py
+++ b/test/ganeti.asyncnotifier_unittest.py
@@ -41,79 +41,80 @@ import testutils
class TestSingleFileEventHandler(testutils.GanetiTestCase):
"""Test daemon.Mainloop"""
+ NOTIFIERS = [NOTIFIER_TERM, NOTIFIER_NORM] = range(2)
+
def setUp(self):
testutils.GanetiTestCase.setUp(self)
self.mainloop = daemon.Mainloop()
- notifier_count = 2
- self.chk_files = [self._CreateTempFile() for i in range(notifier_count)]
- self.notified = [False for i in range(notifier_count)]
+ self.chk_files = [self._CreateTempFile() for i in self.NOTIFIERS]
+ self.notified = [False for i in self.NOTIFIERS]
# 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 range(notifier_count)]
- self.cbk = [self.OnInotifyCallback(self.notified, i)
- for i in range(notifier_count)]
+ 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(notifier_count)]
+ for i in range(len(self.NOTIFIERS))]
self.notifiers = [asyncnotifier.AsyncNotifier(self.wms[i],
self.ihandler[i])
- for i in range(notifier_count)]
- # notifier 0 is enabled by default, as we use it to get out of the loop
- self.ihandler[0].enable()
+ for i in range(len(self.NOTIFIERS))]
+ # TERM notifier is enabled by default, as we use it to get out of the loop
+ self.ihandler[self.NOTIFIER_TERM].enable()
class OnInotifyCallback:
- def __init__(self, notified, i):
- self.notified = notified
+ def __init__(self, testobj, i):
+ self.testobj = testobj
+ self.notified = testobj.notified
self.i = i
def __call__(self, enabled):
self.notified[self.i] = True
- # notifier 0 is special as we use it to terminate the mainloop
- if self.i == 0:
+ if self.i == self.testobj.NOTIFIER_TERM:
os.kill(os.getpid(), signal.SIGTERM)
def testReplace(self):
- utils.WriteFile(self.chk_files[0], data="dummy")
+ utils.WriteFile(self.chk_files[self.NOTIFIER_TERM], data="dummy")
self.mainloop.Run()
- self.assert_(self.notified[0])
- self.assert_(not self.notified[1])
+ self.assert_(self.notified[self.NOTIFIER_TERM])
+ self.assert_(not self.notified[self.NOTIFIER_NORM])
def testEnableDisable(self):
- self.ihandler[0].enable()
- self.ihandler[0].disable()
- self.ihandler[0].disable()
- self.ihandler[0].enable()
- self.ihandler[0].disable()
- self.ihandler[0].enable()
- utils.WriteFile(self.chk_files[0], data="dummy")
+ self.ihandler[self.NOTIFIER_TERM].enable()
+ self.ihandler[self.NOTIFIER_TERM].disable()
+ self.ihandler[self.NOTIFIER_TERM].disable()
+ self.ihandler[self.NOTIFIER_TERM].enable()
+ self.ihandler[self.NOTIFIER_TERM].disable()
+ self.ihandler[self.NOTIFIER_TERM].enable()
+ utils.WriteFile(self.chk_files[self.NOTIFIER_TERM], data="dummy")
self.mainloop.Run()
- self.assert_(self.notified[0])
- self.assert_(not self.notified[1])
+ self.assert_(self.notified[self.NOTIFIER_TERM])
+ self.assert_(not self.notified[self.NOTIFIER_NORM])
def testDoubleEnable(self):
- self.ihandler[0].enable()
- self.ihandler[0].enable()
- utils.WriteFile(self.chk_files[0], data="dummy")
+ self.ihandler[self.NOTIFIER_TERM].enable()
+ self.ihandler[self.NOTIFIER_TERM].enable()
+ utils.WriteFile(self.chk_files[self.NOTIFIER_TERM], data="dummy")
self.mainloop.Run()
- self.assert_(self.notified[0])
- self.assert_(not self.notified[1])
+ self.assert_(self.notified[self.NOTIFIER_TERM])
+ self.assert_(not self.notified[self.NOTIFIER_NORM])
def testDefaultDisabled(self):
- utils.WriteFile(self.chk_files[1], data="dummy")
- utils.WriteFile(self.chk_files[0], data="dummy")
+ utils.WriteFile(self.chk_files[self.NOTIFIER_NORM], data="dummy")
+ utils.WriteFile(self.chk_files[self.NOTIFIER_TERM], data="dummy")
self.mainloop.Run()
- self.assert_(self.notified[0])
- # notifier 1 is disabled by default
- self.assert_(not self.notified[1])
+ self.assert_(self.notified[self.NOTIFIER_TERM])
+ # NORM notifier is disabled by default
+ self.assert_(not self.notified[self.NOTIFIER_NORM])
def testBothEnabled(self):
- self.ihandler[1].enable()
- utils.WriteFile(self.chk_files[1], data="dummy")
- utils.WriteFile(self.chk_files[0], data="dummy")
+ self.ihandler[self.NOTIFIER_NORM].enable()
+ utils.WriteFile(self.chk_files[self.NOTIFIER_NORM], data="dummy")
+ utils.WriteFile(self.chk_files[self.NOTIFIER_TERM], data="dummy")
self.mainloop.Run()
- self.assert_(self.notified[0])
- self.assert_(self.notified[1])
+ self.assert_(self.notified[self.NOTIFIER_TERM])
+ self.assert_(self.notified[self.NOTIFIER_NORM])
if __name__ == "__main__":
--
GitLab