Synnefo » snf-ganeti

TESTS = ganeti.hooks_unittest.py ganeti.utils_unittest.py

TESTS_ENVIRONMENT = PYTHONPATH=.:$(srcdir)

check_DATA = ganeti

ganeti:

	rm -f ganeti

	ln -s $(top_srcdir)/lib ganeti

EXTRA_DIST = $(TESTS) mocks.py

CLEANFILES = *.py[co]

MAINTAINERCLEANFILES = ganeti

#!/usr/bin/python

#

# Copyright (C) 2006, 2007 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for unittesting the config module"""

import unittest

import os

import time

import tempfile

import os.path

import socket

from ganeti import errors

from ganeti import constants

from ganeti import config

from ganeti import objects

class TestConfigRunner(unittest.TestCase):

  """Testing case for HooksRunner"""

  def setUp(self):

    fd, self.cfg_file = tempfile.mkstemp()

    os.close(fd)

  def tearDown(self):

    try:

      os.unlink(self.cfg_file)

    except OSError:

      pass

  def _get_object(self):

    """Returns a instance of ConfigWriter"""

    cfg = config.ConfigWriter(cfg_file=self.cfg_file, offline=True)

    return cfg

  def _init_cluster(self, cfg):

    """Initializes the cfg object"""

    cfg.InitConfig(socket.gethostname(), '127.0.0.1', None, '', 'aa:00:00',

                   'xenvg', constants.DEFAULT_BRIDGE)

  def _create_instance(self):

    """Create and return an instance object"""

    inst = objects.Instance(name="test.example.com", disks=[],

                            disk_template=constants.DT_DISKLESS)

    return inst

  def testEmpty(self):

    """Test instantiate config object"""

    self._get_object()

  def testInit(self):

    """Test initialize the config file"""

    cfg = self._get_object()

    self._init_cluster(cfg)

    self.failUnlessEqual(1, len(cfg.GetNodeList()))

    self.failUnlessEqual(0, len(cfg.GetInstanceList()))

  def testUpdateCluster(self):

    """Test updates on the cluster object"""

    cfg = self._get_object()

    # construct a fake cluster object

    fake_cl = objects.Cluster()

    # fail if we didn't read the config

    self.failUnlessRaises(errors.ProgrammerError, cfg.Update, fake_cl)

    self._init_cluster(cfg)

    cl = cfg.GetClusterInfo()

    # first pass, must not fail

    cfg.Update(cl)

    # second pass, also must not fail (after the config has been written)

    cfg.Update(cl)

    # but the fake_cl update should still fail

    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_cl)

  def testUpdateNode(self):

    """Test updates on one node object"""

    cfg = self._get_object()

    # construct a fake node

    fake_node = objects.Node()

    # fail if we didn't read the config

    self.failUnlessRaises(errors.ProgrammerError, cfg.Update, fake_node)

    self._init_cluster(cfg)

    node = cfg.GetNodeInfo(cfg.GetNodeList()[0])

    # first pass, must not fail

    cfg.Update(node)

    # second pass, also must not fail (after the config has been written)

    cfg.Update(node)

    # but the fake_node update should still fail

    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_node)

  def testUpdateInstance(self):

    """Test updates on one instance object"""

    cfg = self._get_object()

    # construct a fake instance

    inst = self._create_instance()

    fake_instance = objects.Instance()

    # fail if we didn't read the config

    self.failUnlessRaises(errors.ProgrammerError, cfg.Update, fake_instance)

    self._init_cluster(cfg)

    cfg.AddInstance(inst)

    instance = cfg.GetInstanceInfo(cfg.GetInstanceList()[0])

    # first pass, must not fail

    cfg.Update(instance)

    # second pass, also must not fail (after the config has been written)

    cfg.Update(instance)

    # but the fake_instance update should still fail

    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_instance)

if __name__ == '__main__':

  unittest.main()

#!/usr/bin/python

#

# Copyright (C) 2006, 2007 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for unittesting the hooks module"""

import unittest

import os

import time

import tempfile

import os.path

from ganeti import errors

from ganeti import opcodes

from ganeti import mcpu

from ganeti import backend

