Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.config_unittest.py @ 36b8c2c1

History | View | Annotate | Download (4.8 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 errors
33
from ganeti import constants
34
from ganeti import config
35
from ganeti import objects
36
from ganeti import utils
37

    
38

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

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

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

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

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

    
77
    master_node_config = objects.Node(name=me.name,
78
                                      primary_ip=me.ip,
79
                                      secondary_ip=ip,
80
                                      serial_no=1)
81

    
82
    cfg.InitConfig(constants.CONFIG_VERSION,
83
                   cluster_config, master_node_config)
84

    
85
  def _create_instance(self):
86
    """Create and return an instance object"""
87
    inst = objects.Instance(name="test.example.com", disks=[],
88
                            disk_template=constants.DT_DISKLESS)
89
    return inst
90

    
91
  def testEmpty(self):
92
    """Test instantiate config object"""
93
    self._get_object()
94

    
95
  def testInit(self):
96
    """Test initialize the config file"""
97
    cfg = self._get_object()
98
    self._init_cluster(cfg)
99
    self.failUnlessEqual(1, len(cfg.GetNodeList()))
100
    self.failUnlessEqual(0, len(cfg.GetInstanceList()))
101

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

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

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

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

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

    
145
    self._init_cluster(cfg)
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

    
156
if __name__ == '__main__':
157
  unittest.main()