code.grnet.gr Git - ganeti-local/blob - test/cfgupgrade_unittest.py

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 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 testing tools/cfgupgrade"""
  23
  24 import os
  25 import sys
  26 import unittest
  27 import shutil
  28 import tempfile
  29 import operator
  30
  31 from ganeti import constants
  32 from ganeti import utils
  33 from ganeti import errors
  34 from ganeti import serializer
  35
  36 import testutils
  37
  38
  39 def _RunUpgrade(path, dry_run, no_verify):
  40   cmd = [sys.executable, "%s/tools/cfgupgrade" % testutils.GetSourceDir(),
  41          "--debug", "--force", "--path=%s" % path]
  42   if dry_run:
  43     cmd.append("--dry-run")
  44   if no_verify:
  45     cmd.append("--no-verify")
  46
  47   result = utils.RunCmd(cmd, cwd=os.getcwd())
  48   if result.failed:
  49     raise Exception("cfgupgrade failed: %s, output %r" %
  50                     (result.fail_reason, result.output))
  51
  52
  53 class TestCfgupgrade(unittest.TestCase):
  54   def setUp(self):
  55     self.tmpdir = tempfile.mkdtemp()
  56
  57     self.config_path = utils.PathJoin(self.tmpdir, "config.data")
  58     self.noded_cert_path = utils.PathJoin(self.tmpdir, "server.pem")
  59     self.rapi_cert_path = utils.PathJoin(self.tmpdir, "rapi.pem")
  60     self.known_hosts_path = utils.PathJoin(self.tmpdir, "known_hosts")
  61     self.confd_hmac_path = utils.PathJoin(self.tmpdir, "hmac.key")
  62     self.cds_path = utils.PathJoin(self.tmpdir, "cluster-domain-secret")
  63
  64   def tearDown(self):
  65     shutil.rmtree(self.tmpdir)
  66
  67   def _LoadConfig(self):
  68     return serializer.LoadJson(utils.ReadFile(self.config_path))
  69
  70   def _CreateValidConfigDir(self):
  71     utils.WriteFile(self.noded_cert_path, data="")
  72     utils.WriteFile(self.known_hosts_path, data="")
  73
  74   def testNoConfigDir(self):
  75     self.assertFalse(utils.ListVisibleFiles(self.tmpdir))
  76     self.assertRaises(Exception, _RunUpgrade, self.tmpdir, False, True)
  77     self.assertRaises(Exception, _RunUpgrade, self.tmpdir, True, True)
  78
  79   def testInconsistentConfig(self):
  80     self._CreateValidConfigDir()
  81     # There should be no "config_version"
  82     cfg = {
  83       "version": 0,
  84       "cluster": {
  85         "config_version": 0,
  86         },
  87       }
  88     utils.WriteFile(self.config_path, data=serializer.DumpJson(cfg))
  89     self.assertRaises(Exception, _RunUpgrade, self.tmpdir, False, True)
  90
  91   def testInvalidConfig(self):
  92     self._CreateValidConfigDir()
  93     # Missing version from config
  94     utils.WriteFile(self.config_path, data=serializer.DumpJson({}))
  95     self.assertRaises(Exception, _RunUpgrade, self.tmpdir, False, True)
  96
  97   def _TestSimpleUpgrade(self, from_version, dry_run):
  98     cfg = {
  99       "version": from_version,
 100       "cluster": {},
 101       }
 102     self._CreateValidConfigDir()
 103     utils.WriteFile(self.config_path, data=serializer.DumpJson(cfg))
 104
 105     self.assertFalse(os.path.isfile(self.rapi_cert_path))
 106     self.assertFalse(os.path.isfile(self.confd_hmac_path))
 107     self.assertFalse(os.path.isfile(self.cds_path))
 108
 109     _RunUpgrade(self.tmpdir, dry_run, True)
 110
 111     if dry_run:
 112       expversion = from_version
 113       checkfn = operator.not_
 114     else:
 115       expversion = constants.CONFIG_VERSION
 116       checkfn = operator.truth
 117
 118     self.assert_(checkfn(os.path.isfile(self.rapi_cert_path)))
 119     self.assert_(checkfn(os.path.isfile(self.confd_hmac_path)))
 120     self.assert_(checkfn(os.path.isfile(self.cds_path)))
 121
 122     newcfg = self._LoadConfig()
 123     self.assertEqual(newcfg["version"], expversion)
 124
 125   def testUpgradeFrom_2_0(self):
 126     self._TestSimpleUpgrade(constants.BuildVersion(2, 0, 0), False)
 127
 128   def testUpgradeFrom_2_1(self):
 129     self._TestSimpleUpgrade(constants.BuildVersion(2, 1, 0), False)
 130
 131   def testUpgradeCurrent(self):
 132     self._TestSimpleUpgrade(constants.CONFIG_VERSION, False)
 133
 134   def testUpgradeDryRunFrom_2_0(self):
 135     self._TestSimpleUpgrade(constants.BuildVersion(2, 0, 0), True)
 136
 137   def testUpgradeDryRunFrom_2_1(self):
 138     self._TestSimpleUpgrade(constants.BuildVersion(2, 1, 0), True)
 139
 140   def testUpgradeCurrentDryRun(self):
 141     self._TestSimpleUpgrade(constants.CONFIG_VERSION, True)
 142
 143
 144 if __name__ == "__main__":
 145   testutils.GanetiTestProgram()