Commit 1ca7b773 authored by Michael Hanselmann's avatar Michael Hanselmann
Browse files

Add basic unit tests for "gnt-cluster epo"



This patch adds some unit tests for “gnt-cluster epo”. Not everything is
covered, but at least the bug fixed in the previous patch is.
Signed-off-by: default avatarMichael Hanselmann <hansmi@google.com>
Reviewed-by: default avatarIustin Pop <iustin@google.com>
parent 318853ab
......@@ -1397,7 +1397,9 @@ def _EpoOff(opts, node_list, inst_map):
return constants.EXIT_FAILURE
def Epo(opts, args):
def Epo(opts, args, cl=None, _on_fn=_EpoOn, _off_fn=_EpoOff,
_confirm_fn=ConfirmOperation,
_stdout_fn=ToStdout, _stderr_fn=ToStderr):
"""EPO operations.
@param opts: the command line options selected by the user
......@@ -1408,25 +1410,24 @@ def Epo(opts, args):
"""
if opts.groups and opts.show_all:
ToStderr("Only one of --groups or --all are allowed")
_stderr_fn("Only one of --groups or --all are allowed")
return constants.EXIT_FAILURE
elif args and opts.show_all:
ToStderr("Arguments in combination with --all are not allowed")
_stderr_fn("Arguments in combination with --all are not allowed")
return constants.EXIT_FAILURE
client = GetClient()
if cl is None:
cl = GetClient()
if opts.groups:
node_query_list = itertools.chain(*client.QueryGroups(names=args,
fields=["node_list"],
use_locking=False))
node_query_list = \
itertools.chain(*cl.QueryGroups(args, ["node_list"], False))
else:
node_query_list = args
result = client.QueryNodes(names=node_query_list,
fields=["name", "master", "pinst_list",
"sinst_list", "powered", "offline"],
use_locking=False)
result = cl.QueryNodes(node_query_list, ["name", "master", "pinst_list",
"sinst_list", "powered", "offline"],
False)
all_nodes = map(compat.fst, result)
node_list = []
......@@ -1447,25 +1448,26 @@ def Epo(opts, args):
# already operating on the master at this point :)
continue
elif master and not opts.show_all:
ToStderr("%s is the master node, please do a master-failover to another"
" node not affected by the EPO or use --all if you intend to"
" shutdown the whole cluster", node)
_stderr_fn("%s is the master node, please do a master-failover to another"
" node not affected by the EPO or use --all if you intend to"
" shutdown the whole cluster", node)
return constants.EXIT_FAILURE
elif powered is None:
ToStdout("Node %s does not support out-of-band handling, it can not be"
" handled in a fully automated manner", node)
_stdout_fn("Node %s does not support out-of-band handling, it can not be"
" handled in a fully automated manner", node)
elif powered == opts.on:
ToStdout("Node %s is already in desired power state, skipping", node)
_stdout_fn("Node %s is already in desired power state, skipping", node)
elif not offline or (offline and powered):
node_list.append(node)
if not opts.force and not ConfirmOperation(all_nodes, "nodes", "epo"):
if not (opts.force or _confirm_fn(all_nodes, "nodes", "epo")):
return constants.EXIT_FAILURE
if opts.on:
return _EpoOn(opts, all_nodes, node_list, inst_map)
return _on_fn(opts, all_nodes, node_list, inst_map)
else:
return _EpoOff(opts, node_list, inst_map)
return _off_fn(opts, node_list, inst_map)
commands = {
"init": (
......
......@@ -22,15 +22,17 @@
"""Script for testing ganeti.client.gnt_cluster"""
import unittest
import optparse
from ganeti.client import gnt_cluster
from ganeti import utils
from ganeti import compat
from ganeti import constants
import testutils
class TestEpo(unittest.TestCase):
class TestEpoUtilities(unittest.TestCase):
def setUp(self):
self.nodes2ip = dict(("node%s" % i, "192.0.2.%s" % i) for i in range(1, 10))
self.nodes = set(self.nodes2ip.keys())
......@@ -143,5 +145,118 @@ class TestEpo(unittest.TestCase):
self.assertFalse(inst_map)
class _ClientForEpo:
def __init__(self, groups, nodes):
self._groups = groups
self._nodes = nodes
def QueryGroups(self, names, fields, use_locking):
assert not use_locking
assert fields == ["node_list"]
return self._groups
def QueryNodes(self, names, fields, use_locking):
assert not use_locking
assert fields == ["name", "master", "pinst_list", "sinst_list", "powered",
"offline"]
return self._nodes
class TestEpo(unittest.TestCase):
_ON_EXITCODE = 253
_OFF_EXITCODE = 254
def _ConfirmForce(self, *args):
self.fail("Shouldn't need confirmation")
def _Confirm(self, exp_names, result, names, ltype, text):
self.assertEqual(names, exp_names)
self.assertFalse(result is NotImplemented)
return result
def _Off(self, exp_node_list, opts, node_list, inst_map):
self.assertEqual(node_list, exp_node_list)
self.assertFalse(inst_map)
return self._OFF_EXITCODE
def _Test(self, *args, **kwargs):
defaults = dict(cl=NotImplemented, _on_fn=NotImplemented,
_off_fn=NotImplemented,
_stdout_fn=lambda *args: None,
_stderr_fn=lambda *args: None)
defaults.update(kwargs)
return gnt_cluster.Epo(*args, **defaults)
def testShowAllWithGroups(self):
opts = optparse.Values(dict(groups=True, show_all=True))
result = self._Test(opts, NotImplemented)
self.assertEqual(result, constants.EXIT_FAILURE)
def testShowAllWithArgs(self):
opts = optparse.Values(dict(groups=False, show_all=True))
result = self._Test(opts, ["a", "b", "c"])
self.assertEqual(result, constants.EXIT_FAILURE)
def testNoArgumentsNoParameters(self):
for (force, confirm_result) in [(True, NotImplemented), (False, False),
(False, True)]:
opts = optparse.Values(dict(groups=False, show_all=False, force=force,
on=False))
client = _ClientForEpo(NotImplemented, [
("node1.example.com", False, [], [], True, False),
])
if force:
confirm_fn = self._ConfirmForce
else:
confirm_fn = compat.partial(self._Confirm, ["node1.example.com"],
confirm_result)
off_fn = compat.partial(self._Off, ["node1.example.com"])
result = self._Test(opts, [], cl=client, _off_fn=off_fn,
_confirm_fn=confirm_fn)
if force or confirm_result:
self.assertEqual(result, self._OFF_EXITCODE)
else:
self.assertEqual(result, constants.EXIT_FAILURE)
def testPowerOn(self):
for master in [False, True]:
opts = optparse.Values(dict(groups=False, show_all=True,
force=True, on=True))
client = _ClientForEpo(NotImplemented, [
("node1.example.com", False, [], [], True, False),
("node2.example.com", False, [], [], False, False),
("node3.example.com", False, [], [], True, True),
("node4.example.com", False, [], [], None, True),
("node5.example.com", master, [], [], False, False),
])
def _On(_, all_nodes, node_list, inst_map):
self.assertEqual(all_nodes,
["node%s.example.com" % i for i in range(1, 6)])
if master:
self.assertEqual(node_list, ["node2.example.com"])
else:
self.assertEqual(node_list, ["node2.example.com",
"node5.example.com"])
self.assertFalse(inst_map)
return self._ON_EXITCODE
result = self._Test(opts, [], cl=client, _on_fn=_On,
_confirm_fn=self._ConfirmForce)
self.assertEqual(result, self._ON_EXITCODE)
def testMasterWithoutShowAll(self):
opts = optparse.Values(dict(groups=False, show_all=False,
force=True, on=False))
client = _ClientForEpo(NotImplemented, [
("node1.example.com", True, [], [], True, False),
])
result = self._Test(opts, [], cl=client, _confirm_fn=self._ConfirmForce)
self.assertEqual(result, constants.EXIT_FAILURE)
if __name__ == "__main__":
testutils.GanetiTestProgram()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment