Synnefo » snf-ganeti » ganeti-local

import optparse

from ganeti import constants

class TestEpoUtilities(unittest.TestCase):

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)