#!/usr/bin/python
#

# Copyright (C) 2008, 2011, 2012, 2013 Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
# IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.


"""Script for unittesting the cli module"""

import copy
import testutils
import time
import unittest
import yaml
from cStringIO import StringIO

from ganeti import constants
from ganeti import cli
from ganeti import errors
from ganeti import utils
from ganeti import objects
from ganeti import qlang
from ganeti.errors import OpPrereqError, ParameterError


class TestParseTimespec(unittest.TestCase):
  """Testing case for ParseTimespec"""

  def testValidTimes(self):
    """Test valid timespecs"""
    test_data = [
      ("1s", 1),
      ("1", 1),
      ("1m", 60),
      ("1h", 60 * 60),
      ("1d", 60 * 60 * 24),
      ("1w", 60 * 60 * 24 * 7),
      ("4h", 4 * 60 * 60),
      ("61m", 61 * 60),
      ]
    for value, expected_result in test_data:
      self.failUnlessEqual(cli.ParseTimespec(value), expected_result)

  def testInvalidTime(self):
    """Test invalid timespecs"""
    test_data = [
      "1y",
      "",
      "aaa",
      "s",
      ]
    for value in test_data:
      self.failUnlessRaises(OpPrereqError, cli.ParseTimespec, value)


class TestToStream(unittest.TestCase):
  """Test the ToStream functions"""

  def testBasic(self):
    for data in ["foo",
                 "foo %s",
                 "foo %(test)s",
                 "foo %s %s",
                 "",
                 ]:
      buf = StringIO()
      cli._ToStream(buf, data)
      self.failUnlessEqual(buf.getvalue(), data + "\n")

  def testParams(self):
      buf = StringIO()
      cli._ToStream(buf, "foo %s", 1)
      self.failUnlessEqual(buf.getvalue(), "foo 1\n")
      buf = StringIO()
      cli._ToStream(buf, "foo %s", (15,16))
      self.failUnlessEqual(buf.getvalue(), "foo (15, 16)\n")
      buf = StringIO()
      cli._ToStream(buf, "foo %s %s", "a", "b")
      self.failUnlessEqual(buf.getvalue(), "foo a b\n")


class TestGenerateTable(unittest.TestCase):
  HEADERS = dict([("f%s" % i, "Field%s" % i) for i in range(5)])

  FIELDS1 = ["f1", "f2"]
  DATA1 = [
    ["abc", 1234],
    ["foobar", 56],
    ["b", -14],
    ]

  def _test(self, headers, fields, separator, data,
            numfields, unitfields, units, expected):
    table = cli.GenerateTable(headers, fields, separator, data,
                              numfields=numfields, unitfields=unitfields,
                              units=units)
    self.assertEqual(table, expected)

  def testPlain(self):
    exp = [
      "Field1 Field2",
      "abc    1234",
      "foobar 56",
      "b      -14",
      ]
    self._test(self.HEADERS, self.FIELDS1, None, self.DATA1,
               None, None, "m", exp)

  def testNoFields(self):
    self._test(self.HEADERS, [], None, [[], []],
               None, None, "m", ["", "", ""])
    self._test(None, [], None, [[], []],
               None, None, "m", ["", ""])

  def testSeparator(self):
    for sep in ["#", ":", ",", "^", "!", "%", "|", "###", "%%", "!!!", "||"]:
      exp = [
        "Field1%sField2" % sep,
        "abc%s1234" % sep,
        "foobar%s56" % sep,
        "b%s-14" % sep,
        ]
      self._test(self.HEADERS, self.FIELDS1, sep, self.DATA1,
                 None, None, "m", exp)

  def testNoHeader(self):
    exp = [
      "abc    1234",
      "foobar 56",
      "b      -14",
      ]
    self._test(None, self.FIELDS1, None, self.DATA1,
               None, None, "m", exp)

  def testUnknownField(self):
    headers = {
      "f1": "Field1",
      }
    exp = [
      "Field1 UNKNOWN",
      "abc    1234",
      "foobar 56",
      "b      -14",
      ]
    self._test(headers, ["f1", "UNKNOWN"], None, self.DATA1,
               None, None, "m", exp)

  def testNumfields(self):
    fields = ["f1", "f2", "f3"]
    data = [
      ["abc", 1234, 0],
      ["foobar", 56, 3],
      ["b", -14, "-"],
      ]
    exp = [
      "Field1 Field2 Field3",
      "abc      1234      0",
      "foobar     56      3",
      "b         -14      -",
      ]
    self._test(self.HEADERS, fields, None, data,
               ["f2", "f3"], None, "m", exp)

  def testUnitfields(self):
    expnosep = [
      "Field1 Field2 Field3",
      "abc      1234     0M",
      "foobar     56     3M",
      "b         -14      -",
      ]

    expsep = [
      "Field1:Field2:Field3",
      "abc:1234:0M",
      "foobar:56:3M",
      "b:-14:-",
      ]

    for sep, expected in [(None, expnosep), (":", expsep)]:
      fields = ["f1", "f2", "f3"]
      data = [
        ["abc", 1234, 0],
        ["foobar", 56, 3],
        ["b", -14, "-"],
        ]
      self._test(self.HEADERS, fields, sep, data,
                 ["f2", "f3"], ["f3"], "h", expected)

  def testUnusual(self):
    data = [
      ["%", "xyz"],
      ["%%", "abc"],
      ]
    exp = [
      "Field1 Field2",
      "%      xyz",
      "%%     abc",
      ]
    self._test(self.HEADERS, ["f1", "f2"], None, data,
               None, None, "m", exp)


