test/data/proc_drbd83.txt
python_tests = \
+ test/ganeti.backend_unittest.py \
test/ganeti.bdev_unittest.py \
test/ganeti.cli_unittest.py \
test/ganeti.cmdlib_unittest.py \
"""
result = {}
+ my_name = utils.HostInfo().name
+ port = utils.GetDaemonPort(constants.NODED)
if constants.NV_HYPERVISOR in what:
result[constants.NV_HYPERVISOR] = tmp = {}
if constants.NV_NODENETTEST in what:
result[constants.NV_NODENETTEST] = tmp = {}
- my_name = utils.HostInfo().name
my_pip = my_sip = None
for name, pip, sip in what[constants.NV_NODENETTEST]:
if name == my_name:
tmp[my_name] = ("Can't find my own primary/secondary IP"
" in the node list")
else:
- port = utils.GetDaemonPort(constants.NODED)
for name, pip, sip in what[constants.NV_NODENETTEST]:
fail = []
if not utils.TcpPing(pip, port, source=my_pip):
tmp[name] = ("failure using the %s interface(s)" %
" and ".join(fail))
+ if constants.NV_MASTERIP in what:
+ # FIXME: add checks on incoming data structures (here and in the
+ # rest of the function)
+ master_name, master_ip = what[constants.NV_MASTERIP]
+ if master_name == my_name:
+ source = constants.LOCALHOST_IP_ADDRESS
+ else:
+ source = None
+ result[constants.NV_MASTERIP] = utils.TcpPing(master_ip, port,
+ source=source)
+
if constants.NV_LVLIST in what:
try:
val = GetVolumeList(what[constants.NV_LVLIST])
"tcp communication with node '%s': %s",
anode, nresult[constants.NV_NODENETTEST][anode])
+ test = constants.NV_MASTERIP not in nresult
+ _ErrorIf(test, self.ENODENET, node,
+ "node hasn't returned node master IP reachability data")
+ if not test:
+ if not nresult[constants.NV_MASTERIP]:
+ if node == self.master_node:
+ msg = "the master node cannot reach the master IP (not configured?)"
+ else:
+ msg = "cannot reach the master IP"
+ _ErrorIf(True, self.ENODENET, node, msg)
+
+
def _VerifyInstance(self, instance, instanceconfig, node_image):
"""Verify an instance.
# FIXME: verify OS list
# do local checksums
master_files = [constants.CLUSTER_CONF_FILE]
+ master_node = self.master_node = self.cfg.GetMasterNode()
+ master_ip = self.cfg.GetMasterIP()
file_names = ssconf.SimpleStore().GetFileList()
file_names.extend(constants.ALL_CERT_FILES)
constants.NV_HVINFO: self.cfg.GetHypervisorType(),
constants.NV_NODESETUP: None,
constants.NV_TIME: None,
+ constants.NV_MASTERIP: (master_node, master_ip),
}
if vg_name is not None:
self.cfg.GetClusterName())
nvinfo_endtime = time.time()
- master_node = self.cfg.GetMasterNode()
all_drbd_map = self.cfg.ComputeDRBDMap()
feedback_fn("* Verifying node status")
NV_DRBDLIST = "drbd-list"
NV_NODESETUP = "nodesetup"
NV_TIME = "time"
+NV_MASTERIP = "master-ip"
# SSL certificate check constants (in days)
SSL_CERT_EXPIRATION_WARN = 30
--- /dev/null
+#!/usr/bin/python
+#
+
+# Copyright (C) 2010 Google Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 0.0510-1301, USA.
+
+
+"""Script for unittesting the backend module"""
+
+
+import os
+import unittest
+import time
+import tempfile
+import shutil
+
+from ganeti import backend
+from ganeti import constants
+from ganeti import utils
+
+import testutils
+
+
+class TestNodeVerify(testutils.GanetiTestCase):
+ def testMasterIPLocalhost(self):
+ # this a real functional test, but requires localhost to be reachable
+ local_data = (utils.HostInfo().name, constants.LOCALHOST_IP_ADDRESS)
+ result = backend.VerifyNode({constants.NV_MASTERIP: local_data}, None)
+ self.failUnless(constants.NV_MASTERIP in result,
+ "Master IP data not returned")
+ self.failUnless(result[constants.NV_MASTERIP], "Cannot reach localhost")
+
+ def testMasterIPUnreachable(self):
+ # Network 192.0.2.0/24 is reserved for test/documentation as per
+ # RFC 5735
+ bad_data = ("master.example.com", "192.0.2.1")
+ # we just test that whatever TcpPing returns, VerifyNode returns too
+ utils.TcpPing = lambda a, b, source=None: False
+ result = backend.VerifyNode({constants.NV_MASTERIP: bad_data}, None)
+ self.failUnless(constants.NV_MASTERIP in result,
+ "Master IP data not returned")
+ self.failIf(result[constants.NV_MASTERIP],
+ "Result from utils.TcpPing corrupted")
+
+
+if __name__ == "__main__":
+ testutils.GanetiTestProgram()
def testNowOwnAddress(self):
"""check that I don't own an address"""
- # network 192.0.2.0/24 is reserved for test/documentation as per
- # rfc 3330, so we *should* not have an address of this range... if
+ # Network 192.0.2.0/24 is reserved for test/documentation as per
+ # RFC 5735, so we *should* not have an address of this range... if
# this fails, we should extend the test to multiple addresses
DST_IP = "192.0.2.1"
self.failIf(OwnIpAddress(DST_IP), "Should not own IP address %s" % DST_IP)