from ganeti import constants

from ganeti import cmdlib

from ganeti.constants import HKR_SUCCESS, HKR_FAIL, HKR_SKIP

from mocks import FakeConfig, FakeSStore

class FakeLU(cmdlib.LogicalUnit):

  HPATH = "test"

  def BuildHooksEnv(self):

    return {}, ["localhost"], ["localhost"]

class TestHooksRunner(unittest.TestCase):

  """Testing case for HooksRunner"""

  def setUp(self):

    self.torm = []

    self.tmpdir = tempfile.mkdtemp()

    self.torm.append((self.tmpdir, True))

    self.logdir = tempfile.mkdtemp()

    self.torm.append((self.logdir, True))

    self.hpath = "fake"

    self.ph_dirs = {}

    for i in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      dname = "%s/%s-%s.d" % (self.tmpdir, self.hpath, i)

      os.mkdir(dname)

      self.torm.append((dname, True))

      self.ph_dirs[i] = dname

    self.hr = backend.HooksRunner(hooks_base_dir=self.tmpdir)

  def tearDown(self):

    self.torm.reverse()

    for path, kind in self.torm:

      if kind:

        os.rmdir(path)

      else:

        os.unlink(path)

  def _rname(self, fname):

    return "/".join(fname.split("/")[-2:])

  def testEmpty(self):

    """Test no hooks"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), [])

  def testSkipNonExec(self):

    """Test skip non-exec file"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      fname = "%s/test" % self.ph_dirs[phase]

      f = open(fname, "w")

      f.close()

      self.torm.append((fname, False))

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),

                           [(self._rname(fname), HKR_SKIP, "")])

  def testSkipInvalidName(self):

    """Test skip script with invalid name"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      fname = "%s/a.off" % self.ph_dirs[phase]

      f = open(fname, "w")

      f.write("#!/bin/sh\nexit 0\n")

      f.close()

      os.chmod(fname, 0700)

      self.torm.append((fname, False))

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),

                           [(self._rname(fname), HKR_SKIP, "")])

  def testSkipDir(self):

    """Test skip directory"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      fname = "%s/testdir" % self.ph_dirs[phase]

      os.mkdir(fname)

      self.torm.append((fname, True))

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),

                           [(self._rname(fname), HKR_SKIP, "")])

  def testSuccess(self):

    """Test success execution"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      fname = "%s/success" % self.ph_dirs[phase]

      f = open(fname, "w")

      f.write("#!/bin/sh\nexit 0\n")

      f.close()

      self.torm.append((fname, False))

      os.chmod(fname, 0700)

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),

                           [(self._rname(fname), HKR_SUCCESS, "")])

  def testSymlink(self):

    """Test running a symlink"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      fname = "%s/success" % self.ph_dirs[phase]

      os.symlink("/bin/true", fname)

      self.torm.append((fname, False))

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),

                           [(self._rname(fname), HKR_SUCCESS, "")])

  def testFail(self):

    """Test success execution"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      fname = "%s/success" % self.ph_dirs[phase]

      f = open(fname, "w")

      f.write("#!/bin/sh\nexit 1\n")

      f.close()

      self.torm.append((fname, False))

      os.chmod(fname, 0700)

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),

                           [(self._rname(fname), HKR_FAIL, "")])

  def testCombined(self):

    """Test success, failure and skip all in one test"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      expect = []

      for fbase, ecode, rs in [("00succ", 0, HKR_SUCCESS),

                               ("10fail", 1, HKR_FAIL),

                               ("20inv.", 0, HKR_SKIP),

                               ]:

        fname = "%s/%s" % (self.ph_dirs[phase], fbase)

        f = open(fname, "w")

        f.write("#!/bin/sh\nexit %d\n" % ecode)

        f.close()

        self.torm.append((fname, False))

        os.chmod(fname, 0700)

        expect.append((self._rname(fname), rs, ""))

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), expect)

  def testOrdering(self):

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      expect = []

      for fbase in ["10s1",

                    "00s0",

                    "10sa",

                    "80sc",

                    "60sd",

                    ]:

        fname = "%s/%s" % (self.ph_dirs[phase], fbase)

        os.symlink("/bin/true", fname)

        self.torm.append((fname, False))

        expect.append((self._rname(fname), HKR_SUCCESS, ""))

      expect.sort()

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), expect)

  def testEnv(self):

    """Test environment execution"""

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      fbase = "success"

      fname = "%s/%s" % (self.ph_dirs[phase], fbase)

      os.symlink("/usr/bin/env", fname)

      self.torm.append((fname, False))

      env_snt = {"PHASE": phase}

      env_exp = "PHASE=%s\n" % phase

      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, env_snt),

                           [(self._rname(fname), HKR_SUCCESS, env_exp)])

