backend: Check for shared storage also

[ganeti-local] / test / ganeti.backend_unittest.py

diff --git a/test/ganeti.backend_unittest.py b/test/ganeti.backend_unittest.py
old mode 100644 (file)
new mode 100755 (executable)
index 2c94a75..ff9a33c
--- a/test/ganeti.backend_unittest.py
+++ b/test/ganeti.backend_unittest.py
@@ -16,29 +16,66 @@
 # 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.
+# 02110-1301, USA.
 
 
-"""Script for unittesting the backend module"""
-
+"""Script for testing ganeti.backend"""
 
 import os
-import unittest
-import time
-import tempfile
+import sys
 import shutil
+import tempfile
+import unittest
 
-from ganeti import backend
-from ganeti import constants
 from ganeti import utils
+from ganeti import constants
+from ganeti import backend
+from ganeti import netutils
 
 import testutils
 
 
+class TestX509Certificates(unittest.TestCase):
+  def setUp(self):
+    self.tmpdir = tempfile.mkdtemp()
+
+  def tearDown(self):
+    shutil.rmtree(self.tmpdir)
+
+  def test(self):
+    (name, cert_pem) = backend.CreateX509Certificate(300, cryptodir=self.tmpdir)
+
+    self.assertEqual(utils.ReadFile(os.path.join(self.tmpdir, name,
+                                                 backend._X509_CERT_FILE)),
+                     cert_pem)
+    self.assert_(0 < os.path.getsize(os.path.join(self.tmpdir, name,
+                                                  backend._X509_KEY_FILE)))
+
+    (name2, cert_pem2) = \
+      backend.CreateX509Certificate(300, cryptodir=self.tmpdir)
+
+    backend.RemoveX509Certificate(name, cryptodir=self.tmpdir)
+    backend.RemoveX509Certificate(name2, cryptodir=self.tmpdir)
+
+    self.assertEqual(utils.ListVisibleFiles(self.tmpdir), [])
+
+  def testNonEmpty(self):
+    (name, _) = backend.CreateX509Certificate(300, cryptodir=self.tmpdir)
+
+    utils.WriteFile(utils.PathJoin(self.tmpdir, name, "hello-world"),
+                    data="Hello World")
+
+    self.assertRaises(backend.RPCFail, backend.RemoveX509Certificate,
+                      name, cryptodir=self.tmpdir)
+
+    self.assertEqual(utils.ListVisibleFiles(self.tmpdir), [name])
+
+
 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)
+    local_data = (netutils.Hostname.GetSysName(),
+                  constants.IP4_ADDRESS_LOCALHOST)
     result = backend.VerifyNode({constants.NV_MASTERIP: local_data}, None)
     self.failUnless(constants.NV_MASTERIP in result,
                     "Master IP data not returned")
@@ -46,15 +83,15 @@ class TestNodeVerify(testutils.GanetiTestCase):
 
   def testMasterIPUnreachable(self):
     # Network 192.0.2.0/24 is reserved for test/documentation as per
-    # RFC 5735
+    # RFC 5737
     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
+    netutils.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")
+                "Result from netutils.TcpPing corrupted")
 
 
 if __name__ == "__main__":