class TestFormatQueryResult(unittest.TestCase):
  def test(self):
    fields = [
      objects.QueryFieldDefinition(name="name", title="Name",
                                   kind=constants.QFT_TEXT),
      objects.QueryFieldDefinition(name="size", title="Size",
                                   kind=constants.QFT_NUMBER),
      objects.QueryFieldDefinition(name="act", title="Active",
                                   kind=constants.QFT_BOOL),
      objects.QueryFieldDefinition(name="mem", title="Memory",
                                   kind=constants.QFT_UNIT),
      objects.QueryFieldDefinition(name="other", title="SomeList",
                                   kind=constants.QFT_OTHER),
      ]

    response = objects.QueryResponse(fields=fields, data=[
      [(constants.RS_NORMAL, "nodeA"), (constants.RS_NORMAL, 128),
       (constants.RS_NORMAL, False), (constants.RS_NORMAL, 1468006),
       (constants.RS_NORMAL, [])],
      [(constants.RS_NORMAL, "other"), (constants.RS_NORMAL, 512),
       (constants.RS_NORMAL, True), (constants.RS_NORMAL, 16),
       (constants.RS_NORMAL, [1, 2, 3])],
      [(constants.RS_NORMAL, "xyz"), (constants.RS_NORMAL, 1024),
       (constants.RS_NORMAL, True), (constants.RS_NORMAL, 4096),
       (constants.RS_NORMAL, [{}, {}])],
      ])

    self.assertEqual(cli.FormatQueryResult(response, unit="h", header=True),
      (cli.QR_NORMAL, [
      "Name  Size Active Memory SomeList",
      "nodeA  128 N        1.4T []",
      "other  512 Y         16M [1, 2, 3]",
      "xyz   1024 Y        4.0G [{}, {}]",
      ]))

  def testTimestampAndUnit(self):
    fields = [
      objects.QueryFieldDefinition(name="name", title="Name",
                                   kind=constants.QFT_TEXT),
      objects.QueryFieldDefinition(name="size", title="Size",
                                   kind=constants.QFT_UNIT),
      objects.QueryFieldDefinition(name="mtime", title="ModTime",
                                   kind=constants.QFT_TIMESTAMP),
      ]

    response = objects.QueryResponse(fields=fields, data=[
      [(constants.RS_NORMAL, "a"), (constants.RS_NORMAL, 1024),
       (constants.RS_NORMAL, 0)],
      [(constants.RS_NORMAL, "b"), (constants.RS_NORMAL, 144996),
       (constants.RS_NORMAL, 1291746295)],
      ])

    self.assertEqual(cli.FormatQueryResult(response, unit="m", header=True),
      (cli.QR_NORMAL, [
      "Name   Size ModTime",
      "a      1024 %s" % utils.FormatTime(0),
      "b    144996 %s" % utils.FormatTime(1291746295),
      ]))

  def testOverride(self):
    fields = [
      objects.QueryFieldDefinition(name="name", title="Name",
                                   kind=constants.QFT_TEXT),
      objects.QueryFieldDefinition(name="cust", title="Custom",
                                   kind=constants.QFT_OTHER),
      objects.QueryFieldDefinition(name="xt", title="XTime",
                                   kind=constants.QFT_TIMESTAMP),
      ]

    response = objects.QueryResponse(fields=fields, data=[
      [(constants.RS_NORMAL, "x"), (constants.RS_NORMAL, ["a", "b", "c"]),
       (constants.RS_NORMAL, 1234)],
      [(constants.RS_NORMAL, "y"), (constants.RS_NORMAL, range(10)),
       (constants.RS_NORMAL, 1291746295)],
      ])

    override = {
      "cust": (utils.CommaJoin, False),
      "xt": (hex, True),
      }

    self.assertEqual(cli.FormatQueryResult(response, unit="h", header=True,
                                           format_override=override),
      (cli.QR_NORMAL, [
      "Name Custom                            XTime",
      "x    a, b, c                           0x4d2",
      "y    0, 1, 2, 3, 4, 5, 6, 7, 8, 9 0x4cfe7bf7",
      ]))

  def testSeparator(self):
    fields = [
      objects.QueryFieldDefinition(name="name", title="Name",
                                   kind=constants.QFT_TEXT),
      objects.QueryFieldDefinition(name="count", title="Count",
                                   kind=constants.QFT_NUMBER),
      objects.QueryFieldDefinition(name="desc", title="Description",
                                   kind=constants.QFT_TEXT),
      ]

    response = objects.QueryResponse(fields=fields, data=[
      [(constants.RS_NORMAL, "instance1.example.com"),
       (constants.RS_NORMAL, 21125), (constants.RS_NORMAL, "Hello World!")],
      [(constants.RS_NORMAL, "mail.other.net"),
       (constants.RS_NORMAL, -9000), (constants.RS_NORMAL, "a,b,c")],
      ])

    for sep in [":", "|", "#", "|||", "###", "@@@", "@#@"]:
      for header in [None, "Name%sCount%sDescription" % (sep, sep)]:
        exp = []
        if header:
          exp.append(header)
        exp.extend([
          "instance1.example.com%s21125%sHello World!" % (sep, sep),
          "mail.other.net%s-9000%sa,b,c" % (sep, sep),
          ])

        self.assertEqual(cli.FormatQueryResult(response, separator=sep,
                                               header=bool(header)),
                         (cli.QR_NORMAL, exp))

  def testStatusWithUnknown(self):
    fields = [
      objects.QueryFieldDefinition(name="id", title="ID",
                                   kind=constants.QFT_NUMBER),
      objects.QueryFieldDefinition(name="unk", title="unk",
                                   kind=constants.QFT_UNKNOWN),
      objects.QueryFieldDefinition(name="unavail", title="Unavail",
                                   kind=constants.QFT_BOOL),
      objects.QueryFieldDefinition(name="nodata", title="NoData",
                                   kind=constants.QFT_TEXT),
      objects.QueryFieldDefinition(name="offline", title="OffLine",
                                   kind=constants.QFT_TEXT),
      ]

    response = objects.QueryResponse(fields=fields, data=[
      [(constants.RS_NORMAL, 1), (constants.RS_UNKNOWN, None),
       (constants.RS_NORMAL, False), (constants.RS_NORMAL, ""),
       (constants.RS_OFFLINE, None)],
      [(constants.RS_NORMAL, 2), (constants.RS_UNKNOWN, None),
       (constants.RS_NODATA, None), (constants.RS_NORMAL, "x"),
       (constants.RS_OFFLINE, None)],
      [(constants.RS_NORMAL, 3), (constants.RS_UNKNOWN, None),
       (constants.RS_NORMAL, False), (constants.RS_UNAVAIL, None),
       (constants.RS_OFFLINE, None)],
      ])

    self.assertEqual(cli.FormatQueryResult(response, header=True,
                                           separator="|", verbose=True),
      (cli.QR_UNKNOWN, [
      "ID|unk|Unavail|NoData|OffLine",
      "1|(unknown)|N||(offline)",
      "2|(unknown)|(nodata)|x|(offline)",
      "3|(unknown)|N|(unavail)|(offline)",
      ]))
    self.assertEqual(cli.FormatQueryResult(response, header=True,
                                           separator="|", verbose=False),
      (cli.QR_UNKNOWN, [
      "ID|unk|Unavail|NoData|OffLine",
      "1|??|N||*",
      "2|??|?|x|*",
      "3|??|N|-|*",
      ]))

  def testNoData(self):
    fields = [
      objects.QueryFieldDefinition(name="id", title="ID",
                                   kind=constants.QFT_NUMBER),
      objects.QueryFieldDefinition(name="name", title="Name",
                                   kind=constants.QFT_TEXT),
      ]

    response = objects.QueryResponse(fields=fields, data=[])

    self.assertEqual(cli.FormatQueryResult(response, header=True),
                     (cli.QR_NORMAL, ["ID Name"]))

  def testNoDataWithUnknown(self):
    fields = [
      objects.QueryFieldDefinition(name="id", title="ID",
                                   kind=constants.QFT_NUMBER),
      objects.QueryFieldDefinition(name="unk", title="unk",
                                   kind=constants.QFT_UNKNOWN),
      ]

    response = objects.QueryResponse(fields=fields, data=[])

    self.assertEqual(cli.FormatQueryResult(response, header=False),
                     (cli.QR_UNKNOWN, []))

  def testStatus(self):
    fields = [
      objects.QueryFieldDefinition(name="id", title="ID",
                                   kind=constants.QFT_NUMBER),
      objects.QueryFieldDefinition(name="unavail", title="Unavail",
                                   kind=constants.QFT_BOOL),
      objects.QueryFieldDefinition(name="nodata", title="NoData",
                                   kind=constants.QFT_TEXT),
      objects.QueryFieldDefinition(name="offline", title="OffLine",
                                   kind=constants.QFT_TEXT),
      ]

    response = objects.QueryResponse(fields=fields, data=[
      [(constants.RS_NORMAL, 1), (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, ""), (constants.RS_OFFLINE, None)],
      [(constants.RS_NORMAL, 2), (constants.RS_NODATA, None),
       (constants.RS_NORMAL, "x"), (constants.RS_NORMAL, "abc")],
      [(constants.RS_NORMAL, 3), (constants.RS_NORMAL, False),
       (constants.RS_UNAVAIL, None), (constants.RS_OFFLINE, None)],
      ])

    self.assertEqual(cli.FormatQueryResult(response, header=False,
                                           separator="|", verbose=True),
      (cli.QR_INCOMPLETE, [
      "1|N||(offline)",
      "2|(nodata)|x|abc",
      "3|N|(unavail)|(offline)",
      ]))
    self.assertEqual(cli.FormatQueryResult(response, header=False,
                                           separator="|", verbose=False),
      (cli.QR_INCOMPLETE, [
      "1|N||*",
      "2|?|x|abc",
      "3|N|-|*",
      ]))

  def testInvalidFieldType(self):
    fields = [
      objects.QueryFieldDefinition(name="x", title="x",
                                   kind="#some#other#type"),
      ]

    response = objects.QueryResponse(fields=fields, data=[])

    self.assertRaises(NotImplementedError, cli.FormatQueryResult, response)

  def testInvalidFieldStatus(self):
    fields = [
      objects.QueryFieldDefinition(name="x", title="x",
                                   kind=constants.QFT_TEXT),
      ]

    response = objects.QueryResponse(fields=fields, data=[[(-1, None)]])
    self.assertRaises(NotImplementedError, cli.FormatQueryResult, response)

    response = objects.QueryResponse(fields=fields, data=[[(-1, "x")]])
    self.assertRaises(AssertionError, cli.FormatQueryResult, response)

  def testEmptyFieldTitle(self):
    fields = [
      objects.QueryFieldDefinition(name="x", title="",
                                   kind=constants.QFT_TEXT),
      ]

    response = objects.QueryResponse(fields=fields, data=[])
    self.assertRaises(AssertionError, cli.FormatQueryResult, response)