class TestHooksMaster(unittest.TestCase):

  """Testing case for HooksMaster"""

  def _call_false(*args):

    """Fake call_hooks_runner function which returns False."""

    return False

  @staticmethod

  def _call_nodes_false(node_list, hpath, phase, env):

    """Fake call_hooks_runner function.

    Returns:

      - list of False values with the same len as the node_list argument

    """

    return [False for node_name in node_list]

  @staticmethod

  def _call_script_fail(node_list, hpath, phase, env):

    """Fake call_hooks_runner function.

    Returns:

      - list of False values with the same len as the node_list argument

    """

    return dict([(node_name, [("unittest", constants.HKR_FAIL, "error")])

                 for node_name in node_list])

  @staticmethod

  def _call_script_succeed(node_list, hpath, phase, env):

    """Fake call_hooks_runner function.

    Returns:

      - list of False values with the same len as the node_list argument

    """

    return dict([(node_name, [("unittest", constants.HKR_SUCCESS, "ok")])

                 for node_name in node_list])

  def testTotalFalse(self):

    """Test complete rpc failure"""

    cfg = FakeConfig()

    sstore = FakeSStore()

    op = opcodes.OpCode()

    lu = FakeLU(None, op, cfg, sstore)

    hm = mcpu.HooksMaster(self._call_false, cfg, sstore, lu)

    self.failUnlessRaises(errors.HooksFailure,

                          hm.RunPhase, constants.HOOKS_PHASE_PRE)

    hm.RunPhase(constants.HOOKS_PHASE_POST)

  def testIndividualFalse(self):

    """Test individual rpc failure"""

    cfg = FakeConfig()

    sstore = FakeSStore()

    op = opcodes.OpCode()

    lu = FakeLU(None, op, cfg, sstore)

    hm = mcpu.HooksMaster(self._call_nodes_false, cfg, sstore, lu)

    self.failUnlessRaises(errors.HooksFailure,

                          hm.RunPhase, constants.HOOKS_PHASE_PRE)

    hm.RunPhase(constants.HOOKS_PHASE_POST)

  def testScriptFalse(self):

    """Test individual rpc failure"""

    cfg = FakeConfig()

    op = opcodes.OpCode()

    sstore = FakeSStore()

    lu = FakeLU(None, op, cfg, sstore)

    hm = mcpu.HooksMaster(self._call_script_fail, cfg, sstore, lu)

    self.failUnlessRaises(errors.HooksAbort,

                          hm.RunPhase, constants.HOOKS_PHASE_PRE)

    hm.RunPhase(constants.HOOKS_PHASE_POST)

  def testScriptSucceed(self):

    """Test individual rpc failure"""

    cfg = FakeConfig()

    op = opcodes.OpCode()

    sstore = FakeSStore()

    lu = FakeLU(None, op, cfg, sstore)

    hm = mcpu.HooksMaster(self._call_script_succeed, cfg, sstore, lu)

    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):

      hm.RunPhase(phase)

if __name__ == '__main__':

  unittest.main()

#!/usr/bin/python

#

# Copyright (C) 2006, 2007 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for unittesting the utils module"""

import unittest

import os

import time

import tempfile

import os.path

import md5

import ganeti

from ganeti.utils import IsProcessAlive, Lock, Unlock, RunCmd, \

     RemoveFile, CheckDict, MatchNameComponent, FormatUnit, \

     ParseUnit, AddAuthorizedKey, RemoveAuthorizedKey, \

     ShellQuote, ShellQuoteArgs, _ParseIpOutput

