[Openais] [PATCH 1/2] CTS: add service load/unload test

Steven Dake sdake at redhat.com
Wed Mar 24 14:29:12 PDT 2010


good for merge

On Thu, 2010-03-18 at 13:43 +1100, Angus Salkeld wrote:
> This just tests that you can unload and load a service
> and that the service objects are removed on unloading.
> 
> (evs doesn't unload - next patch)
> 
> -Angus
> 
> Signed-off-by: Angus Salkeld <asalkeld at redhat.com>
> ---
>  cts/corotests.py |   81 +++++++++++++++++++++++++++++++++++++++++++++++++++++-
>  1 files changed, 80 insertions(+), 1 deletions(-)
> 
> diff --git a/cts/corotests.py b/cts/corotests.py
> index 6c53a6c..d07eb5f 100644
> --- a/cts/corotests.py
> +++ b/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)



More information about the Openais mailing list