class _MockJobPollCb(cli.JobPollCbBase, cli.JobPollReportCbBase):
  def __init__(self, tc, job_id):
    self.tc = tc
    self.job_id = job_id
    self._wfjcr = []
    self._jobstatus = []
    self._expect_notchanged = False
    self._expect_log = []

  def CheckEmpty(self):
    self.tc.assertFalse(self._wfjcr)
    self.tc.assertFalse(self._jobstatus)
    self.tc.assertFalse(self._expect_notchanged)
    self.tc.assertFalse(self._expect_log)

  def AddWfjcResult(self, *args):
    self._wfjcr.append(args)

  def AddQueryJobsResult(self, *args):
    self._jobstatus.append(args)

  def WaitForJobChangeOnce(self, job_id, fields,
                           prev_job_info, prev_log_serial,
                           timeout=constants.DEFAULT_WFJC_TIMEOUT):
    self.tc.assertEqual(job_id, self.job_id)
    self.tc.assertEqualValues(fields, ["status"])
    self.tc.assertFalse(self._expect_notchanged)
    self.tc.assertFalse(self._expect_log)

    (exp_prev_job_info, exp_prev_log_serial, result) = self._wfjcr.pop(0)
    self.tc.assertEqualValues(prev_job_info, exp_prev_job_info)
    self.tc.assertEqual(prev_log_serial, exp_prev_log_serial)

    if result == constants.JOB_NOTCHANGED:
      self._expect_notchanged = True
    elif result:
      (_, logmsgs) = result
      if logmsgs:
        self._expect_log.extend(logmsgs)

    return result

  def QueryJobs(self, job_ids, fields):
    self.tc.assertEqual(job_ids, [self.job_id])
    self.tc.assertEqualValues(fields, ["status", "opstatus", "opresult"])
    self.tc.assertFalse(self._expect_notchanged)
    self.tc.assertFalse(self._expect_log)

    result = self._jobstatus.pop(0)
    self.tc.assertEqual(len(fields), len(result))
    return [result]

  def CancelJob(self, job_id):
    self.tc.assertEqual(job_id, self.job_id)

  def ReportLogMessage(self, job_id, serial, timestamp, log_type, log_msg):
    self.tc.assertEqual(job_id, self.job_id)
    self.tc.assertEqualValues((serial, timestamp, log_type, log_msg),
                              self._expect_log.pop(0))

  def ReportNotChanged(self, job_id, status):
    self.tc.assertEqual(job_id, self.job_id)
    self.tc.assert_(self._expect_notchanged)
    self._expect_notchanged = False


class TestGenericPollJob(testutils.GanetiTestCase):
  def testSuccessWithLog(self):
    job_id = 29609
    cbs = _MockJobPollCb(self, job_id)

    cbs.AddWfjcResult(None, None, constants.JOB_NOTCHANGED)

    cbs.AddWfjcResult(None, None,
                      ((constants.JOB_STATUS_QUEUED, ), None))

    cbs.AddWfjcResult((constants.JOB_STATUS_QUEUED, ), None,
                      constants.JOB_NOTCHANGED)

    cbs.AddWfjcResult((constants.JOB_STATUS_QUEUED, ), None,
                      ((constants.JOB_STATUS_RUNNING, ),
                       [(1, utils.SplitTime(1273491611.0),
                         constants.ELOG_MESSAGE, "Step 1"),
                        (2, utils.SplitTime(1273491615.9),
                         constants.ELOG_MESSAGE, "Step 2"),
                        (3, utils.SplitTime(1273491625.02),
                         constants.ELOG_MESSAGE, "Step 3"),
                        (4, utils.SplitTime(1273491635.05),
                         constants.ELOG_MESSAGE, "Step 4"),
                        (37, utils.SplitTime(1273491645.0),
                         constants.ELOG_MESSAGE, "Step 5"),
                        (203, utils.SplitTime(127349155.0),
                         constants.ELOG_MESSAGE, "Step 6")]))

    cbs.AddWfjcResult((constants.JOB_STATUS_RUNNING, ), 203,
                      ((constants.JOB_STATUS_RUNNING, ),
                       [(300, utils.SplitTime(1273491711.01),
                         constants.ELOG_MESSAGE, "Step X"),
                        (302, utils.SplitTime(1273491815.8),
                         constants.ELOG_MESSAGE, "Step Y"),
                        (303, utils.SplitTime(1273491925.32),
                         constants.ELOG_MESSAGE, "Step Z")]))

    cbs.AddWfjcResult((constants.JOB_STATUS_RUNNING, ), 303,
                      ((constants.JOB_STATUS_SUCCESS, ), None))

    cbs.AddQueryJobsResult(constants.JOB_STATUS_SUCCESS,
                           [constants.OP_STATUS_SUCCESS,
                            constants.OP_STATUS_SUCCESS],
                           ["Hello World", "Foo man bar"])

    self.assertEqual(["Hello World", "Foo man bar"],
                     cli.GenericPollJob(job_id, cbs, cbs))
    cbs.CheckEmpty()

  def testJobLost(self):
    job_id = 13746

    cbs = _MockJobPollCb(self, job_id)
    cbs.AddWfjcResult(None, None, constants.JOB_NOTCHANGED)
    cbs.AddWfjcResult(None, None, None)
    self.assertRaises(errors.JobLost, cli.GenericPollJob, job_id, cbs, cbs)
    cbs.CheckEmpty()

  def testError(self):
    job_id = 31088

    cbs = _MockJobPollCb(self, job_id)
    cbs.AddWfjcResult(None, None, constants.JOB_NOTCHANGED)
    cbs.AddWfjcResult(None, None, ((constants.JOB_STATUS_ERROR, ), None))
    cbs.AddQueryJobsResult(constants.JOB_STATUS_ERROR,
                           [constants.OP_STATUS_SUCCESS,
                            constants.OP_STATUS_ERROR],
                           ["Hello World", "Error code 123"])
    self.assertRaises(errors.OpExecError, cli.GenericPollJob, job_id, cbs, cbs)
    cbs.CheckEmpty()

  def testError2(self):
    job_id = 22235

    cbs = _MockJobPollCb(self, job_id)
    cbs.AddWfjcResult(None, None, ((constants.JOB_STATUS_ERROR, ), None))
    encexc = errors.EncodeException(errors.LockError("problem"))
    cbs.AddQueryJobsResult(constants.JOB_STATUS_ERROR,
                           [constants.OP_STATUS_ERROR], [encexc])
    self.assertRaises(errors.LockError, cli.GenericPollJob, job_id, cbs, cbs)
    cbs.CheckEmpty()

  def testWeirdError(self):
    job_id = 28847

    cbs = _MockJobPollCb(self, job_id)
    cbs.AddWfjcResult(None, None, ((constants.JOB_STATUS_ERROR, ), None))
    cbs.AddQueryJobsResult(constants.JOB_STATUS_ERROR,
                           [constants.OP_STATUS_RUNNING,
                            constants.OP_STATUS_RUNNING],
                           [None, None])
    self.assertRaises(errors.OpExecError, cli.GenericPollJob, job_id, cbs, cbs)
    cbs.CheckEmpty()

  def testCancel(self):
    job_id = 4275

    cbs = _MockJobPollCb(self, job_id)
    cbs.AddWfjcResult(None, None, constants.JOB_NOTCHANGED)
    cbs.AddWfjcResult(None, None, ((constants.JOB_STATUS_CANCELING, ), None))
    cbs.AddQueryJobsResult(constants.JOB_STATUS_CANCELING,
                           [constants.OP_STATUS_CANCELING,
                            constants.OP_STATUS_CANCELING],
                           [None, None])
    self.assertRaises(errors.JobCanceled, cli.GenericPollJob, job_id, cbs, cbs)
    cbs.CheckEmpty()

  def testNegativeUpdateFreqParameter(self):
    job_id = 12345

    cbs = _MockJobPollCb(self, job_id)
    self.assertRaises(errors.ParameterError, cli.GenericPollJob, job_id, cbs,
                      cbs, update_freq=-30)

  def testZeroUpdateFreqParameter(self):
    job_id = 12345

    cbs = _MockJobPollCb(self, job_id)
    self.assertRaises(errors.ParameterError, cli.GenericPollJob, job_id, cbs,
                      cbs, update_freq=0)

  def testShouldCancel(self):
    job_id = 12345

    cbs = _MockJobPollCb(self, job_id)
    cbs.AddWfjcResult(None, None, constants.JOB_NOTCHANGED)
    self.assertRaises(errors.JobCanceled, cli.GenericPollJob, job_id, cbs, cbs,
                      cancel_fn=(lambda: True))

  def testIgnoreCancel(self):
    job_id = 12345
    cbs = _MockJobPollCb(self, job_id)
    cbs.AddWfjcResult(None, None, ((constants.JOB_STATUS_SUCCESS, ), None))
    cbs.AddQueryJobsResult(constants.JOB_STATUS_SUCCESS,
                           [constants.OP_STATUS_SUCCESS,
                            constants.OP_STATUS_SUCCESS],
                           ["Hello World", "Foo man bar"])
    self.assertEqual(["Hello World", "Foo man bar"],
                     cli.GenericPollJob(
                         job_id, cbs, cbs, cancel_fn=(lambda: False)))
    cbs.CheckEmpty()

