Statistics
| Branch: | Tag: | Revision:

root / test / py / ganeti.constants_unittest.py @ d48c944b

History | View | Annotate | Download (6.7 kB)

1
#!/usr/bin/python
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008 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 constants module"""
23

    
24

    
25
import unittest
26
import re
27
import itertools
28

    
29
from ganeti import constants
30
from ganeti import locking
31
from ganeti import utils
32

    
33
from ganeti.utils import version
34

    
35
import testutils
36

    
37

    
38
class TestConstants(unittest.TestCase):
39
  """Constants tests"""
40

    
41
  def testConfigVersion(self):
42
    self.failUnless(constants.CONFIG_MAJOR >= 0 and
43
                    constants.CONFIG_MAJOR <= 99)
44
    self.failUnless(constants.CONFIG_MINOR >= 0 and
45
                    constants.CONFIG_MINOR <= 99)
46
    self.failUnless(constants.CONFIG_REVISION >= 0 and
47
                    constants.CONFIG_REVISION <= 9999)
48
    self.failUnless(constants.CONFIG_VERSION >= 0 and
49
                    constants.CONFIG_VERSION <= 99999999)
50

    
51
    self.failUnless(version.BuildVersion(0, 0, 0) == 0)
52
    self.failUnless(version.BuildVersion(10, 10, 1010) == 10101010)
53
    self.failUnless(version.BuildVersion(12, 34, 5678) == 12345678)
54
    self.failUnless(version.BuildVersion(99, 99, 9999) == 99999999)
55

    
56
    self.failUnless(version.SplitVersion(00000000) == (0, 0, 0))
57
    self.failUnless(version.SplitVersion(10101010) == (10, 10, 1010))
58
    self.failUnless(version.SplitVersion(12345678) == (12, 34, 5678))
59
    self.failUnless(version.SplitVersion(99999999) == (99, 99, 9999))
60
    self.failUnless(version.SplitVersion(constants.CONFIG_VERSION) ==
61
                    (constants.CONFIG_MAJOR, constants.CONFIG_MINOR,
62
                     constants.CONFIG_REVISION))
63

    
64
  def testDiskStatus(self):
65
    self.failUnless(constants.LDS_OKAY < constants.LDS_UNKNOWN)
66
    self.failUnless(constants.LDS_UNKNOWN < constants.LDS_FAULTY)
67

    
68
  def testClockSkew(self):
69
    self.failUnless(constants.NODE_MAX_CLOCK_SKEW <
70
                    (0.8 * constants.CONFD_MAX_CLOCK_SKEW))
71

    
72
  def testSslCertExpiration(self):
73
    self.failUnless(constants.SSL_CERT_EXPIRATION_ERROR <
74
                    constants.SSL_CERT_EXPIRATION_WARN)
75

    
76
  def testOpCodePriority(self):
77
    self.failUnless(constants.OP_PRIO_LOWEST > constants.OP_PRIO_LOW)
78
    self.failUnless(constants.OP_PRIO_LOW > constants.OP_PRIO_NORMAL)
79
    self.failUnlessEqual(constants.OP_PRIO_NORMAL, locking._DEFAULT_PRIORITY)
80
    self.failUnlessEqual(constants.OP_PRIO_DEFAULT, locking._DEFAULT_PRIORITY)
81
    self.failUnless(constants.OP_PRIO_NORMAL > constants.OP_PRIO_HIGH)
82
    self.failUnless(constants.OP_PRIO_HIGH > constants.OP_PRIO_HIGHEST)
83

    
84
  def testDiskDefaults(self):
85
    self.failUnless(
86
        set(constants.DISK_LD_DEFAULTS.keys()) ==
87
        set(constants.DISK_TEMPLATES) - set([constants.DT_DISKLESS]))
88
    self.failUnless(set(constants.DISK_DT_DEFAULTS.keys()) ==
89
                    constants.DISK_TEMPLATES)
90

    
91
  def testJobStatus(self):
92
    self.assertFalse(constants.JOBS_PENDING & constants.JOBS_FINALIZED)
93
    self.assertFalse(constants.JOBS_PENDING - constants.JOB_STATUS_ALL)
94
    self.assertFalse(constants.JOBS_FINALIZED - constants.JOB_STATUS_ALL)
95

    
96
  def testDefaultsForAllHypervisors(self):
97
    self.assertEqual(frozenset(constants.HVC_DEFAULTS.keys()),
98
                     constants.HYPER_TYPES)
99

    
100
  def testDefaultHypervisor(self):
101
    self.assertTrue(constants.DEFAULT_ENABLED_HYPERVISOR in
102
                    constants.HYPER_TYPES)
103

    
104
  def testExtraLogfiles(self):
105
    for daemon in constants.DAEMONS_EXTRA_LOGBASE:
106
      self.assertTrue(daemon in constants.DAEMONS)
107
      for log_reason in constants.DAEMONS_EXTRA_LOGBASE[daemon]:
108
        self.assertTrue(log_reason in constants.VALID_EXTRA_LOGREASONS)
109

    
110

    
111
class TestExportedNames(unittest.TestCase):
112
  _VALID_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]+$")
113
  _BUILTIN_NAME_RE = re.compile(r"^__\w+__$")
114
  _EXCEPTIONS = frozenset([
115
    "SplitVersion",
116
    "BuildVersion",
117
    ])
118

    
119
  def test(self):
120
    wrong = \
121
      set(itertools.ifilterfalse(self._BUILTIN_NAME_RE.match,
122
            itertools.ifilterfalse(self._VALID_NAME_RE.match,
123
                                   dir(constants))))
124
    wrong -= self._EXCEPTIONS
125
    self.assertFalse(wrong,
126
                     msg=("Invalid names exported from constants module: %s" %
127
                          utils.CommaJoin(sorted(wrong))))
128

    
129

    
130
class TestParameterNames(unittest.TestCase):
131
  """HV/BE parameter tests"""
132
  VALID_NAME = re.compile("^[a-zA-Z_][a-zA-Z0-9_]*$")
133

    
134
  def testNoDashes(self):
135
    for kind, source in [("hypervisor", constants.HVS_PARAMETER_TYPES),
136
                         ("backend", constants.BES_PARAMETER_TYPES),
137
                         ("nic", constants.NICS_PARAMETER_TYPES),
138
                         ("instdisk", constants.IDISK_PARAMS_TYPES),
139
                         ("instnic", constants.INIC_PARAMS_TYPES),
140
                        ]:
141
      for key in source:
142
        self.failUnless(self.VALID_NAME.match(key),
143
                        "The %s parameter '%s' contains invalid characters" %
144
                        (kind, key))
145

    
146

    
147
class TestConfdConstants(unittest.TestCase):
148
  """Test the confd constants"""
149

    
150
  def testFourCc(self):
151
    self.assertEqual(len(constants.CONFD_MAGIC_FOURCC), 4,
152
                     msg="Invalid fourcc len, should be 4")
153

    
154
  def testReqs(self):
155
    self.assertFalse(utils.FindDuplicates(constants.CONFD_REQS),
156
                     msg="Duplicated confd request code")
157

    
158
  def testReplStatuses(self):
159
    self.assertFalse(utils.FindDuplicates(constants.CONFD_REPL_STATUSES),
160
                     msg="Duplicated confd reply status code")
161

    
162
class TestDiskTemplateConstants(unittest.TestCase):
163

    
164
  def testPreference(self):
165
    self.assertEqual(set(constants.DISK_TEMPLATE_PREFERENCE),
166
                     set(constants.DISK_TEMPLATES))
167

    
168
  def testMapToStorageTypes(self):
169
    for disk_template in constants.DISK_TEMPLATES:
170
      self.assertTrue(
171
          constants.MAP_DISK_TEMPLATE_STORAGE_TYPE[disk_template] is not None)
172

    
173
  def testLvmDiskTemplates(self):
174
    lvm_by_storage_type = [
175
        dt for dt in constants.DISK_TEMPLATES
176
        if constants.ST_LVM_VG == constants.MAP_DISK_TEMPLATE_STORAGE_TYPE[dt]]
177
    self.assertEqual(set(lvm_by_storage_type), set(constants.DTS_LVM))
178

    
179

    
180
if __name__ == "__main__":
181
  testutils.GanetiTestProgram()