Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.config_unittest.py @ 255e19d4

History | View | Annotate | Download (5.7 kB)

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

    
40
class TestConfigRunner(unittest.TestCase):
41
  """Testing case for HooksRunner"""
42
  def setUp(self):
43
    fd, self.cfg_file = tempfile.mkstemp()
44
    os.close(fd)
45
    self._init_cluster(self.cfg_file)
46

    
47
  def tearDown(self):
48
    try:
49
      os.unlink(self.cfg_file)
50
    except OSError:
51
      pass
52

    
53
  def _get_object(self):
54
    """Returns a instance of ConfigWriter"""
55
    cfg = config.ConfigWriter(cfg_file=self.cfg_file, offline=True)
56
    return cfg
57

    
58
  def _init_cluster(self, cfg):
59
    """Initializes the cfg object"""
60
    me = utils.HostInfo()
61
    ip = constants.LOCALHOST_IP_ADDRESS
62

    
63
    cluster_config = objects.Cluster(
64
      serial_no=1,
65
      rsahostkeypub="",
66
      highest_used_port=(constants.FIRST_DRBD_PORT - 1),
67
      mac_prefix="aa:00:00",
68
      volume_group_name="xenvg",
69
      default_bridge=constants.DEFAULT_BRIDGE,
70
      tcpudp_port_pool=set(),
71
      default_hypervisor=constants.HT_FAKE,
72
      master_node=me.name,
73
      master_ip="127.0.0.1",
74
      master_netdev=constants.DEFAULT_BRIDGE,
75
      cluster_name="cluster.local",
76
      file_storage_dir="/tmp",
77
      )
78

    
79
    master_node_config = objects.Node(name=me.name,
80
                                      primary_ip=me.ip,
81
                                      secondary_ip=ip,
82
                                      serial_no=1,
83
                                      master_candidate=True)
84

    
85
    bootstrap.InitConfig(constants.CONFIG_VERSION,
86
                         cluster_config, master_node_config, self.cfg_file)
87

    
88
  def _create_instance(self):
89
    """Create and return an instance object"""
90
    inst = objects.Instance(name="test.example.com", disks=[], nics=[],
91
                            disk_template=constants.DT_DISKLESS,
92
                            primary_node=self._get_object().GetMasterNode())
93
    return inst
94

    
95
  def testEmpty(self):
96
    """Test instantiate config object"""
97
    self._get_object()
98

    
99
  def testInit(self):
100
    """Test initialize the config file"""
101
    cfg = self._get_object()
102
    self.failUnlessEqual(1, len(cfg.GetNodeList()))
103
    self.failUnlessEqual(0, len(cfg.GetInstanceList()))
104

    
105
  def testUpdateCluster(self):
106
    """Test updates on the cluster object"""
107
    cfg = self._get_object()
108
    # construct a fake cluster object
109
    fake_cl = objects.Cluster()
110
    # fail if we didn't read the config
111
    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_cl)
112

    
113
    cl = cfg.GetClusterInfo()
114
    # first pass, must not fail
115
    cfg.Update(cl)
116
    # second pass, also must not fail (after the config has been written)
117
    cfg.Update(cl)
118
    # but the fake_cl update should still fail
119
    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_cl)
120

    
121
  def testUpdateNode(self):
122
    """Test updates on one node object"""
123
    cfg = self._get_object()
124
    # construct a fake node
125
    fake_node = objects.Node()
126
    # fail if we didn't read the config
127
    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_node)
128

    
129
    node = cfg.GetNodeInfo(cfg.GetNodeList()[0])
130
    # first pass, must not fail
131
    cfg.Update(node)
132
    # second pass, also must not fail (after the config has been written)
133
    cfg.Update(node)
134
    # but the fake_node update should still fail
135
    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_node)
136

    
137
  def testUpdateInstance(self):
138
    """Test updates on one instance object"""
139
    cfg = self._get_object()
140
    # construct a fake instance
141
    inst = self._create_instance()
142
    fake_instance = objects.Instance()
143
    # fail if we didn't read the config
144
    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_instance)
145

    
146
    cfg.AddInstance(inst)
147
    instance = cfg.GetInstanceInfo(cfg.GetInstanceList()[0])
148
    # first pass, must not fail
149
    cfg.Update(instance)
150
    # second pass, also must not fail (after the config has been written)
151
    cfg.Update(instance)
152
    # but the fake_instance update should still fail
153
    self.failUnlessRaises(errors.ConfigurationError, cfg.Update, fake_instance)
154

    
155
  def testNICParameterSyntaxCheck(self):
156
    """Test the NIC's CheckParameterSyntax function"""
157
    mode = constants.NIC_MODE
158
    link = constants.NIC_LINK
159
    m_bridged = constants.NIC_MODE_BRIDGED
160
    m_routed = constants.NIC_MODE_ROUTED
161
    CheckSyntax = objects.NIC.CheckParameterSyntax
162

    
163
    CheckSyntax(constants.NICC_DEFAULTS)
164
    CheckSyntax({mode: m_bridged, link: 'br1'})
165
    CheckSyntax({mode: m_routed, link: 'default'})
166
    self.assertRaises(errors.ConfigurationError,
167
                      CheckSyntax, {mode: '000invalid', link: 'any'})
168
    self.assertRaises(errors.ConfigurationError,
169
                      CheckSyntax, {mode: m_bridged, link: None})
170
    self.assertRaises(errors.ConfigurationError,
171
                      CheckSyntax, {mode: m_bridged, link: ''})
172

    
173

    
174
if __name__ == '__main__':
175
  unittest.main()