Synnefo » snf-ganeti » ganeti-local

  @classmethod

  def _ParseLvInfoLine(cls, line, sep):

    """Parse one line of the lvs output used in L{_GetLvInfo}.

    elems = line.strip().rstrip(sep).split(sep)

    if len(elems) != 5:

      base.ThrowError("Can't parse LVS output, len(%s) != 5", str(elems))

    (status, major, minor, pe_size, stripes) = elems

      base.ThrowError("lvs lv_attr is not at least 6 characters (%s)", status)

      base.ThrowError("lvs major/minor cannot be parsed: %s", str(err))

      base.ThrowError("Can't parse vg extent size: %s", err)

      base.ThrowError("Can't parse the number of stripes: %s", err)

    return (status, major, minor, pe_size, stripes)

  @classmethod

  def _GetLvInfo(cls, dev_path, _run_cmd=utils.RunCmd):

    """Get info about the given existing LV to be used.

    """

    result = _run_cmd(["lvs", "--noheadings", "--separator=,",

                       "--units=k", "--nosuffix",

                       "-olv_attr,lv_kernel_major,lv_kernel_minor,"

                       "vg_extent_size,stripes", dev_path])

    if result.failed:

      base.ThrowError("Can't find LV %s: %s, %s",

                      dev_path, result.fail_reason, result.output)

    # the output can (and will) have multiple lines for multi-segment

    # LVs, as the 'stripes' parameter is a segment one, so we take

    # only the last entry, which is the one we're interested in; note

    # that with LVM2 anyway the 'stripes' value must be constant

    # across segments, so this is a no-op actually

    out = result.stdout.splitlines()

    if not out: # totally empty result? splitlines() returns at least

                # one line for any non-empty string

      base.ThrowError("Can't parse LVS output, no lines? Got '%s'", str(out))

    return cls._ParseLvInfoLine(out[-1], ",")

  def Attach(self):

    """Attach to an existing LV.

    This method will try to see if an existing and active LV exists

    which matches our name. If so, its major/minor will be

    recorded.

    """

    self.attached = False

    try:

      (status, major, minor, pe_size, stripes) = \

        self._GetLvInfo(self.dev_path)

    except errors.BlockDeviceError:

class TestLogicalVolume(unittest.TestCase):

  """Tests for bdev.LogicalVolume."""

  def testParseLvInfoLine(self):

    """Tests for LogicalVolume._ParseLvInfoLine."""

    broken_lines = [

      "  toomuch#-wi-ao#253#3#4096.00#2",

      "  -wi-ao#253#3#4096.00",

      "  -wi-a#253#3#4096.00#2",

      "  -wi-ao#25.3#3#4096.00#2",

      "  -wi-ao#twenty#3#4096.00#2",

      "  -wi-ao#253#3.1#4096.00#2",

      "  -wi-ao#253#three#4096.00#2",

      "  -wi-ao#253#3#four#2",

      "  -wi-ao#253#3#4096..00#2",

      "  -wi-ao#253#3#4096.00#2.0",

      "  -wi-ao#253#3#4096.00#two",

      ]

    for broken in broken_lines:

      self.assertRaises(errors.BlockDeviceError,

                        bdev.LogicalVolume._ParseLvInfoLine, broken, "#")

    true_out = [

      ("-wi-ao", 253, 3, 4096.00, 2),

      ("-wi-a-", 253, 7, 4096.00, 4),

      ("-ri-a-", 253, 4, 4.00, 5),

      ("-wc-ao", 15, 18, 4096.00, 32),

      ]

    for exp in true_out:

      for sep in "#;|,":

        lvs_line = sep.join(("  %s", "%d", "%d", "%.2f", "%d")) % exp

        parsed = bdev.LogicalVolume._ParseLvInfoLine(lvs_line, sep)

        self.assertEqual(parsed, exp)

  @staticmethod

  def _FakeRunCmd(success, stdout):

    if success:

      exit_code = 0

    else:

      exit_code = 1

    return lambda cmd: utils.RunResult(exit_code, None, stdout, "", cmd,

                                       utils.process._TIMEOUT_NONE, 5)

  def testGetLvInfo(self):

    """Tests for LogicalVolume._GetLvInfo."""

    self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume._GetLvInfo,

                      "fake_path",

                      _run_cmd=self._FakeRunCmd(False, "Fake error msg"))

    self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume._GetLvInfo,

                      "fake_path", _run_cmd=self._FakeRunCmd(True, ""))

    self.assertRaises(errors.BlockDeviceError, bdev.LogicalVolume._GetLvInfo,

                      "fake_path", _run_cmd=self._FakeRunCmd(True, "BadStdOut"))

    good_line = "  -wi-ao,253,3,4096.00,2"

    fake_cmd = self._FakeRunCmd(True, good_line)

    good_res = bdev.LogicalVolume._GetLvInfo("fake_path", _run_cmd=fake_cmd)

    # Only the last line should be parsed and taken into account

    for lines in [

      [good_line] * 2,

      [good_line] * 3,

      ["bad line", good_line],

      ]:

      fake_cmd = self._FakeRunCmd(True, "\n".join(lines))

      same_res = bdev.LogicalVolume._GetLvInfo("fake_path", fake_cmd)

      self.assertEqual(same_res, good_res)