from ganeti.errors import LockError, UnitParseError

class TestIsProcessAlive(unittest.TestCase):

  """Testing case for IsProcessAlive"""

  def setUp(self):

    # create a zombie and a (hopefully) non-existing process id

    self.pid_zombie = os.fork()

    if self.pid_zombie == 0:

      os._exit(0)

    elif self.pid_zombie < 0:

      raise SystemError("can't fork")

    self.pid_non_existing = os.fork()

    if self.pid_non_existing == 0:

      os._exit(0)

    elif self.pid_non_existing > 0:

      os.waitpid(self.pid_non_existing, 0)

    else:

      raise SystemError("can't fork")

  def testExists(self):

    mypid = os.getpid()

    self.assert_(IsProcessAlive(mypid),

                 "can't find myself running")

  def testZombie(self):

    self.assert_(not IsProcessAlive(self.pid_zombie),

                 "zombie not detected as zombie")

  def testNotExisting(self):

    self.assert_(not IsProcessAlive(self.pid_non_existing),

                 "noexisting process detected")

class TestLocking(unittest.TestCase):

  """Testing case for the Lock/Unlock functions"""

  def clean_lock(self, name):

    try:

      ganeti.utils.Unlock("unittest")

    except LockError:

      pass

  def testLock(self):

    self.clean_lock("unittest")

    self.assertEqual(None, Lock("unittest"))

  def testUnlock(self):

    self.clean_lock("unittest")

    ganeti.utils.Lock("unittest")

    self.assertEqual(None, Unlock("unittest"))

  def testDoubleLock(self):

    self.clean_lock("unittest")

    ganeti.utils.Lock("unittest")

    self.assertRaises(LockError, Lock, "unittest")

class TestRunCmd(unittest.TestCase):

  """Testing case for the RunCmd function"""

  def setUp(self):

    self.magic = time.ctime() + " ganeti test"

  def testOk(self):

    """Test successful exit code"""

    result = RunCmd("/bin/sh -c 'exit 0'")

    self.assertEqual(result.exit_code, 0)

  def testFail(self):

    """Test fail exit code"""

    result = RunCmd("/bin/sh -c 'exit 1'")

    self.assertEqual(result.exit_code, 1)

  def testStdout(self):

    """Test standard output"""

    cmd = 'echo -n "%s"' % self.magic

    result = RunCmd("/bin/sh -c '%s'" % cmd)

    self.assertEqual(result.stdout, self.magic)

  def testStderr(self):

    """Test standard error"""

    cmd = 'echo -n "%s"' % self.magic

    result = RunCmd("/bin/sh -c '%s' 1>&2" % cmd)

    self.assertEqual(result.stderr, self.magic)

  def testCombined(self):

    """Test combined output"""

    cmd = 'echo -n "A%s"; echo -n "B%s" 1>&2' % (self.magic, self.magic)

    result = RunCmd("/bin/sh -c '%s'" % cmd)

    self.assertEqual(result.output, "A" + self.magic + "B" + self.magic)

  def testSignal(self):

    """Test standard error"""

    result = RunCmd("/bin/sh -c 'kill -15 $$'")

    self.assertEqual(result.signal, 15)

  def testListRun(self):

    """Test list runs"""

    result = RunCmd(["true"])

    self.assertEqual(result.signal, None)

    self.assertEqual(result.exit_code, 0)

    result = RunCmd(["/bin/sh", "-c", "exit 1"])

    self.assertEqual(result.signal, None)

    self.assertEqual(result.exit_code, 1)

    result = RunCmd(["echo", "-n", self.magic])

    self.assertEqual(result.signal, None)

    self.assertEqual(result.exit_code, 0)

    self.assertEqual(result.stdout, self.magic)

  def testLang(self):

    """Test locale environment"""

    old_env = os.environ.copy()

    try:

      os.environ["LANG"] = "en_US.UTF-8"

      os.environ["LC_ALL"] = "en_US.UTF-8"

      result = RunCmd(["locale"])

      for line in result.output.splitlines():

        key, value = line.split("=", 1)

        # Ignore these variables, they're overridden by LC_ALL

        if key == "LANG" or key == "LANGUAGE":

          continue

        self.failIf(value and value != "C" and value != '"C"',

            "Variable %s is set to the invalid value '%s'" % (key, value))

    finally:

      os.environ = old_env

