This adds a unit test which tests a successful run of LURenewCrypto. While writing this, some options for improvement became apparent and are fixed in this patch as well.
Signed-off-by: Helga Velroyen <[email protected]> --- lib/cmdlib/cluster.py | 16 +++++----- test/py/cmdlib/cluster_unittest.py | 62 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 7 deletions(-) diff --git a/lib/cmdlib/cluster.py b/lib/cmdlib/cluster.py index 24972fa..7959ff0 100644 --- a/lib/cmdlib/cluster.py +++ b/lib/cmdlib/cluster.py @@ -1,8 +1,8 @@ # # -# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 Google Inc. -# All rights reserved. +# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015 +# Google Inc. All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are @@ -129,12 +129,14 @@ class LUClusterRenewCrypto(NoHooksLU): except IOError: logging.info("No old certificate available.") - new_master_digest = _UpdateMasterClientCert(self, master_uuid, cluster, - feedback_fn) + # Technically it should not be necessary to set the cert + # paths. However, due to a bug in the mock library, we + # have to do this to be able to test the function properly. + _UpdateMasterClientCert( + self, master_uuid, cluster, feedback_fn, + client_cert=pathutils.NODED_CLIENT_CERT_FILE, + client_cert_tmp=pathutils.NODED_CLIENT_CERT_FILE_TMP) - utils.AddNodeToCandidateCerts(master_uuid, - new_master_digest, - cluster.candidate_certs) nodes = self.cfg.GetAllNodesInfo() for (node_uuid, node_info) in nodes.items(): if node_info.offline: diff --git a/test/py/cmdlib/cluster_unittest.py b/test/py/cmdlib/cluster_unittest.py index dc3a946..e3ea305 100644 --- a/test/py/cmdlib/cluster_unittest.py +++ b/test/py/cmdlib/cluster_unittest.py @@ -37,6 +37,8 @@ import OpenSSL import copy import unittest import operator +import shutil +import os from ganeti.cmdlib import cluster from ganeti import constants @@ -2264,5 +2266,65 @@ class TestLUClusterVerifyDisks(CmdlibTestCase): self.assertEqual(1, len(result["jobs"])) +class TestLUClusterRenewCrypto(CmdlibTestCase): + + def setUp(self): + super(TestLUClusterRenewCrypto, self).setUp() + self._node_cert = self._CreateTempFile() + shutil.copy(testutils.TestDataFilename("cert1.pem"), self._node_cert) + self._client_node_cert = self._CreateTempFile() + shutil.copy(testutils.TestDataFilename("cert2.pem"), self._client_node_cert) + self._client_node_cert_tmp = self._CreateTempFile() + + def tearDown(self): + super(TestLUClusterRenewCrypto, self).tearDown() + + def _GetFakeDigest(self, uuid): + """Creates a fake SSL digest depending on the UUID of a node. + + @type uuid: string + @param uuid: node UUID + @returns: a string impersonating a SSL digest + + """ + return "FA:KE:%s:%s:%s:%s" % (uuid[0:2], uuid[2:4], uuid[4:6], uuid[6:8]) + + @patchPathutils("cluster") + def testSuccessfulCase(self, pathutils): + + # patch pathutils to point to temporary files + pathutils.NODED_CERT_FILE = self._node_cert + pathutils.NODED_CLIENT_CERT_FILE = self._client_node_cert + pathutils.NODED_CLIENT_CERT_FILE_TMP = \ + self._client_node_cert_tmp + + # create a few non-master, online nodes + num_nodes = 3 + for _ in range(num_nodes): + self.cfg.AddNewNode() + + # make sure the RPC calls are successful for all nodes + self.rpc.call_node_crypto_tokens = \ + lambda node_uuid, _: self.RpcResultsBuilder() \ + .CreateSuccessfulNodeResult(node_uuid, + [(constants.CRYPTO_TYPE_SSL_DIGEST, self._GetFakeDigest(node_uuid))]) + + op = opcodes.OpClusterRenewCrypto() + self.ExecOpCode(op) + + # Check if the correct certificates exist and don't exist on the master + self.assertTrue(os.path.exists(pathutils.NODED_CERT_FILE)) + self.assertTrue(os.path.exists(pathutils.NODED_CLIENT_CERT_FILE)) + self.assertFalse(os.path.exists(pathutils.NODED_CLIENT_CERT_FILE_TMP)) + + # Check if we have the correct digests in the configuration + cluster = self.cfg.GetClusterInfo() + self.assertEqual(num_nodes + 1, len(cluster.candidate_certs)) + nodes = self.cfg.GetAllNodesInfo() + for (node_uuid, _) in nodes.items(): + expected_digest = self._GetFakeDigest(node_uuid) + self.assertEqual(expected_digest, cluster.candidate_certs[node_uuid]) + + if __name__ == "__main__": testutils.GanetiTestProgram() -- 2.2.0.rc0.207.ga3a616c
