Synnefo » snf-ganeti » ganeti-local

def PrepareContainerMods(mods, private_fn):

  """Prepares a list of container modifications by adding a private data field.

  @type mods: list of tuples; (operation, index, parameters)

  @param mods: List of modifications

  @type private_fn: callable or None

  @param private_fn: Callable for constructing a private data field for a

    modification

  @rtype: list

  """

  if private_fn is None:

    fn = lambda: None

  else:

    fn = private_fn

  return [(op, idx, params, fn()) for (op, idx, params) in mods]

def ApplyContainerMods(kind, container, chgdesc, mods,

                       create_fn, modify_fn, remove_fn):

  """Applies descriptions in C{mods} to C{container}.

  @type kind: string

  @param kind: One-word item description

  @type container: list

  @param container: Container to modify

  @type chgdesc: None or list

  @param chgdesc: List of applied changes

  @type mods: list

  @param mods: Modifications as returned by L{PrepareContainerMods}

  @type create_fn: callable

  @param create_fn: Callback for creating a new item (L{constants.DDM_ADD});

    receives absolute item index, parameters, list of applied changes and

    private data object as added by L{PrepareContainerMods}, returns new item

  @type modify_fn: callable

  @param modify_fn: Callback for modifying an existing item

    (L{constants.DDM_MODIFY}); receives absolute item index, item, parameters,

    list of applied changes and private data object as added by

    L{PrepareContainerMods}

  @type remove_fn: callable

  @param remove_fn: Callback on removing item; receives absolute item index,

    item, list of applied changes and private data object as added by

    L{PrepareContainerMods}

  """

  for (op, idx, params, private) in mods:

    if idx == -1:

      # Append

      absidx = len(container) - 1

    elif idx < 0:

      raise IndexError("Not accepting negative indices")

    else:

      absidx = idx

    if op == constants.DDM_ADD:

      if create_fn is None:

        item = params

      else:

        item = create_fn(absidx + 1, params, chgdesc, private)

      if idx == -1:

        container.append(item)

      else:

        assert idx >= 0

        # list.insert does so before the specified index

        container.insert(idx, item)

    else:

      # Retrieve existing item

      try:

        item = container[absidx]

      except IndexError:

        raise IndexError("Invalid %s index %s" % (kind, idx))

      if op == constants.DDM_REMOVE:

        assert not params

        if remove_fn is not None:

          remove_fn(absidx, item, chgdesc, private)

        if chgdesc is not None:

          chgdesc.append(("%s/%s" % (kind, absidx), "remove"))

        assert container[absidx] == item

        del container[absidx]

      elif op == constants.DDM_MODIFY:

        if modify_fn is not None:

          modify_fn(absidx, item, params, chgdesc, private)

      else:

        raise errors.ProgrammerError("Unhandled operation '%s'" % op)

class TestApplyContainerMods(unittest.TestCase):

  def testEmptyContainer(self):

    container = []

    chgdesc = []

    cmdlib.ApplyContainerMods("test", container, chgdesc, [], None, None, None)

    self.assertEqual(container, [])

    self.assertEqual(chgdesc, [])

  def testAdd(self):

    container = []

    chgdesc = []

    mods = cmdlib.PrepareContainerMods([

      (constants.DDM_ADD, -1, "Hello"),

      (constants.DDM_ADD, -1, "World"),

      (constants.DDM_ADD, 0, "Start"),

      (constants.DDM_ADD, -1, "End"),

      ], None)

    cmdlib.ApplyContainerMods("test", container, chgdesc, mods,

                              None, None, None)

    self.assertEqual(container, ["Start", "Hello", "World", "End"])

    self.assertEqual(chgdesc, [])

  def testRemoveError(self):

    for idx in [0, 1, 2, 100, -1, -4]:

      mods = cmdlib.PrepareContainerMods([

        (constants.DDM_REMOVE, idx, None),

        ], None)

      self.assertRaises(IndexError, cmdlib.ApplyContainerMods,

                        "test", [], None, mods, None, None, None)

    mods = cmdlib.PrepareContainerMods([

      (constants.DDM_REMOVE, 0, object()),

      ], None)

    self.assertRaises(AssertionError, cmdlib.ApplyContainerMods,

                      "test", [""], None, mods, None, None, None)

  def testAddError(self):

    for idx in range(-100, -1):

      mods = cmdlib.PrepareContainerMods([

        (constants.DDM_ADD, idx, None),

        ], None)

      self.assertRaises(IndexError, cmdlib.ApplyContainerMods,

                        "test", [], None, mods, None, None, None)

  def testRemove(self):

    container = ["item 1", "item 2"]

    mods = cmdlib.PrepareContainerMods([

      (constants.DDM_ADD, -1, "aaa"),

      (constants.DDM_REMOVE, -1, None),

      (constants.DDM_ADD, -1, "bbb"),

      ], None)

    chgdesc = []

    cmdlib.ApplyContainerMods("test", container, chgdesc, mods,

                              None, None, None)

    self.assertEqual(container, ["item 1", "item 2", "bbb"])

    self.assertEqual(chgdesc, [

      ("test/2", "remove"),

      ])

  class _PrivateData:

    def __init__(self):

      self.data = None

  @staticmethod

  def _CreateTestFn(idx, params, chgdesc, private):

    chgdesc.append(("test/%s" % idx, hex(idx)))

    private.data = ("add", idx, params)

    return (100 * idx, params)

  @staticmethod

  def _ModifyTestFn(idx, item, params, chgdesc, private):

    chgdesc.append(("test/%s" % idx, "modify %s" % params))

    private.data = ("modify", idx, params)

  @staticmethod

  def _RemoveTestFn(idx, item, chgdesc, private):

    private.data = ("remove", idx, item)

  def testAddWithCreateFunction(self):

    container = []

    chgdesc = []

    mods = cmdlib.PrepareContainerMods([

      (constants.DDM_ADD, -1, "Hello"),

      (constants.DDM_ADD, -1, "World"),

      (constants.DDM_ADD, 0, "Start"),

      (constants.DDM_ADD, -1, "End"),

      (constants.DDM_REMOVE, 2, None),

      (constants.DDM_MODIFY, -1, "foobar"),

      (constants.DDM_REMOVE, 2, None),

      (constants.DDM_ADD, 1, "More"),

      ], self._PrivateData)

    cmdlib.ApplyContainerMods("test", container, chgdesc, mods,

      self._CreateTestFn, self._ModifyTestFn, self._RemoveTestFn)

    self.assertEqual(container, [

      (100, "Start"),

      (200, "More"),

      (0, "Hello"),

      ])

    self.assertEqual(chgdesc, [

      ("test/0", "0x0"),

      ("test/1", "0x1"),

      ("test/1", "0x1"),

      ("test/3", "0x3"),

      ("test/2", "remove"),

      ("test/2", "modify foobar"),

      ("test/2", "remove"),

      ("test/2", "0x2")

      ])

    self.assertTrue(compat.all(op == private.data[0]

                               for (op, _, _, private) in mods))

    self.assertEqual([private.data for (op, _, _, private) in mods], [

      ("add", 0, "Hello"),

      ("add", 1, "World"),

      ("add", 1, "Start"),

      ("add", 3, "End"),

      ("remove", 2, (100, "World")),

      ("modify", 2, "foobar"),

      ("remove", 2, (300, "End")),

      ("add", 2, "More"),

      ])