class TestRemoveFile(unittest.TestCase):

  """Test case for the RemoveFile function"""

  def setUp(self):

    """Create a temp dir and file for each case"""

    self.tmpdir = tempfile.mkdtemp('', 'ganeti-unittest-')

    fd, self.tmpfile = tempfile.mkstemp('', '', self.tmpdir)

    os.close(fd)

  def tearDown(self):

    if os.path.exists(self.tmpfile):

      os.unlink(self.tmpfile)

    os.rmdir(self.tmpdir)

  def testIgnoreDirs(self):

    """Test that RemoveFile() ignores directories"""

    self.assertEqual(None, RemoveFile(self.tmpdir))

  def testIgnoreNotExisting(self):

    """Test that RemoveFile() ignores non-existing files"""

    RemoveFile(self.tmpfile)

    RemoveFile(self.tmpfile)

  def testRemoveFile(self):

    """Test that RemoveFile does remove a file"""

    RemoveFile(self.tmpfile)

    if os.path.exists(self.tmpfile):

      self.fail("File '%s' not removed" % self.tmpfile)

  def testRemoveSymlink(self):

    """Test that RemoveFile does remove symlinks"""

    symlink = self.tmpdir + "/symlink"

    os.symlink("no-such-file", symlink)

    RemoveFile(symlink)

    if os.path.exists(symlink):

      self.fail("File '%s' not removed" % symlink)

    os.symlink(self.tmpfile, symlink)

    RemoveFile(symlink)

    if os.path.exists(symlink):

      self.fail("File '%s' not removed" % symlink)

class TestCheckdict(unittest.TestCase):

  """Test case for the CheckDict function"""

  def testAdd(self):

    """Test that CheckDict adds a missing key with the correct value"""

    tgt = {'a':1}

    tmpl = {'b': 2}

    CheckDict(tgt, tmpl)

    if 'b' not in tgt or tgt['b'] != 2:

      self.fail("Failed to update dict")

  def testNoUpdate(self):

    """Test that CheckDict does not overwrite an existing key"""

    tgt = {'a':1, 'b': 3}

    tmpl = {'b': 2}

    CheckDict(tgt, tmpl)

    self.failUnlessEqual(tgt['b'], 3)

class TestMatchNameComponent(unittest.TestCase):

  """Test case for the MatchNameComponent function"""

  def testEmptyList(self):

    """Test that there is no match against an empty list"""

    self.failUnlessEqual(MatchNameComponent("", []), None)

    self.failUnlessEqual(MatchNameComponent("test", []), None)

  def testSingleMatch(self):

    """Test that a single match is performed correctly"""

    mlist = ["test1.example.com", "test2.example.com", "test3.example.com"]

    for key in "test2", "test2.example", "test2.example.com":

      self.failUnlessEqual(MatchNameComponent(key, mlist), mlist[1])

  def testMultipleMatches(self):

    """Test that a multiple match is returned as None"""

    mlist = ["test1.example.com", "test1.example.org", "test1.example.net"]

    for key in "test1", "test1.example":

      self.failUnlessEqual(MatchNameComponent(key, mlist), None)

class TestFormatUnit(unittest.TestCase):

  """Test case for the FormatUnit function"""

  def testMiB(self):

    self.assertEqual(FormatUnit(1), '1M')

    self.assertEqual(FormatUnit(100), '100M')

    self.assertEqual(FormatUnit(1023), '1023M')

  def testGiB(self):

    self.assertEqual(FormatUnit(1024), '1.0G')

    self.assertEqual(FormatUnit(1536), '1.5G')

    self.assertEqual(FormatUnit(17133), '16.7G')

    self.assertEqual(FormatUnit(1024 * 1024 - 1), '1024.0G')

  def testTiB(self):

    self.assertEqual(FormatUnit(1024 * 1024), '1.0T')

    self.assertEqual(FormatUnit(5120 * 1024), '5.0T')

    self.assertEqual(FormatUnit(29829 * 1024), '29.1T')

