code.grnet.gr Git - ganeti-local/blob - test/ganeti.config_unittest.py

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2010 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Script for unittesting the config module"""
  23
  24
  25 import unittest
  26 import os
  27 import time
  28 import tempfile
  29 import os.path
  30 import socket
  31
  32 from ganeti import bootstrap
  33 from ganeti import config
  34 from ganeti import constants
  35 from ganeti import errors
  36 from ganeti import objects
  37 from ganeti import utils
  38 from ganeti import netutils
  39
  40 import testutils
  41 import mocks
  42
  43
  44 def _StubGetEntResolver():
  45   return mocks.FakeGetentResolver()
  46
  47
  48 class TestConfigRunner(unittest.TestCase):
  49   """Testing case for HooksRunner"""
  50   def setUp(self):
  51     fd, self.cfg_file = tempfile.mkstemp()
  52     os.close(fd)
  53     self._init_cluster(self.cfg_file)
  54
  55   def tearDown(self):
  56     try:
  57       os.unlink(self.cfg_file)
  58     except OSError:
  59       pass
  60
  61   def _get_object(self):
  62     """Returns a instance of ConfigWriter"""
  63     cfg = config.ConfigWriter(cfg_file=self.cfg_file, offline=True,
  64                               _getents=_StubGetEntResolver)
  65     return cfg
  66
  67   def _init_cluster(self, cfg):
  68     """Initializes the cfg object"""
  69     me = netutils.Hostname()
  70     ip = constants.IP4_ADDRESS_LOCALHOST
  71
  72     cluster_config = objects.Cluster(
  73       serial_no=1,
  74       rsahostkeypub="",
  75       highest_used_port=(constants.FIRST_DRBD_PORT - 1),
  76       mac_prefix="aa:00:00",
  77       volume_group_name="xenvg",
  78       drbd_usermode_helper="/bin/true",
  79       nicparams={constants.PP_DEFAULT: constants.NICC_DEFAULTS},
  80       tcpudp_port_pool=set(),
  81       enabled_hypervisors=[constants.HT_FAKE],
  82       master_node=me.name,
  83       master_ip="127.0.0.1",
  84       master_netdev=constants.DEFAULT_BRIDGE,
  85       cluster_name="cluster.local",
  86       file_storage_dir="/tmp",
  87       uid_pool=[],
  88       )
  89
  90     master_node_config = objects.Node(name=me.name,
  91                                       primary_ip=me.ip,
  92                                       secondary_ip=ip,
  93                                       serial_no=1,
  94                                       master_candidate=True)
  95
  96     bootstrap.InitConfig(constants.CONFIG_VERSION,
  97                          cluster_config, master_node_config, self.cfg_file)
  98
  99   def _create_instance(self):
 100     """Create and return an instance object"""
 101     inst = objects.Instance(name="test.example.com", disks=[], nics=[],
 102                             disk_template=constants.DT_DISKLESS,
 103                             primary_node=self._get_object().GetMasterNode())
 104     return inst
 105
 106   def testEmpty(self):
 107     """Test instantiate config object"""
 108     self._get_object()
 109
 110   def testInit(self):
 111     """Test initialize the config file"""
 112     cfg = self._get_object()
 113     self.failUnlessEqual(1, len(cfg.GetNodeList()))
 114     self.failUnlessEqual(0, len(cfg.GetInstanceList()))
 115
 116   def testUpdateCluster(self):
 117     """Test updates on the cluster object"""
 118     cfg = self._get_object()
 119     # construct a fake cluster object
 120     fake_cl = objects.Cluster()
 121     # fail if we didn't read the config
 122     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_cl, None)
 123
 124     cl = cfg.GetClusterInfo()
 125     # first pass, must not fail
 126     cfg.Update(cl, None)
 127     # second pass, also must not fail (after the config has been written)
 128     cfg.Update(cl, None)
 129     # but the fake_cl update should still fail
 130     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_cl, None)
 131
 132   def testUpdateNode(self):
 133     """Test updates on one node object"""
 134     cfg = self._get_object()
 135     # construct a fake node
 136     fake_node = objects.Node()
 137     # fail if we didn't read the config
 138     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_node,
 139                           None)
 140
 141     node = cfg.GetNodeInfo(cfg.GetNodeList()[0])
 142     # first pass, must not fail
 143     cfg.Update(node, None)
 144     # second pass, also must not fail (after the config has been written)
 145     cfg.Update(node, None)
 146     # but the fake_node update should still fail
 147     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_node,
 148                           None)
 149
 150   def testUpdateInstance(self):
 151     """Test updates on one instance object"""
 152     cfg = self._get_object()
 153     # construct a fake instance
 154     inst = self._create_instance()
 155     fake_instance = objects.Instance()
 156     # fail if we didn't read the config
 157     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_instance,
 158                           None)
 159
 160     cfg.AddInstance(inst, "my-job")
 161     instance = cfg.GetInstanceInfo(cfg.GetInstanceList()[0])
 162     # first pass, must not fail
 163     cfg.Update(instance, None)
 164     # second pass, also must not fail (after the config has been written)
 165     cfg.Update(instance, None)
 166     # but the fake_instance update should still fail
 167     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_instance,
 168                           None)
 169
 170   def testNICParameterSyntaxCheck(self):
 171     """Test the NIC's CheckParameterSyntax function"""
 172     mode = constants.NIC_MODE
 173     link = constants.NIC_LINK
 174     m_bridged = constants.NIC_MODE_BRIDGED
 175     m_routed = constants.NIC_MODE_ROUTED
 176     CheckSyntax = objects.NIC.CheckParameterSyntax
 177
 178     CheckSyntax(constants.NICC_DEFAULTS)
 179     CheckSyntax({mode: m_bridged, link: 'br1'})
 180     CheckSyntax({mode: m_routed, link: 'default'})
 181     self.assertRaises(errors.ConfigurationError,
 182                       CheckSyntax, {mode: '000invalid', link: 'any'})
 183     self.assertRaises(errors.ConfigurationError,
 184                       CheckSyntax, {mode: m_bridged, link: None})
 185     self.assertRaises(errors.ConfigurationError,
 186                       CheckSyntax, {mode: m_bridged, link: ''})
 187
 188
 189 if __name__ == '__main__':
 190   testutils.GanetiTestProgram()