4 # Copyright (C) 2012 Google Inc.
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.
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.
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
22 """Script for testing ganeti.ssconf"""
30 from ganeti import utils
31 from ganeti import constants
32 from ganeti import errors
33 from ganeti import ssconf
38 class TestReadSsconfFile(unittest.TestCase):
40 self.tmpdir = tempfile.mkdtemp()
43 shutil.rmtree(self.tmpdir)
45 def testReadDirectory(self):
46 self.assertRaises(EnvironmentError, ssconf.ReadSsconfFile, self.tmpdir)
48 def testNonExistantFile(self):
49 testfile = utils.PathJoin(self.tmpdir, "does.not.exist")
51 self.assertFalse(os.path.exists(testfile))
54 ssconf.ReadSsconfFile(testfile)
55 except EnvironmentError, err:
56 self.assertEqual(err.errno, errno.ENOENT)
58 self.fail("Exception was not raised")
60 def testEmptyFile(self):
61 testfile = utils.PathJoin(self.tmpdir, "empty")
63 utils.WriteFile(testfile, data="")
65 self.assertEqual(ssconf.ReadSsconfFile(testfile), "")
67 def testSingleLine(self):
68 testfile = utils.PathJoin(self.tmpdir, "data")
70 for nl in range(0, 10):
71 utils.WriteFile(testfile, data="Hello World" + ("\n" * nl))
73 self.assertEqual(ssconf.ReadSsconfFile(testfile),
76 def testExactlyMaxSize(self):
77 testfile = utils.PathJoin(self.tmpdir, "data")
79 data = "A" * ssconf._MAX_SIZE
80 utils.WriteFile(testfile, data=data)
82 self.assertEqual(os.path.getsize(testfile), ssconf._MAX_SIZE)
84 self.assertEqual(ssconf.ReadSsconfFile(testfile),
87 def testLargeFile(self):
88 testfile = utils.PathJoin(self.tmpdir, "data")
90 for size in [ssconf._MAX_SIZE + 1, ssconf._MAX_SIZE * 2]:
91 utils.WriteFile(testfile, data="A" * size)
92 self.assertTrue(os.path.getsize(testfile) > ssconf._MAX_SIZE)
93 self.assertRaises(RuntimeError, ssconf.ReadSsconfFile, testfile)
96 class TestSimpleStore(unittest.TestCase):
98 self.tmpdir = tempfile.mkdtemp()
99 self.sstore = ssconf.SimpleStore(cfg_location=self.tmpdir)
102 shutil.rmtree(self.tmpdir)
104 def testInvalidKey(self):
105 self.assertRaises(errors.ProgrammerError, self.sstore.KeyToFilename,
107 self.assertRaises(errors.ProgrammerError, self.sstore._ReadFile,
110 def testKeyToFilename(self):
111 for key in ssconf._VALID_KEYS:
112 result = self.sstore.KeyToFilename(key)
113 self.assertTrue(utils.IsBelowDir(self.tmpdir, result))
114 self.assertTrue(os.path.basename(result).startswith("ssconf_"))
116 def testReadFileNonExistingFile(self):
117 filename = self.sstore.KeyToFilename(constants.SS_CLUSTER_NAME)
119 self.assertFalse(os.path.exists(filename))
121 self.sstore._ReadFile(constants.SS_CLUSTER_NAME)
122 except errors.ConfigurationError, err:
123 self.assertTrue(str(err).startswith("Can't read ssconf file"))
125 self.fail("Exception was not raised")
127 for default in ["", "Hello World", 0, 100]:
128 self.assertFalse(os.path.exists(filename))
129 result = self.sstore._ReadFile(constants.SS_CLUSTER_NAME, default=default)
130 self.assertEqual(result, default)
132 def testReadFile(self):
133 utils.WriteFile(self.sstore.KeyToFilename(constants.SS_CLUSTER_NAME),
134 data="cluster.example.com")
136 self.assertEqual(self.sstore._ReadFile(constants.SS_CLUSTER_NAME),
137 "cluster.example.com")
139 self.assertEqual(self.sstore._ReadFile(constants.SS_CLUSTER_NAME,
140 default="something.example.com"),
141 "cluster.example.com")
144 class TestVerifyClusterName(unittest.TestCase):
146 self.tmpdir = tempfile.mkdtemp()
149 shutil.rmtree(self.tmpdir)
151 def testMissingFile(self):
152 tmploc = utils.PathJoin(self.tmpdir, "does-not-exist")
153 ssconf.VerifyClusterName(NotImplemented, _cfg_location=tmploc)
155 def testMatchingName(self):
156 tmpfile = utils.PathJoin(self.tmpdir, "ssconf_cluster_name")
158 for content in ["cluster.example.com", "cluster.example.com\n\n"]:
159 utils.WriteFile(tmpfile, data=content)
160 ssconf.VerifyClusterName("cluster.example.com",
161 _cfg_location=self.tmpdir)
163 def testNameMismatch(self):
164 tmpfile = utils.PathJoin(self.tmpdir, "ssconf_cluster_name")
166 for content in ["something.example.com", "foobar\n\ncluster.example.com"]:
167 utils.WriteFile(tmpfile, data=content)
168 self.assertRaises(errors.GenericError, ssconf.VerifyClusterName,
169 "cluster.example.com", _cfg_location=self.tmpdir)
172 if __name__ == "__main__":
173 testutils.GanetiTestProgram()