class TestParseUnit(unittest.TestCase):

  """Test case for the ParseUnit function"""

  SCALES = (('', 1),

            ('M', 1), ('G', 1024), ('T', 1024 * 1024),

            ('MB', 1), ('GB', 1024), ('TB', 1024 * 1024),

            ('MiB', 1), ('GiB', 1024), ('TiB', 1024 * 1024))

  def testRounding(self):

    self.assertEqual(ParseUnit('0'), 0)

    self.assertEqual(ParseUnit('1'), 4)

    self.assertEqual(ParseUnit('2'), 4)

    self.assertEqual(ParseUnit('3'), 4)

    self.assertEqual(ParseUnit('124'), 124)

    self.assertEqual(ParseUnit('125'), 128)

    self.assertEqual(ParseUnit('126'), 128)

    self.assertEqual(ParseUnit('127'), 128)

    self.assertEqual(ParseUnit('128'), 128)

    self.assertEqual(ParseUnit('129'), 132)

    self.assertEqual(ParseUnit('130'), 132)

  def testFloating(self):

    self.assertEqual(ParseUnit('0'), 0)

    self.assertEqual(ParseUnit('0.5'), 4)

    self.assertEqual(ParseUnit('1.75'), 4)

    self.assertEqual(ParseUnit('1.99'), 4)

    self.assertEqual(ParseUnit('2.00'), 4)

    self.assertEqual(ParseUnit('2.01'), 4)

    self.assertEqual(ParseUnit('3.99'), 4)

    self.assertEqual(ParseUnit('4.00'), 4)

    self.assertEqual(ParseUnit('4.01'), 8)

    self.assertEqual(ParseUnit('1.5G'), 1536)

    self.assertEqual(ParseUnit('1.8G'), 1844)

    self.assertEqual(ParseUnit('8.28T'), 8682212)

  def testSuffixes(self):

    for sep in ('', ' ', '   ', "\t", "\t "):

      for suffix, scale in TestParseUnit.SCALES:

        for func in (lambda x: x, str.lower, str.upper):

          self.assertEqual(ParseUnit('1024' + sep + func(suffix)), 1024 * scale)

  def testInvalidInput(self):

    for sep in ('-', '_', ',', 'a'):

      for suffix, _ in TestParseUnit.SCALES:

        self.assertRaises(UnitParseError, ParseUnit, '1' + sep + suffix)

    for suffix, _ in TestParseUnit.SCALES:

      self.assertRaises(UnitParseError, ParseUnit, '1,3' + suffix)