class TestFormatLogMessage(unittest.TestCase):
  def test(self):
    self.assertEqual(cli.FormatLogMessage(constants.ELOG_MESSAGE,
                                          "Hello World"),
                     "Hello World")
    self.assertRaises(TypeError, cli.FormatLogMessage,
                      constants.ELOG_MESSAGE, [1, 2, 3])

    self.assert_(cli.FormatLogMessage("some other type", (1, 2, 3)))


class TestParseFields(unittest.TestCase):
  def test(self):
    self.assertEqual(cli.ParseFields(None, []), [])
    self.assertEqual(cli.ParseFields("name,foo,hello", []),
                     ["name", "foo", "hello"])
    self.assertEqual(cli.ParseFields(None, ["def", "ault", "fields", "here"]),
                     ["def", "ault", "fields", "here"])
    self.assertEqual(cli.ParseFields("name,foo", ["def", "ault"]),
                     ["name", "foo"])
    self.assertEqual(cli.ParseFields("+name,foo", ["def", "ault"]),
                     ["def", "ault", "name", "foo"])


class TestParseNicOption(unittest.TestCase):
  def test(self):
    self.assertEqual(cli.ParseNicOption([("0", { "link": "eth0", })]),
                     [{ "link": "eth0", }])
    self.assertEqual(cli.ParseNicOption([("5", { "ip": "192.0.2.7", })]),
                     [{}, {}, {}, {}, {}, { "ip": "192.0.2.7", }])

  def testErrors(self):
    for i in [None, "", "abc", "zero", "Hello World", "\0", []]:
      self.assertRaises(errors.OpPrereqError, cli.ParseNicOption,
                        [(i, { "link": "eth0", })])
      self.assertRaises(errors.OpPrereqError, cli.ParseNicOption,
                        [("0", i)])

    self.assertRaises(errors.TypeEnforcementError, cli.ParseNicOption,
                      [(0, { True: False, })])

    self.assertRaises(errors.TypeEnforcementError, cli.ParseNicOption,
                      [(3, { "mode": [], })])


class TestFormatResultError(unittest.TestCase):
  def testNormal(self):
    for verbose in [False, True]:
      self.assertRaises(AssertionError, cli.FormatResultError,
                        constants.RS_NORMAL, verbose)

  def testUnknown(self):
    for verbose in [False, True]:
      self.assertRaises(NotImplementedError, cli.FormatResultError,
                        "#some!other!status#", verbose)

  def test(self):
    for status in constants.RS_ALL:
      if status == constants.RS_NORMAL:
        continue

      self.assertNotEqual(cli.FormatResultError(status, False),
                          cli.FormatResultError(status, True))

      result = cli.FormatResultError(status, True)
      self.assertTrue(result.startswith("("))
      self.assertTrue(result.endswith(")"))


class TestGetOnlineNodes(unittest.TestCase):
  class _FakeClient:
    def __init__(self):
      self._query = []

    def AddQueryResult(self, *args):
      self._query.append(args)

    def CountPending(self):
      return len(self._query)

    def Query(self, res, fields, qfilter):
      if res != constants.QR_NODE:
        raise Exception("Querying wrong resource")

      (exp_fields, check_filter, result) = self._query.pop(0)

      if exp_fields != fields:
        raise Exception("Expected fields %s, got %s" % (exp_fields, fields))

      if not (qfilter is None or check_filter(qfilter)):
        raise Exception("Filter doesn't match expectations")

      return objects.QueryResponse(fields=None, data=result)

  def testEmpty(self):
    cl = self._FakeClient()

    cl.AddQueryResult(["name", "offline", "sip"], None, [])
    self.assertEqual(cli.GetOnlineNodes(None, cl=cl), [])
    self.assertEqual(cl.CountPending(), 0)

  def testNoSpecialFilter(self):
    cl = self._FakeClient()

    cl.AddQueryResult(["name", "offline", "sip"], None, [
      [(constants.RS_NORMAL, "master.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.1")],
      [(constants.RS_NORMAL, "node2.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.2")],
      ])
    self.assertEqual(cli.GetOnlineNodes(None, cl=cl),
                     ["master.example.com", "node2.example.com"])
    self.assertEqual(cl.CountPending(), 0)

  def testNoMaster(self):
    cl = self._FakeClient()

    def _CheckFilter(qfilter):
      self.assertEqual(qfilter, [qlang.OP_NOT, [qlang.OP_TRUE, "master"]])
      return True

    cl.AddQueryResult(["name", "offline", "sip"], _CheckFilter, [
      [(constants.RS_NORMAL, "node2.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.2")],
      ])
    self.assertEqual(cli.GetOnlineNodes(None, cl=cl, filter_master=True),
                     ["node2.example.com"])
    self.assertEqual(cl.CountPending(), 0)

  def testSecondaryIpAddress(self):
    cl = self._FakeClient()

    cl.AddQueryResult(["name", "offline", "sip"], None, [
      [(constants.RS_NORMAL, "master.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.1")],
      [(constants.RS_NORMAL, "node2.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.2")],
      ])
    self.assertEqual(cli.GetOnlineNodes(None, cl=cl, secondary_ips=True),
                     ["192.0.2.1", "192.0.2.2"])
    self.assertEqual(cl.CountPending(), 0)

  def testNoMasterFilterNodeName(self):
    cl = self._FakeClient()

    def _CheckFilter(qfilter):
      self.assertEqual(qfilter,
        [qlang.OP_AND,
         [qlang.OP_OR] + [[qlang.OP_EQUAL, "name", name]
                          for name in ["node2", "node3"]],
         [qlang.OP_NOT, [qlang.OP_TRUE, "master"]]])
      return True

    cl.AddQueryResult(["name", "offline", "sip"], _CheckFilter, [
      [(constants.RS_NORMAL, "node2.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.12")],
      [(constants.RS_NORMAL, "node3.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.13")],
      ])
    self.assertEqual(cli.GetOnlineNodes(["node2", "node3"], cl=cl,
                                        secondary_ips=True, filter_master=True),
                     ["192.0.2.12", "192.0.2.13"])
    self.assertEqual(cl.CountPending(), 0)

  def testOfflineNodes(self):
    cl = self._FakeClient()

    cl.AddQueryResult(["name", "offline", "sip"], None, [
      [(constants.RS_NORMAL, "master.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.1")],
      [(constants.RS_NORMAL, "node2.example.com"),
       (constants.RS_NORMAL, True),
       (constants.RS_NORMAL, "192.0.2.2")],
      [(constants.RS_NORMAL, "node3.example.com"),
       (constants.RS_NORMAL, True),
       (constants.RS_NORMAL, "192.0.2.3")],
      ])
    self.assertEqual(cli.GetOnlineNodes(None, cl=cl, nowarn=True),
                     ["master.example.com"])
    self.assertEqual(cl.CountPending(), 0)

  def testNodeGroup(self):
    cl = self._FakeClient()

    def _CheckFilter(qfilter):
      self.assertEqual(qfilter,
        [qlang.OP_OR, [qlang.OP_EQUAL, "group", "foobar"],
                      [qlang.OP_EQUAL, "group.uuid", "foobar"]])
      return True

    cl.AddQueryResult(["name", "offline", "sip"], _CheckFilter, [
      [(constants.RS_NORMAL, "master.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.1")],
      [(constants.RS_NORMAL, "node3.example.com"),
       (constants.RS_NORMAL, False),
       (constants.RS_NORMAL, "192.0.2.3")],
      ])
    self.assertEqual(cli.GetOnlineNodes(None, cl=cl, nodegroup="foobar"),
                     ["master.example.com", "node3.example.com"])
    self.assertEqual(cl.CountPending(), 0)


