CTS: add service load/unload test

git-svn-id: http://svn.fedorahosted.org/svn/corosync/trunk@2730 fd59a12c-fef9-0310-b244-a6a79926bd2f

cts/corotests.py

@@ -180,7 +180,7 @@ class CpgCfgChgOnExecCrash(CpgConfigChangeBase):
 
     def failure_action(self):
         self.CM.log("sending SIGSEGV to corosync on " + self.wobbly)
-        self.CM.rsh(self.wobbly, "killall -SIGSEGV corosync")
+        self.CM.rsh(self.wobbly, "killall -9 corosync")
         self.CM.rsh(self.wobbly, "rm -f /var/run/corosync.pid")
 
     def __call__(self, node):
@@ -428,9 +428,88 @@ class MemLeakSession(CoroTest):
         else:
             return self.failure(str(mem_leaked) + 'kB memory leaked.')
 
+###################################################################
+class ServiceLoadTest(CoroTest):
+    '''
+    Test loading and unloading of service engines
+    '''
+    def __init__(self, cm):
+        CoroTest.__init__(self, cm)
+        self.name="ServiceLoadTest"
+
+    def is_loaded(self, node):
+        check = 'corosync-objctl runtime.services. | grep evs'
+
+        (res, out) = self.CM.rsh(node, check, stdout=2)
+        if res is 0:
+            return True
+        else:
+            return False
+
+    def service_unload(self, node):
+        # unload evs
+        pats = []
+        pats.append("%s .*Service engine unloaded: corosync extended.*" % node)
+        unloaded = self.create_watch(pats, 60)
+        unloaded.setwatch()
+
+        self.CM.rsh(node, 'corosync-cfgtool -u corosync_evs')
+
+        if not unloaded.lookforall():
+            self.CM.log("Patterns not found: " + repr(unloaded.unmatched))
+            self.error_message = "evs service not unloaded"
+            return False
+
+        if self.is_loaded(node):
+            self.error_message = "evs has been unload, why are it's session objects are still there?"
+            return False
+        return True
+
+    def service_load(self, node):
+        # now reload it.
+        pats = []
+        pats.append("%s .*Service engine loaded.*" % node)
+        loaded = self.create_watch(pats, 60)
+        loaded.setwatch()
+
+        self.CM.rsh(node, 'corosync-cfgtool -l corosync_evs')
+
+        if not loaded.lookforall():
+            self.CM.log("Patterns not found: " + repr(loaded.unmatched))
+            self.error_message = "evs service not unloaded"
+            return False
+
+        return True
+
+
+    def __call__(self, node):
+        self.incr("calls")
+        should_be_loaded = True
+
+        if self.is_loaded(node):
+            ret = self.service_unload(node)
+            should_be_loaded = False
+        else:
+            ret = self.service_load(node)
+            should_be_loaded = True
+
+        if not ret:
+            return self.failure(self.error_message)
+
+        if self.is_loaded(node):
+            ret = self.service_unload(node)
+        else:
+            ret = self.service_load(node)
+
+        if not ret:
+            return self.failure(self.error_message)
+
+        return self.success()
+
 
 
 AllTestClasses = []
+AllTestClasses.append(ServiceLoadTest)
 AllTestClasses.append(CpgMsgOrderBasic)
 AllTestClasses.append(CpgMsgOrderThreads)
 AllTestClasses.append(CpgMsgOrderSecNss)