class TestSshKeys(unittest.TestCase):

  """Test case for the AddAuthorizedKey function"""

  KEY_A = 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a'

  KEY_B = ('command="/usr/bin/fooserver -t --verbose",from="1.2.3.4" '

           'ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b')

  # NOTE: The MD5 sums below were calculated after manually

  #       checking the output files.

  def writeTestFile(self):

    (fd, tmpname) = tempfile.mkstemp(prefix = 'ganeti-test')

    f = os.fdopen(fd, 'w')

    try:

      f.write(TestSshKeys.KEY_A)

      f.write("\n")

      f.write(TestSshKeys.KEY_B)

      f.write("\n")

    finally:

      f.close()

    return tmpname

  def testAddingNewKey(self):

    tmpname = self.writeTestFile()

    try:

      AddAuthorizedKey(tmpname, 'ssh-dss AAAAB3NzaC1kc3MAAACB root@test')

      f = open(tmpname, 'r')

      try:

        self.assertEqual(md5.new(f.read(8192)).hexdigest(),

                         'ccc71523108ca6e9d0343797dc3e9f16')

      finally:

        f.close()

    finally:

      os.unlink(tmpname)

  def testAddingAlmostButNotCompletlyTheSameKey(self):

    tmpname = self.writeTestFile()

    try:

      AddAuthorizedKey(tmpname,

          'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@test')

      f = open(tmpname, 'r')

      try:

        self.assertEqual(md5.new(f.read(8192)).hexdigest(),

                         'f2c939d57addb5b3a6846884be896b46')

      finally:

        f.close()

    finally:

      os.unlink(tmpname)

  def testAddingExistingKeyWithSomeMoreSpaces(self):

    tmpname = self.writeTestFile()

    try:

      AddAuthorizedKey(tmpname,

          'ssh-dss  AAAAB3NzaC1w5256closdj32mZaQU   root@key-a')

      f = open(tmpname, 'r')

      try:

        self.assertEqual(md5.new(f.read(8192)).hexdigest(),

                         '4e612764808bd46337eb0f575415fc30')

      finally:

        f.close()

    finally:

      os.unlink(tmpname)

  def testRemovingExistingKeyWithSomeMoreSpaces(self):

    tmpname = self.writeTestFile()

    try:

      RemoveAuthorizedKey(tmpname,

          'ssh-dss  AAAAB3NzaC1w5256closdj32mZaQU   root@key-a')

      f = open(tmpname, 'r')

      try:

        self.assertEqual(md5.new(f.read(8192)).hexdigest(),

                         '77516d987fca07f70e30b830b3e4f2ed')

      finally:

        f.close()

    finally:

      os.unlink(tmpname)

  def testRemovingNonExistingKey(self):

    tmpname = self.writeTestFile()

    try:

      RemoveAuthorizedKey(tmpname,

          'ssh-dss  AAAAB3Nsdfj230xxjxJjsjwjsjdjU   root@test')

      f = open(tmpname, 'r')

      try:

        self.assertEqual(md5.new(f.read(8192)).hexdigest(),

                         '4e612764808bd46337eb0f575415fc30')

      finally:

        f.close()

    finally:

      os.unlink(tmpname)

class TestShellQuoting(unittest.TestCase):

  """Test case for shell quoting functions"""

  def testShellQuote(self):

    self.assertEqual(ShellQuote('abc'), "abc")

    self.assertEqual(ShellQuote('ab"c'), "'ab\"c'")

    self.assertEqual(ShellQuote("a'bc"), "'a'\\''bc'")

    self.assertEqual(ShellQuote("a b c"), "'a b c'")

    self.assertEqual(ShellQuote("a b\\ c"), "'a b\\ c'")

  def testShellQuoteArgs(self):

    self.assertEqual(ShellQuoteArgs(['a', 'b', 'c']), "a b c")

    self.assertEqual(ShellQuoteArgs(['a', 'b"', 'c']), "a 'b\"' c")

    self.assertEqual(ShellQuoteArgs(['a', 'b\'', 'c']), "a 'b'\\\''' c")

class TestIpAdressList(unittest.TestCase):

  """Test case for local IP addresses"""

  def _test(self, output, required):

    ips = _ParseIpOutput(output)

    # Sort the output, so our check below works in all cases

    ips.sort()

    required.sort()

    self.assertEqual(required, ips)

  def testSingleIpAddress(self):

    output = \

      ("3: lo    inet 127.0.0.1/8 brd 127.255.255.255 scope host lo\n"

       "5: eth0    inet 10.0.0.1/24 brd 172.30.15.127 scope global eth0\n")

    self._test(output, ['127.0.0.1', '10.0.0.1'])

  def testMultipleIpAddresses(self):

    output = \

      ("3: lo    inet 127.0.0.1/8 brd 127.255.255.255 scope host lo\n"

       "5: eth0    inet 10.0.0.1/24 brd 172.30.15.127 scope global eth0\n"

       "5: eth0    inet 1.2.3.4/8 brd 1.255.255.255 scope global eth0:test\n")

    self._test(output, ['127.0.0.1', '10.0.0.1', '1.2.3.4'])

if __name__ == '__main__':

  unittest.main()

#!/usr/bin/python

#

# Copyright (C) 2006, 2007 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Module implementing a fake ConfigWriter"""

import socket

class FakeConfig:

    """Fake configuration object"""

    def IsCluster(self):

        return True

    def GetNodeList(self):

        return ["a", "b", "c"]

    def GetMaster(self):

        return socket.gethostname()

class FakeSStore:

    """Fake simplestore object"""

    def GetClusterName(self):

        return "test.cluster"

    def GetMasterNode(self):

        return socket.gethostname()