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

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2006, 2007 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
  39 import testutils
  40
  41
  42 class TestConfigRunner(unittest.TestCase):
  43   """Testing case for HooksRunner"""
  44   def setUp(self):
  45     fd, self.cfg_file = tempfile.mkstemp()
  46     os.close(fd)
  47     self._init_cluster(self.cfg_file)
  48
  49   def tearDown(self):
  50     try:
  51       os.unlink(self.cfg_file)
  52     except OSError:
  53       pass
  54
  55   def _get_object(self):
  56     """Returns a instance of ConfigWriter"""
  57     cfg = config.ConfigWriter(cfg_file=self.cfg_file, offline=True)
  58     return cfg
  59
  60   def _init_cluster(self, cfg):
  61     """Initializes the cfg object"""
  62     me = utils.HostInfo()
  63     ip = constants.LOCALHOST_IP_ADDRESS
  64
  65     cluster_config = objects.Cluster(
  66       serial_no=1,
  67       rsahostkeypub="",
  68       highest_used_port=(constants.FIRST_DRBD_PORT - 1),
  69       mac_prefix="aa:00:00",
  70       volume_group_name="xenvg",
  71       nicparams={constants.PP_DEFAULT: constants.NICC_DEFAULTS},
  72       tcpudp_port_pool=set(),
  73       enabled_hypervisors=[constants.HT_FAKE],
  74       master_node=me.name,
  75       master_ip="127.0.0.1",
  76       master_netdev=constants.DEFAULT_BRIDGE,
  77       cluster_name="cluster.local",
  78       file_storage_dir="/tmp",
  79       uid_pool=[],
  80       )
  81
  82     master_node_config = objects.Node(name=me.name,
  83                                       primary_ip=me.ip,
  84                                       secondary_ip=ip,
  85                                       serial_no=1,
  86                                       master_candidate=True)
  87
  88     bootstrap.InitConfig(constants.CONFIG_VERSION,
  89                          cluster_config, master_node_config, self.cfg_file)
  90
  91   def _create_instance(self):
  92     """Create and return an instance object"""
  93     inst = objects.Instance(name="test.example.com", disks=[], nics=[],
  94                             disk_template=constants.DT_DISKLESS,
  95                             primary_node=self._get_object().GetMasterNode())
  96     return inst
  97
  98   def testEmpty(self):
  99     """Test instantiate config object"""
 100     self._get_object()
 101
 102   def testInit(self):
 103     """Test initialize the config file"""
 104     cfg = self._get_object()
 105     self.failUnlessEqual(1, len(cfg.GetNodeList()))
 106     self.failUnlessEqual(0, len(cfg.GetInstanceList()))
 107
 108   def testUpdateCluster(self):
 109     """Test updates on the cluster object"""
 110     cfg = self._get_object()
 111     # construct a fake cluster object
 112     fake_cl = objects.Cluster()
 113     # fail if we didn't read the config
 114     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_cl, None)
 115
 116     cl = cfg.GetClusterInfo()
 117     # first pass, must not fail
 118     cfg.Update(cl, None)
 119     # second pass, also must not fail (after the config has been written)
 120     cfg.Update(cl, None)
 121     # but the fake_cl update should still fail
 122     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_cl, None)
 123
 124   def testUpdateNode(self):
 125     """Test updates on one node object"""
 126     cfg = self._get_object()
 127     # construct a fake node
 128     fake_node = objects.Node()
 129     # fail if we didn't read the config
 130     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_node,
 131                           None)
 132
 133     node = cfg.GetNodeInfo(cfg.GetNodeList()[0])
 134     # first pass, must not fail
 135     cfg.Update(node, None)
 136     # second pass, also must not fail (after the config has been written)
 137     cfg.Update(node, None)
 138     # but the fake_node update should still fail
 139     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_node,
 140                           None)
 141
 142   def testUpdateInstance(self):
 143     """Test updates on one instance object"""
 144     cfg = self._get_object()
 145     # construct a fake instance
 146     inst = self._create_instance()
 147     fake_instance = objects.Instance()
 148     # fail if we didn't read the config
 149     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_instance,
 150                           None)
 151
 152     cfg.AddInstance(inst, "my-job")
 153     instance = cfg.GetInstanceInfo(cfg.GetInstanceList()[0])
 154     # first pass, must not fail
 155     cfg.Update(instance, None)
 156     # second pass, also must not fail (after the config has been written)
 157     cfg.Update(instance, None)
 158     # but the fake_instance update should still fail
 159     self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_instance,
 160                           None)
 161
 162   def testNICParameterSyntaxCheck(self):
 163     """Test the NIC's CheckParameterSyntax function"""
 164     mode = constants.NIC_MODE
 165     link = constants.NIC_LINK
 166     m_bridged = constants.NIC_MODE_BRIDGED
 167     m_routed = constants.NIC_MODE_ROUTED
 168     CheckSyntax = objects.NIC.CheckParameterSyntax
 169
 170     CheckSyntax(constants.NICC_DEFAULTS)
 171     CheckSyntax({mode: m_bridged, link: 'br1'})
 172     CheckSyntax({mode: m_routed, link: 'default'})
 173     self.assertRaises(errors.ConfigurationError,
 174                       CheckSyntax, {mode: '000invalid', link: 'any'})
 175     self.assertRaises(errors.ConfigurationError,
 176                       CheckSyntax, {mode: m_bridged, link: None})
 177     self.assertRaises(errors.ConfigurationError,
 178                       CheckSyntax, {mode: m_bridged, link: ''})
 179
 180
 181 if __name__ == '__main__':
 182   testutils.GanetiTestProgram()