[PATCH 10/11] Remove tests for the removed devicelibs modules
Vratislav Podzimek
vpodzime at redhat.com
Thu Feb 12 08:51:43 UTC 2015
These tests should be moved and ported as libblockdev tests.
Signed-off-by: Vratislav Podzimek <vpodzime at redhat.com>
---
tests/devicelibs_test/btrfs_test.py | 198 --------------
tests/devicelibs_test/crypto_test.py | 185 -------------
tests/devicelibs_test/lvm_test.py | 335 -----------------------
tests/devicelibs_test/mdraid_interrogate_test.py | 231 ----------------
tests/devicelibs_test/mdraid_test.py | 318 ---------------------
tests/devicelibs_test/swap_test.py | 74 -----
6 files changed, 1341 deletions(-)
delete mode 100755 tests/devicelibs_test/btrfs_test.py
delete mode 100755 tests/devicelibs_test/crypto_test.py
delete mode 100755 tests/devicelibs_test/lvm_test.py
delete mode 100755 tests/devicelibs_test/mdraid_interrogate_test.py
delete mode 100755 tests/devicelibs_test/mdraid_test.py
delete mode 100755 tests/devicelibs_test/swap_test.py
diff --git a/tests/devicelibs_test/btrfs_test.py b/tests/devicelibs_test/btrfs_test.py
deleted file mode 100755
index e3f658e..0000000
--- a/tests/devicelibs_test/btrfs_test.py
+++ /dev/null
@@ -1,198 +0,0 @@
-#!/usr/bin/python
-import os
-import os.path
-import subprocess
-import tempfile
-import unittest
-
-import blivet.devicelibs.btrfs as btrfs
-from blivet.errors import BTRFSError
-
-from tests import loopbackedtestcase
-
-class BTRFSMountDevice(loopbackedtestcase.LoopBackedTestCase):
- """A superclass that mounts and unmounts the filesystem.
- It must create the filesystem on its chosen devices before mounting.
- It always mounts the filesystem using self.device at self.mountpoint.
- """
- def __init__(self, methodName='runTest'):
- super(BTRFSMountDevice, self).__init__(methodName=methodName)
- self.device = None
- self.mountpoint = None
-
- def setUp(self):
- """After the DevicelibsTestCase setup, creates the filesystem on both
- devices and mounts on a tmp directory.
-
- Chooses the device to specify to mount command arbitrarily.
- """
- super(BTRFSMountDevice, self).setUp()
-
- btrfs.create_volume(self.loopDevices)
- self.device = self.loopDevices[0]
-
- self.mountpoint = tempfile.mkdtemp()
- rc = subprocess.call(["mount", self.device, self.mountpoint])
- if rc:
- raise OSError("mount failed to mount device %s" % self.device)
-
- def tearDown(self):
- """Before the DevicelibsTestCase cleanup unmount the device and
- remove the temporary mountpoint.
- """
- proc = subprocess.Popen(["umount", self.device])
- while True:
- proc.communicate()
- if proc.returncode is not None:
- rc = proc.returncode
- break
- if rc:
- raise OSError("failed to unmount device %s" % self.device)
-
- os.rmdir(self.mountpoint)
- super(BTRFSMountDevice, self).tearDown()
-
-class BTRFSAsRootTestCase1(loopbackedtestcase.LoopBackedTestCase):
-
- def testUnmountedBTRFS(self):
- """A series of simple tests on an unmounted file system.
-
- These tests are limited to simple creating and scanning.
- """
- _LOOP_DEV0 = self.loopDevices[0]
-
- ##
- ## create_volume
- ##
- # no devices specified
- with self.assertRaisesRegexp(ValueError, "no devices specified"):
- btrfs.create_volume([], data=0)
-
- # non-existant device
- with self.assertRaisesRegexp(ValueError, "one or more specified devices not present"):
- btrfs.create_volume(["/not/existing/device"])
-
- # bad data
- with self.assertRaisesRegexp(BTRFSError, "1"):
- btrfs.create_volume([_LOOP_DEV0], data="RaID7")
-
- # bad metadata
- with self.assertRaisesRegexp(BTRFSError, "1"):
- btrfs.create_volume([_LOOP_DEV0], metadata="RaID7")
-
- # pass
- self.assertEqual(btrfs.create_volume(self.loopDevices), 0)
-
- # already created
- with self.assertRaisesRegexp(BTRFSError, "1"):
- btrfs.create_volume([_LOOP_DEV0], metadata="RaID7")
-
- def testMkfsDefaults(self):
- _LOOP_DEV0 = self.loopDevices[0]
- _LOOP_DEV1 = self.loopDevices[1]
-
- btrfs.create_volume(self.loopDevices)
-
- self.assertEqual(btrfs.summarize_filesystem(_LOOP_DEV0)["label"], "none")
- self.assertEqual(btrfs.summarize_filesystem(_LOOP_DEV0)["num_devices"], "2")
-
- self.assertEqual(len(btrfs.list_devices(_LOOP_DEV0)), 2)
- self.assertIn(_LOOP_DEV1,
- [ dev["path"] for dev in btrfs.list_devices(_LOOP_DEV0) ])
- self.assertIn(_LOOP_DEV0,
- [ dev["path"] for dev in btrfs.list_devices(_LOOP_DEV1) ])
-
-class BTRFSAsRootTestCase2(BTRFSMountDevice):
- """Tests which require mounting the device."""
-
- def testSubvolume(self):
- """Tests which focus on subvolumes."""
-
- # no subvolumes yet
- self.assertEqual(btrfs.list_subvolumes(self.mountpoint), [])
-
- # the default subvolume is the root subvolume
- self.assertEqual(btrfs.get_default_subvolume(self.mountpoint),
- btrfs.MAIN_VOLUME_ID)
-
- # a new subvolume can be added succesfully below the mountpoint
- self.assertEqual(btrfs.create_subvolume(self.mountpoint, "SV1"), 0)
-
- # expect one subvolume
- subvolumes = btrfs.list_subvolumes(self.mountpoint)
- self.assertEqual(len(subvolumes), 1)
- self.assertEqual(subvolumes[0]['path'], 'SV1')
- self.assertEqual(subvolumes[0]['parent'], btrfs.MAIN_VOLUME_ID)
-
- # the same subvolume can be deleted
- self.assertEqual(btrfs.delete_subvolume(self.mountpoint, "SV1"), 0)
-
- # deleted subvolume is no longer present
- subvolumes = btrfs.list_subvolumes(self.mountpoint)
- self.assertNotIn("SV1", [v['path'] for v in subvolumes])
-
- # two distinct subvolumes, so both should be present
- self.assertEqual(btrfs.create_subvolume(self.mountpoint, "SV1"), 0)
- self.assertEqual(btrfs.create_subvolume(self.mountpoint, "SV2"), 0)
- subvolumes = btrfs.list_subvolumes(self.mountpoint)
- self.assertIn("SV1", [v['path'] for v in subvolumes])
- self.assertIn("SV2", [v['path'] for v in subvolumes])
-
- # we can remove one subvolume
- self.assertEqual(btrfs.delete_subvolume(self.mountpoint, "SV1"), 0)
- subvolumes = btrfs.list_subvolumes(self.mountpoint)
- self.assertNotIn("SV1", [v['path'] for v in subvolumes])
-
- # if the subvolume is already gone, an error is raised by btrfs
- with self.assertRaisesRegexp(BTRFSError, "1"):
- btrfs.delete_subvolume(self.mountpoint, "SV1")
-
- # if the subvolume is already there, an error is raise by btrfs
- with self.assertRaisesRegexp(BTRFSError, "1"):
- btrfs.create_subvolume(self.mountpoint, "SV2")
-
- # if we create SV1 once again it's back
- self.assertEqual(btrfs.create_subvolume(self.mountpoint, "SV1"), 0)
- subvolumes = btrfs.list_subvolumes(self.mountpoint)
- self.assertIn("SV1", [v['path'] for v in subvolumes])
-
- # we can create an additional subvolume beneath SV1
- self.assertEqual(btrfs.create_subvolume(os.path.join(self.mountpoint, "SV1"), "SV1.1"), 0)
- subvolumes = btrfs.list_subvolumes(self.mountpoint)
- self.assertEqual(len([v for v in subvolumes if v['path'].find("SV1.1") != -1]), 1)
-
-class BTRFSAsRootTestCase3(loopbackedtestcase.LoopBackedTestCase):
-
- def __init__(self, methodName='runTest'):
- super(BTRFSAsRootTestCase3, self).__init__(methodName=methodName, deviceSpec=[btrfs.MIN_MEMBER_SIZE, btrfs.MIN_MEMBER_SIZE])
-
- def testSmallDevice(self):
- """ Creation of a smallish device will result in an error if the
- data and metadata levels are specified differently, but not if
- they are unspecified.
- """
- with self.assertRaises(BTRFSError):
- btrfs.create_volume(self.loopDevices, data="single", metadata="dup")
- self.assertEqual(btrfs.create_volume(self.loopDevices), 0)
-
-class BTRFSTooSmallDeviceTestCase(loopbackedtestcase.LoopBackedTestCase):
-
- def __init__(self, methodName='runTest'):
- super(BTRFSTooSmallDeviceTestCase, self).__init__(methodName=methodName, deviceSpec=[btrfs.MIN_MEMBER_SIZE / 2, btrfs.MIN_MEMBER_SIZE])
-
- def testTooSmallDevice(self):
- """ If just one device is too small mkfs.btrfs will fail. """
- with self.assertRaises(BTRFSError):
- btrfs.create_volume(self.loopDevices)
-
-class BTRFSBigEnoughDeviceTestCase(loopbackedtestcase.LoopBackedTestCase):
-
- def __init__(self, methodName='runTest'):
- super(BTRFSBigEnoughDeviceTestCase, self).__init__(methodName=methodName, deviceSpec=[btrfs.MIN_MEMBER_SIZE])
-
- def testBigEnoughDevice(self):
- """ If all devices are large enough, mkfs.btrfs will succeed. """
- self.assertEqual(btrfs.create_volume(self.loopDevices), 0)
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/devicelibs_test/crypto_test.py b/tests/devicelibs_test/crypto_test.py
deleted file mode 100755
index cd7de7a..0000000
--- a/tests/devicelibs_test/crypto_test.py
+++ /dev/null
@@ -1,185 +0,0 @@
-#!/usr/bin/python
-import unittest
-
-import tempfile
-import os
-
-from blivet.devicelibs import crypto
-from blivet.errors import CryptoError
-from tests import loopbackedtestcase
-
-#FIXME: some of these tests expect behavior which is not correct
-#
-#The following is a list of the known incorrect behaviors:
-# *) The CryptSetup constructor raises an IOError if a device cannot be
-# opened. The various luks* methods that use CryptSetup don't catch this
-# exception, so they also raise an IOError. These methods should
-# all catch and handle the IOError appropriately.
-# *) All luks_* methods that take a key_file don't use it and they all raise
-# a value error if they do not get a passphrase, even if they get a keyfile.
-# This should happen only in the case where flags.installer_mode is True.
-# In other cases, the key_file should be used appropriately. The passphrase
-# and key_file arguments are mutually exclusive, and this should be handled
-# appropriately.
-# *) luks_status returns unmassaged int returned by CryptSetup.status,
-# not bool as the documentation states. The documentation should be brought
-# in line with the code, or vice-veras. It may be that the value returned by
-# CryptSetup.status is informative and that it would be useful to preserve it.
-# The numeric values are enumerated in libcryptsetup.h.
-
-class CryptoTestCase(loopbackedtestcase.LoopBackedTestCase):
-
- def testCryptoMisc(self):
- _LOOP_DEV0 = self.loopDevices[0]
- _LOOP_DEV1 = self.loopDevices[1]
-
-
- ##
- ## is_luks
- ##
- # pass
- self.assertEqual(crypto.is_luks(_LOOP_DEV0), -22)
- with self.assertRaisesRegexp(IOError, "Device cannot be opened"):
- crypto.is_luks("/not/existing/device")
-
- ##
- ## luks_format
- ##
- # pass
- self.assertEqual(crypto.luks_format(_LOOP_DEV0, passphrase="secret", cipher="aes-cbc-essiv:sha256", key_size=256), None)
- self.assertEqual(crypto.is_luks(_LOOP_DEV0), 0)
-
- # make a key file
- handle, keyfile = tempfile.mkstemp(prefix="key", text=False)
- os.write(handle, "nobodyknows")
- os.close(handle)
-
- # format with key file
- with self.assertRaisesRegexp(ValueError, "requires passphrase"):
- crypto.luks_format(_LOOP_DEV1, key_file=keyfile)
-
- # fail
- with self.assertRaisesRegexp(IOError, "Device cannot be opened"):
- crypto.luks_format("/not/existing/device", passphrase="secret", cipher="aes-cbc-essiv:sha256", key_size=256)
-
- # no passhprase or key file
- with self.assertRaisesRegexp(ValueError, "requires passphrase"):
- crypto.luks_format(_LOOP_DEV1, cipher="aes-cbc-essiv:sha256", key_size=256)
-
- ##
- ## is_luks
- ##
- # pass
- self.assertEqual(crypto.is_luks(_LOOP_DEV0), 0) # 0 = is luks
- self.assertEqual(crypto.is_luks(_LOOP_DEV1), -22)
-
- ##
- ## luks_add_key
- ##
- # pass
- self.assertEqual(crypto.luks_add_key(_LOOP_DEV0, new_passphrase="another-secret", passphrase="secret"), None)
-
- # fail
- with self.assertRaisesRegexp(CryptoError, "luks add key failed"):
- crypto.luks_add_key(_LOOP_DEV0, new_passphrase="another-secret", passphrase="wrong-passphrase")
-
- ##
- ## luks_remove_key
- ##
- # fail
- with self.assertRaisesRegexp(CryptoError, "luks remove key failed"):
- crypto.luks_remove_key(_LOOP_DEV0, del_passphrase="another-secret", passphrase="wrong-pasphrase")
-
- # pass
- self.assertEqual(crypto.luks_remove_key(_LOOP_DEV0, del_passphrase="another-secret", passphrase="secret"), None)
-
- # fail
- with self.assertRaisesRegexp(IOError, "Device cannot be opened"):
- crypto.luks_open("/not/existing/device", "another-crypted", passphrase="secret")
-
- # no passhprase or key file
- with self.assertRaisesRegexp(ValueError, "luks_format requires passphrase"):
- crypto.luks_open(_LOOP_DEV1, "another-crypted")
-
- with self.assertRaisesRegexp(IOError, "Device cannot be opened"):
- crypto.luks_status("another-crypted")
- with self.assertRaisesRegexp(IOError, "Device cannot be opened"):
- crypto.luks_close("wrong-name")
-
- # cleanup
- os.unlink(keyfile)
-
-class CryptoTestCase2(loopbackedtestcase.LoopBackedTestCase):
-
- def __init__(self, methodName='runTest'):
- """Set up the names by which luks knows these devices."""
- super(CryptoTestCase2, self).__init__(methodName=methodName)
- self._names = ["crypted", "encrypted"]
-
- def tearDown(self):
- """Close all devices just in case they are still open."""
- for name in self._names:
- try:
- crypto.luks_close(name)
- except (IOError, CryptoError):
- pass
- super(CryptoTestCase2, self).tearDown()
-
- def testCryptoOpen(self):
- _LOOP_DEV0 = self.loopDevices[0]
- _LOOP_DEV1 = self.loopDevices[1]
-
- _name0 = self._names[0]
- _name1 = self._names[1]
-
- ##
- ## luks_format
- ##
- # pass
- self.assertEqual(crypto.luks_format(_LOOP_DEV0, passphrase="secret", cipher="aes-cbc-essiv:sha256", key_size=256), None)
- self.assertEqual(crypto.luks_format(_LOOP_DEV1, passphrase="hidden", cipher="aes-cbc-essiv:sha256", key_size=256), None)
-
- ##
- ## luks_open
- ##
- # pass
- self.assertEqual(crypto.luks_open(_LOOP_DEV0, _name0, passphrase="secret"), None)
- self.assertEqual(crypto.luks_open(_LOOP_DEV1, _name1, passphrase="hidden"), None)
-
- ##
- ## luks_status
- ##
- # pass
- self.assertEqual(crypto.luks_status(_name0), 2)
- self.assertEqual(crypto.luks_status(_name1), 2)
-
- ##
- ## luks_uuid
- ##
- # pass
- uuid = crypto.luks_uuid(_LOOP_DEV0)
- self.assertEqual(crypto.luks_uuid(_LOOP_DEV0), uuid)
- uuid = crypto.luks_uuid(_LOOP_DEV1)
- self.assertEqual(crypto.luks_uuid(_LOOP_DEV1), uuid)
-
- ##
- ## luks_close
- ##
- # pass
- self.assertEqual(crypto.luks_close(_name0), None)
- self.assertEqual(crypto.luks_close(_name1), None)
-
- # already closed
- with self.assertRaisesRegexp(IOError, "Device cannot be opened"):
- crypto.luks_close("crypted")
- with self.assertRaisesRegexp(IOError, "Device cannot be opened"):
- crypto.luks_close("encrypted")
-
-class CryptoTestCase3(unittest.TestCase):
- def testPassphrase(self):
- exp = r"([0-9A-Za-z./]{5}-)*[0-9A-Za-z./]{0,4}"
- bp = crypto.generateBackupPassphrase()
- self.assertRegexpMatches(bp, exp)
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/devicelibs_test/lvm_test.py b/tests/devicelibs_test/lvm_test.py
deleted file mode 100755
index d88c451..0000000
--- a/tests/devicelibs_test/lvm_test.py
+++ /dev/null
@@ -1,335 +0,0 @@
-#!/usr/bin/python
-import unittest
-import glob
-
-import blivet.devicelibs.lvm as lvm
-from blivet.size import Size
-from blivet.errors import LVMError
-
-from tests import loopbackedtestcase
-
-# TODO: test cases for lvorigin, lvsnapshot*, thin*
-
-class LVMTestCase(unittest.TestCase):
-
- def testClampSize(self):
- # pass
- self.assertEqual(lvm.clampSize(Size("10 MiB"), Size("4 MiB")),
- Size("8 MiB"))
- self.assertEqual(lvm.clampSize(Size("10 MiB"), Size("4 MiB"),
- True),
- Size("12 MiB"))
-
-# FIXME: Some of these tests expect behavior that is not entirely correct.
-#
-# The following is a list of the known incorrect behaviors:
-# *) All lvm methods that take a device should explicitly raise an exception
-# when the device is non-existent. Currently, an exception is raised by the lvm
-# call if the device is non-existant, and usually that exception is caught and
-# an LVMError is then raised, but not always.
-
-class LVMAsRootTestCaseBase(loopbackedtestcase.LoopBackedTestCase):
-
- def __init__(self, methodName='runTest'):
- """Set up the structure of the volume group."""
- super(LVMAsRootTestCaseBase, self).__init__(methodName=methodName)
- self._vg_name = "test-vg"
- self._lv_name = "test-lv"
-
- def tearDown(self):
- """Destroy volume group."""
- try:
- lvm.lvdeactivate(self._vg_name, self._lv_name)
- lvm.lvremove(self._vg_name, self._lv_name)
- except LVMError:
- pass
-
- try:
- lvm.vgdeactivate(self._vg_name)
- lvm.vgreduce(self._vg_name, None, missing=True)
- except LVMError:
- pass
-
- try:
- lvm.vgremove(self._vg_name)
- except LVMError:
- pass
-
- for dev in self.loopDevices:
- try:
- lvm.pvremove(dev)
- except LVMError:
- pass
-
- super(LVMAsRootTestCaseBase, self).tearDown()
-
-class LVM_Metadata_Backup_TestCase(LVMAsRootTestCaseBase):
- def _list_backups(self):
- return set(glob.glob("/etc/lvm/archive/%s_*" % self._vg_name))
-
- def setUp(self):
- super(LVM_Metadata_Backup_TestCase, self).setUp()
- self._old_backups = self._list_backups()
- for dev in self.loopDevices:
- lvm.pvcreate(dev)
-
- def test_backup_enabled(self):
- lvm.flags.lvm_metadata_backup = True
- lvm.vgcreate(self._vg_name, self.loopDevices, Size("4MiB"))
-
- current_backups = self._list_backups()
- self.assertTrue(current_backups.issuperset(self._old_backups),
- "old backups disappeared??")
- self.assertTrue(current_backups.difference(self._old_backups),
- "lvm_metadata_backup enabled but no backups created")
-
- def test_backup_disabled(self):
- lvm.flags.lvm_metadata_backup = False
- lvm.vgcreate(self._vg_name, self.loopDevices, Size("4MiB"))
-
- self.assertEqual(self._old_backups, self._list_backups(),
- "lvm_metadata_backup disabled but backups created")
-
-
-class LVMAsRootTestCase(LVMAsRootTestCaseBase):
- def testLVM(self):
- _LOOP_DEV0 = self.loopDevices[0]
- _LOOP_DEV1 = self.loopDevices[1]
-
- ##
- ## pvcreate
- ##
- # pass
- for dev in self.loopDevices:
- self.assertEqual(lvm.pvcreate(dev), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.pvcreate("/not/existing/device")
-
- ##
- ## pvresize
- ##
- # pass
- for dev in self.loopDevices:
- self.assertEqual(lvm.pvresize(dev, Size("50MiB")), None)
- self.assertEqual(lvm.pvresize(dev, Size("100MiB")), None)
-
- # fail
- with self.assertRaisesRegexp(LVMError, "pvresize failed"):
- lvm.pvresize("/not/existing/device", Size("50MiB"))
-
- ##
- ## vgcreate
- ##
- # pass
- self.assertEqual(lvm.vgcreate(self._vg_name, self.loopDevices, Size("4MiB")), None)
-
- # fail
- with self.assertRaisesRegexp(LVMError, "vgcreate failed"):
- lvm.vgcreate("another-vg", ["/not/existing/device"], Size("4MiB"))
- # vg already exists
- with self.assertRaisesRegexp(LVMError, "vgcreate failed"):
- lvm.vgcreate(self._vg_name, [_LOOP_DEV0], Size("4MiB"))
- # pe size must be power of 2
- with self.assertRaisesRegexp(LVMError, "vgcreate failed"):
- lvm.vgcreate("another-vg", [_LOOP_DEV0], Size("5MiB"))
-
- ##
- ## vgdeactivate
- ##
- # pass
- self.assertEqual(lvm.vgdeactivate(self._vg_name), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.vgdeactivate("wrong-vg-name")
-
- ##
- ## vgreduce
- ##
- # pass
- self.assertEqual(lvm.vgreduce(self._vg_name, _LOOP_DEV1), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.vgreduce("wrong-vg-name", _LOOP_DEV1)
- with self.assertRaises(LVMError):
- lvm.vgreduce(self._vg_name, "/not/existing/device")
-
- ##
- ## vgactivate
- ##
- # pass
- self.assertEqual(lvm.vgactivate(self._vg_name), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.vgactivate("wrong-vg-name")
-
- ##
- ## pvinfo
- ##
- # pass
- self.assertEqual(lvm.pvinfo(device=_LOOP_DEV0)[_LOOP_DEV0]["LVM2_VG_NAME"], self._vg_name)
- # no vg
- self.assertEqual(lvm.pvinfo(device=_LOOP_DEV1)[_LOOP_DEV1]["LVM2_VG_NAME"], "")
-
- self.assertEqual(lvm.pvinfo(device="/not/existing/device"), {})
-
- ##
- ## vginfo
- ##
- # pass
- self.assertEqual(lvm.vginfo(self._vg_name)['LVM2_PV_COUNT'], "1")
-
- # fail
- with self.assertRaises(LVMError):
- lvm.vginfo("wrong-vg-name")
-
- ##
- ## lvcreate
- ##
- # pass
- self.assertEqual(lvm.lvcreate(self._vg_name, self._lv_name, Size("10MiB")), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.lvcreate("wrong-vg-name", "another-lv", Size("10MiB"))
-
- ##
- ## lvdeactivate
- ##
- # pass
- self.assertEqual(lvm.lvdeactivate(self._vg_name, self._lv_name), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.lvdeactivate(self._vg_name, "wrong-lv-name")
- with self.assertRaises(LVMError):
- lvm.lvdeactivate("wrong-vg-name", self._lv_name)
- with self.assertRaises(LVMError):
- lvm.lvdeactivate("wrong-vg-name", "wrong-lv-name")
-
- ##
- ## lvresize
- ##
- # pass
- self.assertEqual(lvm.lvresize(self._vg_name, self._lv_name, Size("60MiB")), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.lvresize(self._vg_name, "wrong-lv-name", Size("80MiB"))
- with self.assertRaises(LVMError):
- lvm.lvresize("wrong-vg-name", self._lv_name, Size("80MiB"))
- with self.assertRaises(LVMError):
- lvm.lvresize("wrong-vg-name", "wrong-lv-name", Size("80MiB"))
- # changing to same size
- with self.assertRaises(LVMError):
- lvm.lvresize(self._vg_name, self._lv_name, Size("60MiB"))
-
- ##
- ## lvactivate
- ##
- # pass
- self.assertEqual(lvm.lvactivate(self._vg_name, self._lv_name), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.lvactivate(self._vg_name, "wrong-lv-name")
- with self.assertRaises(LVMError):
- lvm.lvactivate("wrong-vg-name", self._lv_name)
- with self.assertRaises(LVMError):
- lvm.lvactivate("wrong-vg-name", "wrong-lv-name")
-
- ##
- ## lvs
- ##
- # pass
- full_name = "%s-%s" % (self._vg_name, self._lv_name)
- info = lvm.lvs(vg_name=self._vg_name)
- self.assertEqual(info[full_name]["LVM2_LV_NAME"], self._lv_name)
-
- # fail
- self.assertEqual(lvm.lvs(vg_name="wrong-vg-name"), {})
-
- ##
- ## lvremove
- ##
- # pass
- self.assertEqual(lvm.lvdeactivate(self._vg_name, self._lv_name), None) # is deactivation needed?
- self.assertEqual(lvm.lvremove(self._vg_name, self._lv_name), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.lvremove(self._vg_name, "wrong-lv-name")
- with self.assertRaises(LVMError):
- lvm.lvremove("wrong-vg-name", self._lv_name)
- with self.assertRaises(LVMError):
- lvm.lvremove("wrong-vg-name", "wrong-lv-name")
- # lv already removed
- with self.assertRaises(LVMError):
- lvm.lvremove(self._vg_name, self._lv_name)
-
- ##
- ## vgremove
- ##
- # pass
- self.assertEqual(lvm.vgremove(self._vg_name), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.vgremove("wrong-vg-name")
- # vg already removed
- with self.assertRaises(LVMError):
- lvm.vgremove(self._vg_name)
-
- ##
- ## pvremove
- ##
- # pass
- for dev in self.loopDevices:
- self.assertEqual(lvm.pvremove(dev), None)
-
- # fail
- with self.assertRaises(LVMError):
- lvm.pvremove("/not/existing/device")
- # pv already removed
- self.assertEqual(lvm.pvremove(_LOOP_DEV0), None)
-
-class LVMAsNonRootTestCase(unittest.TestCase):
- # pylint: disable=unused-argument
- def lvm_passthrough_args(self, args, *other_args, **kwargs):
- """ Just return args as passed so tests can validate them. """
- # pylint: disable=attribute-defined-outside-init
- self.lvm_argv = args
-
- def setUp(self):
- self.orig_lvm_func = lvm.lvm
- lvm.lvm = self.lvm_passthrough_args
-
- def tearDown(self):
- lvm.lvm = self.orig_lvm_func
-
- def testLVM(self):
- #
- # verify we pass appropriate args for various data alignment values
- #
-
- # default => do not specify a data alignment
- lvm.pvcreate('/dev/placeholder')
- argv = self.lvm_argv
- self.assertEqual(any(a.startswith('--dataalignment') for a in argv),
- False)
-
- # sizes get specified in KiB
- lvm.pvcreate('/dev/placeholder', data_alignment=Size('1 MiB'))
- argv = self.lvm_argv
- self.assertEqual("--dataalignment 1024k" in " ".join(argv), True)
-
- lvm.pvcreate('/dev/placeholder', data_alignment=Size(1023))
- argv = self.lvm_argv
- self.assertEqual("--dataalignment 0k" in " ".join(argv), True)
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/devicelibs_test/mdraid_interrogate_test.py b/tests/devicelibs_test/mdraid_interrogate_test.py
deleted file mode 100755
index cee2421..0000000
--- a/tests/devicelibs_test/mdraid_interrogate_test.py
+++ /dev/null
@@ -1,231 +0,0 @@
-#!/usr/bin/python
-import unittest
-import time
-import uuid
-
-import blivet.devicelibs.raid as raid
-import blivet.devicelibs.mdraid as mdraid
-
-from . import mdraid_test
-from blivet.errors import MDRaidError
-from blivet.size import Size
-
-class MDRaidInterrogateTestCase(mdraid_test.MDRaidAsRootTestCase):
-
- def _matchNames(self, found, expected):
- """ Match names found against expected names.
-
- :param found: a list of names found in result
- :type found: list of str
- :param expected: a list of expected names
- :type expected: list of str
- """
- for n in expected:
- self.assertIn(n, found, msg="name '%s' not in info" % n)
-
-class MDExamineTestCase(MDRaidInterrogateTestCase):
-
- def __init__(self, methodName='runTest'):
- super(MDExamineTestCase, self).__init__(methodName=methodName, deviceSpec=[Size("100 MiB") for _ in range(3)])
-
- names_0 = [
- 'DEVICE',
- 'MD_DEVICES',
- 'MD_EVENTS',
- 'MD_LEVEL',
- 'MD_UPDATE_TIME',
- 'MD_UUID'
- ]
-
- names_1 = [
- 'DEVICE',
- 'MD_ARRAY_SIZE',
- 'MD_DEV_UUID',
- 'MD_DEVICES',
- 'MD_EVENTS',
- 'MD_LEVEL',
- 'MD_METADATA',
- 'MD_NAME',
- 'MD_UPDATE_TIME',
- 'MD_UUID'
- ]
-
- names_container = [
- 'MD_DEVICES',
- 'MD_LEVEL',
- 'MD_METADATA',
- 'MD_UUID'
- ]
-
- def testMDExamineMDRaidArray(self):
- mdraid.mdcreate(self._dev_name, raid.RAID1, self.loopDevices)
- time.sleep(2) # wait for raid to settle
-
- # invoking mdexamine on the array itself raises an error
- with self.assertRaisesRegexp(MDRaidError, "mdexamine failed"):
- mdraid.mdexamine(self._dev_name)
-
- def testMDExamineNonMDRaid(self):
- # invoking mdexamine on any non-array member raises an error
- with self.assertRaisesRegexp(MDRaidError, "mdexamine failed"):
- mdraid.mdexamine(self.loopDevices[0])
-
- def _testMDExamine(self, names, metadataVersion=None, level=None, spares=0):
- """ Test mdexamine for a specified metadataVersion.
-
- :param list names: mdexamine's expected list of names to return
- :param str metadataVersion: the metadata version for the array
- :param object level: any valid RAID level descriptor
- :param int spares: the number of spares in this array
-
- Verifies that:
- - at least the predicted names are returned by mdexamine
- - RAID level and number of devices are correct
- - UUIDs have canonical form
- """
- level = raid.getRaidLevel(level or raid.RAID1)
- mdraid.mdcreate(self._dev_name, level, self.loopDevices, metadataVer=metadataVersion, spares=spares)
- time.sleep(2) # wait for raid to settle
-
- info = mdraid.mdexamine(self.loopDevices[0])
-
- self._matchNames(info.keys(), names)
-
- # check names with predictable values
- self.assertEqual(info['MD_DEVICES'], str(len(self.loopDevices) - spares))
- self.assertEqual(info['MD_LEVEL'], str(level))
-
- # verify that uuids are in canonical form
- for name in (k for k in iter(info.keys()) if k.endswith('UUID')):
- self.assertEqual(str(uuid.UUID(info[name])), info[name])
-
- def testMDExamineContainerDefault(self):
- self._testMDExamine(self.names_container, level="container")
-
- def testMDExamineDefault(self):
- self._testMDExamine(self.names_1)
-
- def testMDExamineSpares(self):
- self._testMDExamine(self.names_1, spares=1)
-
- def testMDExamine0(self):
- self._testMDExamine(self.names_0, metadataVersion='0')
-
- def testMDExamine0_90(self):
- self._testMDExamine(self.names_0, metadataVersion='0.90')
-
- def testMDExamine1(self):
- self._testMDExamine(self.names_1, metadataVersion='1')
-
- def testMDExamine1_2(self):
- self._testMDExamine(self.names_1, metadataVersion='1.2')
-
-class MDDetailTestCase(MDRaidInterrogateTestCase):
-
- def __init__(self, methodName='runTest'):
- super(MDDetailTestCase, self).__init__(methodName=methodName, deviceSpec=[Size("100 MiB") for _ in range(3)])
-
- names = [
- 'ACTIVE DEVICES',
- 'ARRAY SIZE',
- 'CREATION TIME',
- 'EVENTS',
- 'FAILED DEVICES',
- 'NAME',
- 'PERSISTENCE',
- 'RAID DEVICES',
- 'RAID LEVEL',
- 'SPARE DEVICES',
- 'STATE',
- 'TOTAL DEVICES',
- 'UPDATE TIME',
- 'USED DEV SIZE',
- 'UUID',
- 'VERSION',
- 'WORKING DEVICES'
- ]
-
- names_0 = [
- 'ACTIVE DEVICES',
- 'ARRAY SIZE',
- 'CREATION TIME',
- 'EVENTS',
- 'FAILED DEVICES',
- 'PERSISTENCE',
- 'PREFERRED MINOR',
- 'RAID DEVICES',
- 'RAID LEVEL',
- 'SPARE DEVICES',
- 'STATE',
- 'TOTAL DEVICES',
- 'UPDATE TIME',
- 'USED DEV SIZE',
- 'UUID',
- 'VERSION',
- 'WORKING DEVICES'
- ]
-
- names_container = [
- 'RAID LEVEL',
- 'WORKING DEVICES',
- 'VERSION',
- 'TOTAL DEVICES'
- ]
-
- def _testMDDetail(self, names, metadataVersion=None, level=None, spares=0):
- """ Test mddetail for a specified metadataVersion.
-
- :param list names: mdexamine's expected list of names to return
- :param str metadataVersion: the metadata version for the array
- :param object level: any valid RAID level descriptor
- :param int spares: the number of spares for this array
-
- Verifies that:
- - at least the predicted names are returned by mddetail
- - UUIDs have canonical form
- """
- level = raid.getRaidLevel(level) if level is not None else raid.RAID1
-
- mdraid.mdcreate(self._dev_name, level, self.loopDevices, metadataVer=metadataVersion, spares=spares)
- time.sleep(2) # wait for raid to settle
-
- info = mdraid.mddetail(self._dev_name)
-
- # info contains values for exactly names
- self._matchNames(info.keys(), names)
-
- # check names with predictable values
- self.assertEqual(info['RAID LEVEL'], str(level))
- self.assertEqual(info['TOTAL DEVICES'], str(len(self.loopDevices)))
- self.assertEqual(info['WORKING DEVICES'], str(len(self.loopDevices)))
-
- if level is not raid.Container:
- self.assertEqual(info['ACTIVE DEVICES'], str(len(self.loopDevices) - spares))
- self.assertEqual(info['FAILED DEVICES'], '0')
- self.assertEqual(info['SPARE DEVICES'], str(spares))
-
- # verify that uuid is in canonical form
- self.assertEqual(str(uuid.UUID(info['UUID'])), info['UUID'])
-
- def testMDDetail(self):
- self._testMDDetail(self.names)
-
- def testMDDetailSpares(self):
- self._testMDDetail(self.names, spares=1)
-
- def testMDDetail0_90(self):
- self._testMDDetail(self.names_0, metadataVersion='0.90')
-
- def testMDDetailMDDevice(self):
- mdraid.mdcreate(self._dev_name, raid.RAID1, self.loopDevices)
- time.sleep(2) # wait for raid to settle
-
- # invoking mddetail on a device raises an error
- with self.assertRaisesRegexp(MDRaidError, "mddetail failed"):
- mdraid.mddetail(self.loopDevices[0])
-
- def testMDDetailContainerDefault(self):
- self._testMDDetail(self.names_container, level="container")
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/devicelibs_test/mdraid_test.py b/tests/devicelibs_test/mdraid_test.py
deleted file mode 100755
index 4cd50a3..0000000
--- a/tests/devicelibs_test/mdraid_test.py
+++ /dev/null
@@ -1,318 +0,0 @@
-#!/usr/bin/python
-import unittest
-import time
-
-import blivet.devicelibs.raid as raid
-import blivet.devicelibs.mdraid as mdraid
-from blivet.errors import MDRaidError
-from blivet.size import Size
-
-from tests import loopbackedtestcase
-
-class MDRaidTestCase(unittest.TestCase):
-
- def testMDRaid(self):
-
- ##
- ## level lookup
- ##
- self.assertEqual(mdraid.RAID_levels.raidLevel("stripe").name, "raid0")
- self.assertEqual(mdraid.RAID_levels.raidLevel("mirror").name, "raid1")
- self.assertEqual(mdraid.RAID_levels.raidLevel("4").name, "raid4")
- self.assertEqual(mdraid.RAID_levels.raidLevel(5).name, "raid5")
- self.assertEqual(mdraid.RAID_levels.raidLevel("RAID6").name, "raid6")
- self.assertEqual(mdraid.RAID_levels.raidLevel("raid10").name, "raid10")
-
- ##
- ## get_raid_superblock_size
- ##
- self.assertEqual(mdraid.get_raid_superblock_size(Size("256 GiB")),
- Size("128 MiB"))
- self.assertEqual(mdraid.get_raid_superblock_size(Size("128 GiB")),
- Size("128 MiB"))
- self.assertEqual(mdraid.get_raid_superblock_size(Size("64 GiB")),
- Size("64 MiB"))
- self.assertEqual(mdraid.get_raid_superblock_size(Size("63 GiB")),
- Size("32 MiB"))
- self.assertEqual(mdraid.get_raid_superblock_size(Size("10 GiB")),
- Size("8 MiB"))
- self.assertEqual(mdraid.get_raid_superblock_size(Size("1 GiB")),
- Size("1 MiB"))
- self.assertEqual(mdraid.get_raid_superblock_size(Size("1023 MiB")),
- Size("1 MiB"))
- self.assertEqual(mdraid.get_raid_superblock_size(Size("512 MiB")),
- Size("1 MiB"))
-
- self.assertEqual(mdraid.get_raid_superblock_size(Size("257 MiB"),
- version="version"),
- mdraid.MD_SUPERBLOCK_SIZE)
-
- def testMisc(self):
- """ Miscellaneous testing. """
- self.assertEqual(
- mdraid.mduuid_from_canonical('3386ff85-f501-2621-4a43-5f061eb47236'),
- '3386ff85:f5012621:4a435f06:1eb47236'
- )
-
-class MDRaidAsRootTestCase(loopbackedtestcase.LoopBackedTestCase):
-
- def __init__(self, methodName='runTest', deviceSpec=None):
- super(MDRaidAsRootTestCase, self).__init__(methodName=methodName, deviceSpec=deviceSpec)
- self._dev_name = "/dev/md0"
-
- def tearDown(self):
- try:
- mdraid.mddeactivate(self._dev_name)
- except MDRaidError:
- pass
- for dev in self.loopDevices:
- try:
- mdraid.mdremove(self._dev_name, dev, fail=True)
- except MDRaidError:
- pass
- try:
- mdraid.mddestroy(dev)
- except MDRaidError:
- pass
-
- super(MDRaidAsRootTestCase, self).tearDown()
-
-class TestCaseFactory(object):
-
- @staticmethod
- def _minMembersForCreation(level):
- min_members = level.min_members
- return min_members if min_members > 1 else 2
-
-class JustAddTestCaseFactory(TestCaseFactory):
-
- @staticmethod
- def makeClass(name, level):
-
- def __init__(self, methodName='runTest'):
- super(self.__class__, self).__init__(methodName=methodName, deviceSpec=[Size("100 MiB") for _ in range(5)])
- self.longMessage = True
-
- def testAdd(self):
- """ Tests adding, not growing, a device. """
- initial_members = TestCaseFactory._minMembersForCreation(level)
- self.assertIsNone(mdraid.mdcreate(self._dev_name, level, self.loopDevices[:initial_members]))
- time.sleep(2) # wait for raid to settle
-
- new_member = self.loopDevices[initial_members]
-
- if level is raid.Container or level.has_redundancy():
- info_pre = mdraid.mddetail(self._dev_name)
- self.assertIsNone(mdraid.mdadd(self._dev_name, new_member))
- info_post = mdraid.mddetail(self._dev_name)
- keys = ['TOTAL DEVICES', 'WORKING DEVICES']
- for k in keys:
- self.assertEqual(int(info_pre[k]) + 1, int(info_post[k]), msg="key: %s" % k)
- if level is not raid.Container:
- keys = ['ACTIVE DEVICES', 'SPARE DEVICES']
- self.assertEqual(sum(int(info_pre[k]) for k in keys) + 1,
- sum(int(info_post[k]) for k in keys))
- self.assertEqual(info_pre['RAID DEVICES'], info_post['RAID DEVICES'])
- dev_info = mdraid.mdexamine(new_member)
- self.assertEqual(info_post['UUID'], dev_info['MD_UUID'])
- else:
- with self.assertRaises(MDRaidError):
- mdraid.mdadd(self._dev_name, new_member)
-
- return type(
- name,
- (MDRaidAsRootTestCase,),
- {
- '__init__': __init__,
- 'testAdd': testAdd,
- })
-
-class GrowTestCaseFactory(object):
-
- @staticmethod
- def makeClass(name, level):
-
- def __init__(self, methodName='runTest'):
- super(self.__class__, self).__init__(methodName=methodName, deviceSpec=[Size("100 MiB") for _ in range(6)])
- self.longMessage = True
-
- def testGrow(self):
- """ Tests growing a device by exactly 1. """
- initial_members = TestCaseFactory._minMembersForCreation(level)
- self.assertIsNone(mdraid.mdcreate(self._dev_name, level, self.loopDevices[:initial_members]))
- time.sleep(3) # wait for raid to settle
-
- new_member = self.loopDevices[initial_members]
- info_pre = mdraid.mddetail(self._dev_name)
-
- # for linear RAID the new number of devices must not be specified
- # for all other levels the new number of devices must be specified
- if level is raid.Linear:
- with self.assertRaises(MDRaidError):
- mdraid.mdadd(self._dev_name, new_member, grow_mode=True, raid_devices=initial_members + 1)
- self.assertIsNone(mdraid.mdadd(self._dev_name, new_member, grow_mode=True))
- else:
- with self.assertRaises(MDRaidError):
- mdraid.mdadd(self._dev_name, new_member, grow_mode=True)
- self.assertIsNone(mdraid.mdadd(self._dev_name, new_member, grow_mode=True, raid_devices=initial_members + 1))
-
- info_post = mdraid.mddetail(self._dev_name)
- keys = ['RAID DEVICES', 'TOTAL DEVICES', 'WORKING DEVICES']
- for k in keys:
- self.assertEqual(int(info_pre[k]) + 1, int(info_post[k]), msg="key: %s" % k)
- if level is not raid.Container:
- keys = ['ACTIVE DEVICES', 'SPARE DEVICES']
- self.assertEqual(sum(int(info_pre[k]) for k in keys) + 1,
- sum(int(info_post[k]) for k in keys),
- msg="%s" % " + ".join(keys))
- dev_info = mdraid.mdexamine(new_member)
- self.assertEqual(info_post['UUID'], dev_info['MD_UUID'], msg="key: UUID")
-
- def testGrowBig(self):
- """ Test growing a device beyond its size. """
- initial_members = TestCaseFactory._minMembersForCreation(level)
- self.assertIsNone(mdraid.mdcreate(self._dev_name, level, self.loopDevices[:initial_members]))
- time.sleep(3) # wait for raid to settle
-
- new_member = self.loopDevices[initial_members]
- if level is raid.Linear:
- self.assertIsNone(mdraid.mdadd(self._dev_name, new_member, grow_mode=True))
- else:
- with self.assertRaises(MDRaidError):
- mdraid.mdadd(self._dev_name, new_member, grow_mode=True, raid_devices=initial_members + 2)
-
- def testGrowSmall(self):
- """ Test decreasing size of device. """
- initial_members = TestCaseFactory._minMembersForCreation(level) + 1
- self.assertIsNone(mdraid.mdcreate(self._dev_name, level, self.loopDevices[:initial_members]))
- time.sleep(2) # wait for raid to settle
-
- new_member = self.loopDevices[initial_members]
-
- with self.assertRaises(MDRaidError):
- mdraid.mdadd(self._dev_name, new_member, grow_mode=True, raid_devices=initial_members - 1)
-
- return type(
- name,
- (MDRaidAsRootTestCase,),
- {
- '__init__': __init__,
- 'testGrow': testGrow,
- 'testGrowBig' : testGrowBig,
- 'testGrowSmall': testGrowSmall
- })
-
-
-# make some test cases for every RAID level
-levels = sorted(mdraid.RAID_levels, cmp=lambda x, y: cmp(x.name, y.name))
-for l in levels:
- classname = "%sJustAddTestCase" % l
- globals()[classname] = JustAddTestCaseFactory.makeClass(classname, l)
-
- if l is not raid.Container:
- classname = "%sGrowTestCase" % l
- globals()[classname] = GrowTestCaseFactory.makeClass(classname, l)
-
- at unittest.skip("temporarily disabled due to bug 1162823")
-class SimpleRaidTest(MDRaidAsRootTestCase):
-
- def __init__(self, methodName='runTest'):
- super(SimpleRaidTest, self).__init__(methodName=methodName, deviceSpec=[Size("100 MiB") for _ in range(3)])
- self.longMessage = True
-
- def testMDRaidAsRoot(self):
- ##
- ## mdcreate
- ##
- # pass
- self.assertIsNone(mdraid.mdcreate(self._dev_name, raid.RAID1, self.loopDevices))
- time.sleep(2) # wait for raid to settle
-
- # fail
- with self.assertRaises(MDRaidError):
- mdraid.mdcreate("/dev/md1", "raid1", ["/not/existing/dev0", "/not/existing/dev1"])
-
- # fail
- with self.assertRaises(MDRaidError):
- mdraid.mdadd(self._dev_name, "/not/existing/device")
- time.sleep(2) # wait for raid to settle
-
- # removing and re-adding a component device should succeed
- self.assertIsNone(mdraid.mdremove(self._dev_name, self.loopDevices[2], fail=True))
- time.sleep(2) # wait for raid to settle
- self.assertIsNone(mdraid.mdadd(self._dev_name, self.loopDevices[2]))
- time.sleep(3) # wait for raid to settle
-
- # it is not possible to add a device that has already been added
- with self.assertRaises(MDRaidError):
- mdraid.mdadd(self._dev_name, self.loopDevices[2])
-
- self.assertIsNone(mdraid.mdremove(self._dev_name, self.loopDevices[2], fail=True))
- time.sleep(2) # wait for raid to settle
-
- # can not re-add in incremental mode because the array is active
- with self.assertRaises(MDRaidError):
- mdraid.mdnominate(self.loopDevices[2])
-
- info_pre = mdraid.mddetail(self._dev_name)
-
- ##
- ## mddeactivate
- ##
- # pass
- self.assertIsNone(mdraid.mddeactivate(self._dev_name))
-
- # once the array is deactivated, can add in incremental mode
- self.assertIsNone(mdraid.mdnominate(self.loopDevices[2]))
-
- # but cannot re-add twice
- with self.assertRaises(MDRaidError):
- mdraid.mdnominate(self.loopDevices[2])
-
- # fail
- with self.assertRaises(MDRaidError):
- mdraid.mddeactivate("/not/existing/md")
-
-
- ##
- ## mdactivate
- ##
- with self.assertRaises(MDRaidError):
- mdraid.mdactivate("/not/existing/md", array_uuid=32)
- # requires uuid
- with self.assertRaises(MDRaidError):
- mdraid.mdactivate("/dev/md1")
-
- self.assertIsNone(mdraid.mdactivate(self._dev_name, array_uuid=mdraid.mduuid_from_canonical(info_pre['UUID'])))
- time.sleep(2)
- info_post = mdraid.mddetail(self._dev_name)
-
- # the array should remain mostly the same across activations
- changeable_values = (
- 'ACTIVE DEVICES',
- 'STATE',
- 'TOTAL DEVICES',
- 'WORKING DEVICES'
- )
- for k in (k for k in info_pre.keys() if k not in changeable_values):
- self.assertEqual(info_pre[k], info_post[k], msg="key: %s" % k)
-
- # deactivating the array one more time
- self.assertIsNone(mdraid.mddeactivate(self._dev_name))
-
- ##
- ## mddestroy
- ##
- # pass
- for dev in self.loopDevices:
- self.assertIsNone(mdraid.mddestroy(dev))
-
- # pass
- # Note that these should fail because mdadm is unable to locate the
- # device. The mdadm Kill function does return 2, but the mdadm process
- # returns 0 for both tests.
- self.assertIsNone(mdraid.mddestroy(self._dev_name))
- self.assertIsNone(mdraid.mddestroy("/not/existing/device"))
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/devicelibs_test/swap_test.py b/tests/devicelibs_test/swap_test.py
deleted file mode 100755
index 1b8e1aa..0000000
--- a/tests/devicelibs_test/swap_test.py
+++ /dev/null
@@ -1,74 +0,0 @@
-#!/usr/bin/python
-import unittest
-
-import blivet.devicelibs.swap as swap
-from blivet.errors import SwapError
-
-from tests import loopbackedtestcase
-
-class SwapTestCase(loopbackedtestcase.LoopBackedTestCase):
-
- def testSwap(self):
- _LOOP_DEV0 = self.loopDevices[0]
- _LOOP_DEV1 = self.loopDevices[1]
-
-
- ##
- ## mkswap
- ##
- # pass
- self.assertEqual(swap.mkswap(_LOOP_DEV0, "swap"), None)
-
- # fail
- with self.assertRaises(SwapError):
- swap.mkswap("/not/existing/device")
-
- ##
- ## swapon
- ##
- # pass
- self.assertEqual(swap.swapon(_LOOP_DEV0, 1), None)
-
- # fail
- with self.assertRaises(SwapError):
- swap.swapon("/not/existing/device")
- # not a swap partition
- with self.assertRaises(SwapError):
- swap.swapon(_LOOP_DEV1)
-
- # pass
- # make another swap
- self.assertEqual(swap.mkswap(_LOOP_DEV1, "another-swap"), None)
- self.assertEqual(swap.swapon(_LOOP_DEV1), None)
-
- ##
- ## swapstatus
- ##
- # pass
- self.assertEqual(swap.swapstatus(_LOOP_DEV0), True)
- self.assertEqual(swap.swapstatus(_LOOP_DEV1), True)
-
- # does not fail
- self.assertEqual(swap.swapstatus("/not/existing/device"), False)
-
- ##
- ## swapoff
- ##
- # pass
- self.assertEqual(swap.swapoff(_LOOP_DEV1), None)
-
- # check status
- self.assertEqual(swap.swapstatus(_LOOP_DEV0), True)
- self.assertEqual(swap.swapstatus(_LOOP_DEV1), False)
-
- self.assertEqual(swap.swapoff(_LOOP_DEV0), None)
-
- # fail
- with self.assertRaises(SwapError):
- swap.swapoff("/not/existing/device")
- # already off
- with self.assertRaises(SwapError):
- swap.swapoff(_LOOP_DEV0)
-
-if __name__ == "__main__":
- unittest.main()
--
2.1.0
More information about the anaconda-patches
mailing list