class TestFormatTimestamp(unittest.TestCase):
  def testGood(self):
    self.assertEqual(cli.FormatTimestamp((0, 1)),
                     time.strftime("%F %T", time.localtime(0)) + ".000001")
    self.assertEqual(cli.FormatTimestamp((1332944009, 17376)),
                     (time.strftime("%F %T", time.localtime(1332944009)) +
                      ".017376"))

  def testWrong(self):
    for i in [0, [], {}, "", [1]]:
      self.assertEqual(cli.FormatTimestamp(i), "?")


class TestFormatUsage(unittest.TestCase):
  def test(self):
    binary = "gnt-unittest"
    commands = {
      "cmdA":
        (NotImplemented, NotImplemented, NotImplemented, NotImplemented,
         "description of A"),
      "bbb":
        (NotImplemented, NotImplemented, NotImplemented, NotImplemented,
         "Hello World," * 10),
      "longname":
        (NotImplemented, NotImplemented, NotImplemented, NotImplemented,
         "Another description"),
      }

    self.assertEqual(list(cli._FormatUsage(binary, commands)), [
      "Usage: gnt-unittest {command} [options...] [argument...]",
      "gnt-unittest <command> --help to see details, or man gnt-unittest",
      "",
      "Commands:",
      (" bbb      - Hello World,Hello World,Hello World,Hello World,Hello"
       " World,Hello"),
      "            World,Hello World,Hello World,Hello World,Hello World,",
      " cmdA     - description of A",
      " longname - Another description",
      "",
      ])


class TestParseArgs(unittest.TestCase):
  def testNoArguments(self):
    for argv in [[], ["gnt-unittest"]]:
      try:
        cli._ParseArgs("gnt-unittest", argv, {}, {}, set())
      except cli._ShowUsage, err:
        self.assertTrue(err.exit_error)
      else:
        self.fail("Did not raise exception")

  def testVersion(self):
    for argv in [["test", "--version"], ["test", "--version", "somethingelse"]]:
      try:
        cli._ParseArgs("test", argv, {}, {}, set())
      except cli._ShowVersion:
        pass
      else:
        self.fail("Did not raise exception")

  def testHelp(self):
    for argv in [["test", "--help"], ["test", "--help", "somethingelse"]]:
      try:
        cli._ParseArgs("test", argv, {}, {}, set())
      except cli._ShowUsage, err:
        self.assertFalse(err.exit_error)
      else:
        self.fail("Did not raise exception")

  def testUnknownCommandOrAlias(self):
    for argv in [["test", "list"], ["test", "somethingelse", "--help"]]:
      try:
        cli._ParseArgs("test", argv, {}, {}, set())
      except cli._ShowUsage, err:
        self.assertTrue(err.exit_error)
      else:
        self.fail("Did not raise exception")

  def testInvalidAliasList(self):
    cmd = {
      "list": NotImplemented,
      "foo": NotImplemented,
      }
    aliases = {
      "list": NotImplemented,
      "foo": NotImplemented,
      }
    assert sorted(cmd.keys()) == sorted(aliases.keys())
    self.assertRaises(AssertionError, cli._ParseArgs, "test",
                      ["test", "list"], cmd, aliases, set())

  def testAliasForNonExistantCommand(self):
    cmd = {}
    aliases = {
      "list": NotImplemented,
      }
    self.assertRaises(errors.ProgrammerError, cli._ParseArgs, "test",
                      ["test", "list"], cmd, aliases, set())


class TestQftNames(unittest.TestCase):
  def testComplete(self):
    self.assertEqual(frozenset(cli._QFT_NAMES), constants.QFT_ALL)

  def testUnique(self):
    lcnames = [s.lower() for s in cli._QFT_NAMES.values()]
    self.assertFalse(utils.FindDuplicates(lcnames))

  def testUppercase(self):
    for name in cli._QFT_NAMES.values():
      self.assertEqual(name[0], name[0].upper())


class TestFieldDescValues(unittest.TestCase):
  def testKnownKind(self):
    fdef = objects.QueryFieldDefinition(name="aname",
                                        title="Atitle",
                                        kind=constants.QFT_TEXT,
                                        doc="aaa doc aaa")
    self.assertEqual(cli._FieldDescValues(fdef),
                     ["aname", "Text", "Atitle", "aaa doc aaa"])

  def testUnknownKind(self):
    kind = "#foo#"

    self.assertFalse(kind in constants.QFT_ALL)
    self.assertFalse(kind in cli._QFT_NAMES)

    fdef = objects.QueryFieldDefinition(name="zname", title="Ztitle",
                                        kind=kind, doc="zzz doc zzz")
    self.assertEqual(cli._FieldDescValues(fdef),
                     ["zname", kind, "Ztitle", "zzz doc zzz"])


