/ - Diff - snf-ganeti - Greek Research and Technology Network's projects

         hr = backend.HooksRunner()
         return hr.RunHooks(hpath, phase, env)
       # test -----------------------
       @staticmethod
       def perspective_test_delay(params):
         """Run test delay.
         """
         duration = params[0]
         return utils.TestDelay(duration)
     class MyRealm:
       """Simple realm that forwards all requests to a ServerObject.

           raise errors.OpRetryError("There has been a modification to the"
                                     " config file and the operation has been"
                                     " aborted. Please retry.")
     class LUTestDelay(NoHooksLU):
       """Sleep for a specified amount of time.
       This LU sleeps on the master and/or nodes for a specified amoutn of
       time.
       """
       _OP_REQP = ["duration", "on_master", "on_nodes"]
       def CheckPrereq(self):
         """Check prerequisites.
         This checks that we have a good list of nodes and/or the duration
         is valid.
         """
         if self.op.on_nodes:
           self.op.on_nodes = _GetWantedNodes(self, self.op.on_nodes)
       def Exec(self, feedback_fn):
         """Do the actual sleep.
         """
         if self.op.on_master:
           if not utils.TestDelay(self.op.duration):
             raise errors.OpExecError("Error during master delay test")
         if self.op.on_nodes:
           result = rpc.call_test_delay(self.op.on_nodes, self.op.duration)
           if not result:
             raise errors.OpExecError("Complete failure from rpc call")
           for node, node_result in result.items():
             if not node_result:
               raise errors.OpExecError("Failure during rpc call to node %s,"
                                        " result: %s" % (node, node_result))

         opcodes.OpSearchTags: cmdlib.LUSearchTags,
         opcodes.OpAddTags: cmdlib.LUAddTags,
         opcodes.OpDelTags: cmdlib.LUDelTags,
         # test lu
         opcodes.OpTestDelay: cmdlib.LUTestDelay,
+        }
       def __init__(self, feedback=None):

       """Remove a list of tags from a given object."""
       OP_ID = "OP_TAGS_DEL"
       __slots__ = ["kind", "name", "tags"]
     # Test opcodes
     class OpTestDelay(OpCode):
       """Sleeps for a configured amount of time.
       This is used just for debugging and testing.
       Parameters:
         - duration: the time to sleep
         - on_master: if true, sleep on the master
         - on_nodes: list of nodes in which to sleep
       If the on_master parameter is true, it will execute a sleep on the
       master (before any node sleep).
       If the on_nodes list is not empty, it will sleep on those nodes
       (after the sleep on the master, if that is enabled).
       As an additional feature, the case of duration < 0 will be reported
       as an execution error, so this opcode can be used as a failure
       generator. The case of duration == 0 will not be treated specially.
       """
       OP_ID = "OP_TEST_DELAY"
       __slots__ = ["duration", "on_master", "on_nodes"]

       c.connect_list(node_list)
       c.run()
       return c.getresult()
     def call_test_delay(node_list, duration):
       """Sleep for a fixed time on given node(s).
       This is a multi-node call.
       """
       c = Client("test_delay", [duration])
       c.connect_list(node_list)
       c.run()
       return c.getresult()

       """
       mac_check = re.compile("^([0-9a-f]{2}(:|$)){6}$")
       return mac_check.match(mac) is not None
     def TestDelay(duration):
       """Sleep for a fixed amount of time.
       """
       if duration < 0:
         return False
       time.sleep(duration)
       return True

Synnefo » snf-ganeti