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