class TestSerializeGenericInfo(unittest.TestCase):
  """Test case for cli._SerializeGenericInfo"""
  def _RunTest(self, data, expected):
    buf = StringIO()
    cli._SerializeGenericInfo(buf, data, 0)
    self.assertEqual(buf.getvalue(), expected)

  def testSimple(self):
    test_samples = [
      ("abc", "abc\n"),
      ([], "\n"),
      ({}, "\n"),
      (["1", "2", "3"], "- 1\n- 2\n- 3\n"),
      ([("z", "26")], "z: 26\n"),
      ({"z": "26"}, "z: 26\n"),
      ([("z", "26"), ("a", "1")], "z: 26\na: 1\n"),
      ({"z": "26", "a": "1"}, "a: 1\nz: 26\n"),
      ]
    for (data, expected) in test_samples:
      self._RunTest(data, expected)

  def testLists(self):
    adict = {
      "aa": "11",
      "bb": "22",
      "cc": "33",
      }
    adict_exp = ("- aa: 11\n"
                 "  bb: 22\n"
                 "  cc: 33\n")
    anobj = [
      ("zz", "11"),
      ("ww", "33"),
      ("xx", "22"),
      ]
    anobj_exp = ("- zz: 11\n"
                 "  ww: 33\n"
                 "  xx: 22\n")
    alist = ["aa", "cc", "bb"]
    alist_exp = ("- - aa\n"
                 "  - cc\n"
                 "  - bb\n")
    test_samples = [
      (adict, adict_exp),
      (anobj, anobj_exp),
      (alist, alist_exp),
      ]
    for (base_data, base_expected) in test_samples:
      for k in range(1, 4):
        data = k * [base_data]
        expected = k * base_expected
        self._RunTest(data, expected)

  def testDictionaries(self):
    data = [
      ("aaa", ["x", "y"]),
      ("bbb", {
          "w": "1",
          "z": "2",
          }),
      ("ccc", [
          ("xyz", "123"),
          ("efg", "456"),
          ]),
      ]
    expected = ("aaa: \n"
                "  - x\n"
                "  - y\n"
                "bbb: \n"
                "  w: 1\n"
                "  z: 2\n"
                "ccc: \n"
                "  xyz: 123\n"
                "  efg: 456\n")
    self._RunTest(data, expected)
    self._RunTest(dict(data), expected)


class TestFormatPolicyInfo(unittest.TestCase):
  """Test case for cli.FormatPolicyInfo.

  These tests rely on cli._SerializeGenericInfo (tested elsewhere).

  """
  def setUp(self):
    # Policies are big, and we want to see the difference in case of an error
    self.maxDiff = None

  def _RenameDictItem(self, parsed, old, new):
    self.assertTrue(old in parsed)
    self.assertTrue(new not in parsed)
    parsed[new] = parsed[old]
    del parsed[old]

  def _TranslateParsedNames(self, parsed):
    for (pretty, raw) in [
      ("bounds specs", constants.ISPECS_MINMAX),
      ("allowed disk templates", constants.IPOLICY_DTS)
      ]:
      self._RenameDictItem(parsed, pretty, raw)
    for minmax in parsed[constants.ISPECS_MINMAX]:
      for key in minmax:
        keyparts = key.split("/", 1)
        if len(keyparts) > 1:
          self._RenameDictItem(minmax, key, keyparts[0])
    self.assertTrue(constants.IPOLICY_DTS in parsed)
    parsed[constants.IPOLICY_DTS] = yaml.load("[%s]" %
                                              parsed[constants.IPOLICY_DTS])

  @staticmethod
  def _PrintAndParsePolicy(custom, effective, iscluster):
    formatted = cli.FormatPolicyInfo(custom, effective, iscluster)
    buf = StringIO()
    cli._SerializeGenericInfo(buf, formatted, 0)
    return yaml.load(buf.getvalue())

  def _PrintAndCheckParsed(self, policy):
    parsed = self._PrintAndParsePolicy(policy, NotImplemented, True)
    self._TranslateParsedNames(parsed)
    self.assertEqual(parsed, policy)

  def _CompareClusterGroupItems(self, cluster, group, skip=None):
    if isinstance(group, dict):
      self.assertTrue(isinstance(cluster, dict))
      if skip is None:
        skip = frozenset()
      self.assertEqual(frozenset(cluster.keys()).difference(skip),
                       frozenset(group.keys()))
      for key in group:
        self._CompareClusterGroupItems(cluster[key], group[key])
    elif isinstance(group, list):
      self.assertTrue(isinstance(cluster, list))
      self.assertEqual(len(cluster), len(group))
      for (cval, gval) in zip(cluster, group):
        self._CompareClusterGroupItems(cval, gval)
    else:
      self.assertTrue(isinstance(group, basestring))
      self.assertEqual("default (%s)" % cluster, group)

  def _TestClusterVsGroup(self, policy):
    cluster = self._PrintAndParsePolicy(policy, NotImplemented, True)
    group = self._PrintAndParsePolicy({}, policy, False)
    self._CompareClusterGroupItems(cluster, group, ["std"])

  def testWithDefaults(self):
    self._PrintAndCheckParsed(constants.IPOLICY_DEFAULTS)
    self._TestClusterVsGroup(constants.IPOLICY_DEFAULTS)


