Synnefo » snf-ganeti » ganeti-local

from ganeti import luxi

from ganeti import utils

import ganeti.rapi.client

KNOWN_UNUSED_LUXI = frozenset([

  luxi.REQ_SUBMIT_MANY_JOBS,

  luxi.REQ_ARCHIVE_JOB,

  luxi.REQ_AUTOARCHIVE_JOBS,

  luxi.REQ_QUERY_EXPORTS,

  luxi.REQ_QUERY_CONFIG_VALUES,

  luxi.REQ_QUERY_TAGS,

  luxi.REQ_QUEUE_SET_DRAIN_FLAG,

  luxi.REQ_SET_WATCHER_PAUSE,

  ])

# Global variable for storing used LUXI calls

_used_luxi_calls = None

class TestInputTestClient(unittest.TestCase):

  def setUp(self):

    self.cl = rapi.testutils.InputTestClient()

  def tearDown(self):

    _used_luxi_calls.update(self.cl._GetLuxiCalls())

  def testGetInfo(self):

    self.assertTrue(self.cl.GetInfo() is NotImplemented)

  def testPrepareExport(self):

    result = self.cl.PrepareExport("inst1.example.com",

                                   constants.EXPORT_MODE_LOCAL)

    self.assertTrue(result is NotImplemented)

    self.assertRaises(rapi.client.GanetiApiError, self.cl.PrepareExport,

                      "inst1.example.com", "###invalid###")

  def testGetJobs(self):

    self.assertTrue(self.cl.GetJobs() is NotImplemented)

  def testQuery(self):

    result = self.cl.Query(constants.QR_NODE, ["name"])

    self.assertTrue(result is NotImplemented)

  def testQueryFields(self):

    result = self.cl.QueryFields(constants.QR_INSTANCE)

    self.assertTrue(result is NotImplemented)

  def testCancelJob(self):

    self.assertTrue(self.cl.CancelJob("1") is NotImplemented)

  def testGetNodes(self):

    self.assertTrue(self.cl.GetNodes() is NotImplemented)

  def testGetInstances(self):

    self.assertTrue(self.cl.GetInstances() is NotImplemented)

  def testGetGroups(self):

    self.assertTrue(self.cl.GetGroups() is NotImplemented)

  def testWaitForJobChange(self):

    result = self.cl.WaitForJobChange("1", ["id"], None, None)

    self.assertTrue(result is NotImplemented)

class CustomTestRunner(unittest.TextTestRunner):

  def run(self, *args):

    global _used_luxi_calls

    assert _used_luxi_calls is None

    diff = (KNOWN_UNUSED_LUXI - luxi.REQ_ALL)

    assert not diff, "Non-existing LUXI calls listed as unused: %s" % diff

    _used_luxi_calls = set()

    try:

      # Run actual tests

      result = unittest.TextTestRunner.run(self, *args)

      diff = _used_luxi_calls & KNOWN_UNUSED_LUXI

      if diff:

        raise AssertionError("LUXI methods marked as unused were called: %s" %

                             utils.CommaJoin(diff))

      diff = (luxi.REQ_ALL - KNOWN_UNUSED_LUXI - _used_luxi_calls)

      if diff:

        raise AssertionError("The following LUXI methods were not used: %s" %

                             utils.CommaJoin(diff))

    finally:

      # Reset global variable

      _used_luxi_calls = None

    return result

  testutils.GanetiTestProgram(testRunner=CustomTestRunner)