class TestCreateIPolicyFromOpts(unittest.TestCase):
  """Test case for cli.CreateIPolicyFromOpts."""
  def setUp(self):
    # Policies are big, and we want to see the difference in case of an error
    self.maxDiff = None

  def _RecursiveCheckMergedDicts(self, default_pol, diff_pol, merged_pol,
                                 merge_minmax=False):
    self.assertTrue(type(default_pol) is dict)
    self.assertTrue(type(diff_pol) is dict)
    self.assertTrue(type(merged_pol) is dict)
    self.assertEqual(frozenset(default_pol.keys()),
                     frozenset(merged_pol.keys()))
    for (key, val) in merged_pol.items():
      if key in diff_pol:
        if type(val) is dict:
          self._RecursiveCheckMergedDicts(default_pol[key], diff_pol[key], val)
        elif (merge_minmax and key == "minmax" and type(val) is list and
              len(val) == 1):
          self.assertEqual(len(default_pol[key]), 1)
          self.assertEqual(len(diff_pol[key]), 1)
          self._RecursiveCheckMergedDicts(default_pol[key][0],
                                          diff_pol[key][0], val[0])
        else:
          self.assertEqual(val, diff_pol[key])
      else:
        self.assertEqual(val, default_pol[key])

  def testClusterPolicy(self):
    pol0 = cli.CreateIPolicyFromOpts(
      ispecs_mem_size={},
      ispecs_cpu_count={},
      ispecs_disk_count={},
      ispecs_disk_size={},
      ispecs_nic_count={},
      ipolicy_disk_templates=None,
      ipolicy_vcpu_ratio=None,
      ipolicy_spindle_ratio=None,
      fill_all=True
      )
    self.assertEqual(pol0, constants.IPOLICY_DEFAULTS)

    exp_pol1 = {
      constants.ISPECS_MINMAX: [
        {
          constants.ISPECS_MIN: {
            constants.ISPEC_CPU_COUNT: 2,
            constants.ISPEC_DISK_COUNT: 1,
            },
          constants.ISPECS_MAX: {
            constants.ISPEC_MEM_SIZE: 12*1024,
            constants.ISPEC_DISK_COUNT: 2,
            },
          },
        ],
      constants.ISPECS_STD: {
        constants.ISPEC_CPU_COUNT: 2,
        constants.ISPEC_DISK_COUNT: 2,
        },
      constants.IPOLICY_VCPU_RATIO: 3.1,
      }
    pol1 = cli.CreateIPolicyFromOpts(
      ispecs_mem_size={"max": "12g"},
      ispecs_cpu_count={"min": 2, "std": 2},
      ispecs_disk_count={"min": 1, "max": 2, "std": 2},
      ispecs_disk_size={},
      ispecs_nic_count={},
      ipolicy_disk_templates=None,
      ipolicy_vcpu_ratio=3.1,
      ipolicy_spindle_ratio=None,
      fill_all=True
      )
    self._RecursiveCheckMergedDicts(constants.IPOLICY_DEFAULTS,
                                    exp_pol1, pol1, merge_minmax=True)

    exp_pol2 = {
      constants.ISPECS_MINMAX: [
        {
          constants.ISPECS_MIN: {
            constants.ISPEC_DISK_SIZE: 512,
            constants.ISPEC_NIC_COUNT: 2,
            },
          constants.ISPECS_MAX: {
            constants.ISPEC_NIC_COUNT: 3,
            },
          },
        ],
      constants.ISPECS_STD: {
        constants.ISPEC_CPU_COUNT: 2,
        constants.ISPEC_NIC_COUNT: 3,
        },
      constants.IPOLICY_SPINDLE_RATIO: 1.3,
      constants.IPOLICY_DTS: ["templates"],
      }
    pol2 = cli.CreateIPolicyFromOpts(
      ispecs_mem_size={},
      ispecs_cpu_count={"std": 2},
      ispecs_disk_count={},
      ispecs_disk_size={"min": "0.5g"},
      ispecs_nic_count={"min": 2, "max": 3, "std": 3},
      ipolicy_disk_templates=["templates"],
      ipolicy_vcpu_ratio=None,
      ipolicy_spindle_ratio=1.3,
      fill_all=True
      )
    self._RecursiveCheckMergedDicts(constants.IPOLICY_DEFAULTS,
                                      exp_pol2, pol2, merge_minmax=True)

    for fill_all in [False, True]:
      exp_pol3 = {
        constants.ISPECS_STD: {
          constants.ISPEC_CPU_COUNT: 2,
          constants.ISPEC_NIC_COUNT: 3,
          },
        }
      pol3 = cli.CreateIPolicyFromOpts(
        std_ispecs={
          constants.ISPEC_CPU_COUNT: "2",
          constants.ISPEC_NIC_COUNT: "3",
          },
        ipolicy_disk_templates=None,
        ipolicy_vcpu_ratio=None,
        ipolicy_spindle_ratio=None,
        fill_all=fill_all
        )
      if fill_all:
        self._RecursiveCheckMergedDicts(constants.IPOLICY_DEFAULTS,
                                        exp_pol3, pol3, merge_minmax=True)
      else:
        self.assertEqual(pol3, exp_pol3)

  def testPartialPolicy(self):
    exp_pol0 = objects.MakeEmptyIPolicy()
    pol0 = cli.CreateIPolicyFromOpts(
      minmax_ispecs=None,
      std_ispecs=None,
      ipolicy_disk_templates=None,
      ipolicy_vcpu_ratio=None,
      ipolicy_spindle_ratio=None,
      fill_all=False
      )
    self.assertEqual(pol0, exp_pol0)

    exp_pol1 = {
      constants.IPOLICY_VCPU_RATIO: 3.1,
      }
    pol1 = cli.CreateIPolicyFromOpts(
      minmax_ispecs=None,
      std_ispecs=None,
      ipolicy_disk_templates=None,
      ipolicy_vcpu_ratio=3.1,
      ipolicy_spindle_ratio=None,
      fill_all=False
      )
    self.assertEqual(pol1, exp_pol1)

    exp_pol2 = {
      constants.IPOLICY_SPINDLE_RATIO: 1.3,
      constants.IPOLICY_DTS: ["templates"],
      }
    pol2 = cli.CreateIPolicyFromOpts(
      minmax_ispecs=None,
      std_ispecs=None,
      ipolicy_disk_templates=["templates"],
      ipolicy_vcpu_ratio=None,
      ipolicy_spindle_ratio=1.3,
      fill_all=False
      )
    self.assertEqual(pol2, exp_pol2)

  def _TestInvalidISpecs(self, minmax_ispecs, std_ispecs, fail=True):
    for fill_all in [False, True]:
      if fail:
        self.assertRaises((errors.OpPrereqError,
                           errors.UnitParseError,
                           errors.TypeEnforcementError),
                          cli.CreateIPolicyFromOpts,
                          minmax_ispecs=minmax_ispecs,
                          std_ispecs=std_ispecs,
                          fill_all=fill_all)
      else:
        cli.CreateIPolicyFromOpts(minmax_ispecs=minmax_ispecs,
                                  std_ispecs=std_ispecs,
                                  fill_all=fill_all)

  def testInvalidPolicies(self):
    self.assertRaises(AssertionError, cli.CreateIPolicyFromOpts,
                      std_ispecs={constants.ISPEC_MEM_SIZE: 1024},
                      ipolicy_disk_templates=None, ipolicy_vcpu_ratio=None,
                      ipolicy_spindle_ratio=None, group_ipolicy=True)
    self.assertRaises(errors.OpPrereqError, cli.CreateIPolicyFromOpts,
                      ispecs_mem_size={"wrong": "x"}, ispecs_cpu_count={},
                      ispecs_disk_count={}, ispecs_disk_size={},
                      ispecs_nic_count={}, ipolicy_disk_templates=None,
                      ipolicy_vcpu_ratio=None, ipolicy_spindle_ratio=None,
                      fill_all=True)
    self.assertRaises(errors.TypeEnforcementError, cli.CreateIPolicyFromOpts,
                      ispecs_mem_size={}, ispecs_cpu_count={"min": "default"},
                      ispecs_disk_count={}, ispecs_disk_size={},
                      ispecs_nic_count={}, ipolicy_disk_templates=None,
                      ipolicy_vcpu_ratio=None, ipolicy_spindle_ratio=None,
                      fill_all=True)

    good_mmspecs = [
      constants.ISPECS_MINMAX_DEFAULTS,
      constants.ISPECS_MINMAX_DEFAULTS,
      ]
    self._TestInvalidISpecs(good_mmspecs, None, fail=False)
    broken_mmspecs = copy.deepcopy(good_mmspecs)
    for minmaxpair in broken_mmspecs:
      for key in constants.ISPECS_MINMAX_KEYS:
        for par in constants.ISPECS_PARAMETERS:
          old = minmaxpair[key][par]
          del minmaxpair[key][par]
          self._TestInvalidISpecs(broken_mmspecs, None)
          minmaxpair[key][par] = "invalid"
          self._TestInvalidISpecs(broken_mmspecs, None)
          minmaxpair[key][par] = old
        minmaxpair[key]["invalid_key"] = None
        self._TestInvalidISpecs(broken_mmspecs, None)
        del minmaxpair[key]["invalid_key"]
      minmaxpair["invalid_key"] = None
      self._TestInvalidISpecs(broken_mmspecs, None)
      del minmaxpair["invalid_key"]
      assert broken_mmspecs == good_mmspecs

    good_stdspecs = constants.IPOLICY_DEFAULTS[constants.ISPECS_STD]
    self._TestInvalidISpecs(None, good_stdspecs, fail=False)
    broken_stdspecs = copy.deepcopy(good_stdspecs)
    for par in constants.ISPECS_PARAMETERS:
      old = broken_stdspecs[par]
      broken_stdspecs[par] = "invalid"
      self._TestInvalidISpecs(None, broken_stdspecs)
      broken_stdspecs[par] = old
    broken_stdspecs["invalid_key"] = None
    self._TestInvalidISpecs(None, broken_stdspecs)
    del broken_stdspecs["invalid_key"]
    assert broken_stdspecs == good_stdspecs

  def testAllowedValues(self):
    allowedv = "blah"
    exp_pol1 = {
      constants.ISPECS_MINMAX: allowedv,
      constants.IPOLICY_DTS: allowedv,
      constants.IPOLICY_VCPU_RATIO: allowedv,
      constants.IPOLICY_SPINDLE_RATIO: allowedv,
      }
    pol1 = cli.CreateIPolicyFromOpts(minmax_ispecs=[{allowedv: {}}],
                                     std_ispecs=None,
                                     ipolicy_disk_templates=allowedv,
                                     ipolicy_vcpu_ratio=allowedv,
                                     ipolicy_spindle_ratio=allowedv,
                                     allowed_values=[allowedv])
    self.assertEqual(pol1, exp_pol1)

  @staticmethod
  def _ConvertSpecToStrings(spec):
    ret = {}
    for (par, val) in spec.items():
      ret[par] = str(val)
    return ret

  def _CheckNewStyleSpecsCall(self, exp_ipolicy, minmax_ispecs, std_ispecs,
                              group_ipolicy, fill_all):
    ipolicy = cli.CreateIPolicyFromOpts(minmax_ispecs=minmax_ispecs,
                                        std_ispecs=std_ispecs,
                                        group_ipolicy=group_ipolicy,
                                        fill_all=fill_all)
    self.assertEqual(ipolicy, exp_ipolicy)

  def _TestFullISpecsInner(self, skel_exp_ipol, exp_minmax, exp_std,
                           group_ipolicy, fill_all):
    exp_ipol = skel_exp_ipol.copy()
    if exp_minmax is not None:
      minmax_ispecs = []
      for exp_mm_pair in exp_minmax:
        mmpair = {}
        for (key, spec) in exp_mm_pair.items():
          mmpair[key] = self._ConvertSpecToStrings(spec)
        minmax_ispecs.append(mmpair)
      exp_ipol[constants.ISPECS_MINMAX] = exp_minmax
    else:
      minmax_ispecs = None
    if exp_std is not None:
      std_ispecs = self._ConvertSpecToStrings(exp_std)
      exp_ipol[constants.ISPECS_STD] = exp_std
    else:
      std_ispecs = None

    self._CheckNewStyleSpecsCall(exp_ipol, minmax_ispecs, std_ispecs,
                                 group_ipolicy, fill_all)
    if minmax_ispecs:
      for mmpair in minmax_ispecs:
        for (key, spec) in mmpair.items():
          for par in [constants.ISPEC_MEM_SIZE, constants.ISPEC_DISK_SIZE]:
            if par in spec:
              spec[par] += "m"
              self._CheckNewStyleSpecsCall(exp_ipol, minmax_ispecs, std_ispecs,
                                           group_ipolicy, fill_all)
    if std_ispecs:
      for par in [constants.ISPEC_MEM_SIZE, constants.ISPEC_DISK_SIZE]:
        if par in std_ispecs:
          std_ispecs[par] += "m"
          self._CheckNewStyleSpecsCall(exp_ipol, minmax_ispecs, std_ispecs,
                                       group_ipolicy, fill_all)

  def testFullISpecs(self):
    exp_minmax1 = [
      {
        constants.ISPECS_MIN: {
          constants.ISPEC_MEM_SIZE: 512,
          constants.ISPEC_CPU_COUNT: 2,
          constants.ISPEC_DISK_COUNT: 2,
          constants.ISPEC_DISK_SIZE: 512,
          constants.ISPEC_NIC_COUNT: 2,
          constants.ISPEC_SPINDLE_USE: 2,
          },
        constants.ISPECS_MAX: {
          constants.ISPEC_MEM_SIZE: 768*1024,
          constants.ISPEC_CPU_COUNT: 7,
          constants.ISPEC_DISK_COUNT: 6,
          constants.ISPEC_DISK_SIZE: 2048*1024,
          constants.ISPEC_NIC_COUNT: 3,
          constants.ISPEC_SPINDLE_USE: 3,
          },
        },
      ]
    exp_minmax2 = [
      {
        constants.ISPECS_MIN: {
          constants.ISPEC_MEM_SIZE: 512,
          constants.ISPEC_CPU_COUNT: 2,
          constants.ISPEC_DISK_COUNT: 2,
          constants.ISPEC_DISK_SIZE: 512,
          constants.ISPEC_NIC_COUNT: 2,
          constants.ISPEC_SPINDLE_USE: 2,
          },
        constants.ISPECS_MAX: {
          constants.ISPEC_MEM_SIZE: 768*1024,
          constants.ISPEC_CPU_COUNT: 7,
          constants.ISPEC_DISK_COUNT: 6,
          constants.ISPEC_DISK_SIZE: 2048*1024,
          constants.ISPEC_NIC_COUNT: 3,
          constants.ISPEC_SPINDLE_USE: 3,
          },
        },
      {
        constants.ISPECS_MIN: {
          constants.ISPEC_MEM_SIZE: 1024*1024,
          constants.ISPEC_CPU_COUNT: 3,
          constants.ISPEC_DISK_COUNT: 3,
          constants.ISPEC_DISK_SIZE: 256,
          constants.ISPEC_NIC_COUNT: 4,
          constants.ISPEC_SPINDLE_USE: 5,
          },
        constants.ISPECS_MAX: {
          constants.ISPEC_MEM_SIZE: 2048*1024,
          constants.ISPEC_CPU_COUNT: 5,
          constants.ISPEC_DISK_COUNT: 5,
          constants.ISPEC_DISK_SIZE: 1024*1024,
          constants.ISPEC_NIC_COUNT: 5,
          constants.ISPEC_SPINDLE_USE: 7,
          },
        },
      ]
    exp_std1 = {
      constants.ISPEC_MEM_SIZE: 768*1024,
      constants.ISPEC_CPU_COUNT: 7,
      constants.ISPEC_DISK_COUNT: 6,
      constants.ISPEC_DISK_SIZE: 2048*1024,
      constants.ISPEC_NIC_COUNT: 3,
      constants.ISPEC_SPINDLE_USE: 1,
      }
    for fill_all in [False, True]:
      if fill_all:
        skel_ipolicy = constants.IPOLICY_DEFAULTS
      else:
        skel_ipolicy = {}
      self._TestFullISpecsInner(skel_ipolicy, None, exp_std1,
                                False, fill_all)
      for exp_minmax in [exp_minmax1, exp_minmax2]:
        self._TestFullISpecsInner(skel_ipolicy, exp_minmax, exp_std1,
                                  False, fill_all)
        self._TestFullISpecsInner(skel_ipolicy, exp_minmax, None,
                                  False, fill_all)


class TestPrintIPolicyCommand(unittest.TestCase):
  """Test case for cli.PrintIPolicyCommand"""
  _SPECS1 = {
    "par1": 42,
    "par2": "xyz",
    }
  _SPECS1_STR = "par1=42,par2=xyz"
  _SPECS2 = {
    "param": 10,
    "another_param": 101,
    }
  _SPECS2_STR = "another_param=101,param=10"
  _SPECS3 = {
    "par1": 1024,
    "param": "abc",
    }
  _SPECS3_STR = "par1=1024,param=abc"

  def _CheckPrintIPolicyCommand(self, ipolicy, isgroup, expected):
    buf = StringIO()
    cli.PrintIPolicyCommand(buf, ipolicy, isgroup)
    self.assertEqual(buf.getvalue(), expected)

  def testIgnoreStdForGroup(self):
    self._CheckPrintIPolicyCommand({"std": self._SPECS1}, True, "")

  def testIgnoreEmpty(self):
    policies = [
      {},
      {"std": {}},
      {"minmax": []},
      {"minmax": [{}]},
      {"minmax": [{
        "min": {},
        "max": {},
        }]},
      {"minmax": [{
        "min": self._SPECS1,
        "max": {},
        }]},
      ]
    for pol in policies:
      self._CheckPrintIPolicyCommand(pol, False, "")

  def testFullPolicies(self):
    cases = [
      ({"std": self._SPECS1},
       " %s %s" % (cli.IPOLICY_STD_SPECS_STR, self._SPECS1_STR)),
      ({"minmax": [{
        "min": self._SPECS1,
        "max": self._SPECS2,
        }]},
       " %s min:%s/max:%s" % (cli.IPOLICY_BOUNDS_SPECS_STR,
                              self._SPECS1_STR, self._SPECS2_STR)),
      ({"minmax": [
        {
          "min": self._SPECS1,
          "max": self._SPECS2,
          },
        {
          "min": self._SPECS2,
          "max": self._SPECS3,
          },
        ]},
       " %s min:%s/max:%s//min:%s/max:%s" %
       (cli.IPOLICY_BOUNDS_SPECS_STR, self._SPECS1_STR, self._SPECS2_STR,
        self._SPECS2_STR, self._SPECS3_STR)),
      ]
    for (pol, exp) in cases:
      self._CheckPrintIPolicyCommand(pol, False, exp)


if __name__ == "__main__":
  testutils.GanetiTestProgram()