mirror of
git://sourceware.org/git/lvm2.git
synced 2024-10-28 20:25:52 +03:00
6f216692b4
When we test Vg.LvCreateRaid some of the hidden LVs volume type go from 'I' to 'i' between the time it takes us to create the LV and the time it takes to call into refresh to verify the service is up to date. This is a fairly rare occurance.
1165 lines
30 KiB
Python
Executable File
1165 lines
30 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
# Copyright (C) 2015-2016 Red Hat, Inc. All rights reserved.
|
|
#
|
|
# This copyrighted material is made available to anyone wishing to use,
|
|
# modify, copy, or redistribute it subject to the terms and conditions
|
|
# of the GNU General Public License v.2.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import dbus
|
|
# noinspection PyUnresolvedReferences
|
|
from dbus.mainloop.glib import DBusGMainLoop
|
|
import unittest
|
|
import sys
|
|
import time
|
|
import pyudev
|
|
import os
|
|
from testlib import *
|
|
|
|
use_session = os.getenv('LVMDBUSD_USE_SESSION', False)
|
|
|
|
if use_session:
|
|
bus = dbus.SessionBus(mainloop=DBusGMainLoop())
|
|
else:
|
|
bus = dbus.SystemBus(mainloop=DBusGMainLoop())
|
|
|
|
|
|
def get_objects():
|
|
rc = {MANAGER_INT: [], PV_INT: [], VG_INT: [], LV_INT: [],
|
|
THINPOOL_INT: [], JOB_INT: [], SNAPSHOT_INT: [], LV_COMMON_INT: [],
|
|
CACHE_POOL_INT: [], CACHE_LV_INT: []}
|
|
|
|
manager = dbus.Interface(bus.get_object(
|
|
BUSNAME, "/com/redhat/lvmdbus1"),
|
|
"org.freedesktop.DBus.ObjectManager")
|
|
|
|
objects = manager.GetManagedObjects()
|
|
|
|
for object_path, val in list(objects.items()):
|
|
for interface, props in list(val.items()):
|
|
o = ClientProxy(bus, object_path, interface, props)
|
|
rc[interface].append(o)
|
|
|
|
return rc, bus
|
|
|
|
|
|
def set_execution(lvmshell):
|
|
lvm_manager = dbus.Interface(bus.get_object(
|
|
BUSNAME, "/com/redhat/lvmdbus1/Manager"),
|
|
"com.redhat.lvmdbus1.Manager")
|
|
lvm_manager.UseLvmShell(lvmshell)
|
|
|
|
|
|
# noinspection PyUnresolvedReferences
|
|
class TestDbusService(unittest.TestCase):
|
|
def setUp(self):
|
|
# Because of the sensitive nature of running LVM tests we will only
|
|
# run if we have PVs and nothing else, so that we can be confident that
|
|
# we are not mucking with someones data on their system
|
|
self.objs, self.bus = get_objects()
|
|
if len(self.objs[PV_INT]) == 0:
|
|
print('No PVs present exiting!')
|
|
sys.exit(1)
|
|
if len(self.objs[MANAGER_INT]) != 1:
|
|
print('Expecting a manager object!')
|
|
sys.exit(1)
|
|
|
|
if len(self.objs[VG_INT]) != 0:
|
|
print('Expecting no VGs to exist!')
|
|
sys.exit(1)
|
|
|
|
self.pvs = []
|
|
for p in self.objs[PV_INT]:
|
|
self.pvs.append(p.Pv.Name)
|
|
|
|
def tearDown(self):
|
|
# If we get here it means we passed setUp, so lets remove anything
|
|
# and everything that remains, besides the PVs themselves
|
|
self.objs, self.bus = get_objects()
|
|
for v in self.objs[VG_INT]:
|
|
# print "DEBUG: Removing VG= ", v.Uuid, v.Name
|
|
v.Vg.Remove(-1, {})
|
|
|
|
# Check to make sure the PVs we had to start exist, else re-create
|
|
# them
|
|
if len(self.pvs) != len(self.objs[PV_INT]):
|
|
for p in self.pvs:
|
|
found = False
|
|
for pc in self.objs[PV_INT]:
|
|
if pc.Pv.Name == p:
|
|
found = True
|
|
break
|
|
|
|
if not found:
|
|
# print('Re-creating PV=', p)
|
|
self._pv_create(p)
|
|
|
|
def _pv_create(self, device):
|
|
pv_path = self.objs[MANAGER_INT][0].Manager.PvCreate(device, -1, {})[0]
|
|
self.assertTrue(pv_path is not None and len(pv_path) > 0)
|
|
return pv_path
|
|
|
|
def _manager(self):
|
|
return self.objs[MANAGER_INT][0]
|
|
|
|
def _refresh(self):
|
|
return self._manager().Manager.Refresh()
|
|
|
|
def test_refresh(self):
|
|
rc = self._refresh()
|
|
self.assertEqual(rc, 0)
|
|
|
|
def test_version(self):
|
|
rc = self.objs[MANAGER_INT][0].Manager.Version
|
|
self.assertTrue(rc is not None and len(rc) > 0)
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def _vg_create(self, pv_paths=None):
|
|
|
|
if not pv_paths:
|
|
pv_paths = [self.objs[PV_INT][0].object_path]
|
|
|
|
vg_name = rs(8, '_vg')
|
|
|
|
vg_path = self.objs[MANAGER_INT][0].Manager.VgCreate(
|
|
vg_name,
|
|
pv_paths,
|
|
-1,
|
|
{})[0]
|
|
self.assertTrue(vg_path is not None and len(vg_path) > 0)
|
|
return ClientProxy(self.bus, vg_path)
|
|
|
|
def test_vg_create(self):
|
|
self._vg_create()
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def test_vg_delete(self):
|
|
vg = self._vg_create().Vg
|
|
vg.Remove(-1, {})
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def _pv_remove(self, pv):
|
|
rc = pv.Pv.Remove(-1, {})
|
|
return rc
|
|
|
|
def test_pv_remove_add(self):
|
|
target = self.objs[PV_INT][0]
|
|
|
|
# Remove the PV
|
|
rc = self._pv_remove(target)
|
|
self.assertTrue(rc == '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# Add it back
|
|
rc = self._pv_create(target.Pv.Name)[0]
|
|
self.assertTrue(rc == '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def _create_raid5_thin_pool(self, vg=None):
|
|
|
|
if not vg:
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
|
|
lv_meta_path = vg.LvCreateRaid(
|
|
"meta_r5", "raid5", mib(4), 0, 0, -1, {})[0]
|
|
|
|
lv_data_path = vg.LvCreateRaid(
|
|
"data_r5", "raid5", mib(16), 0, 0, -1, {})[0]
|
|
|
|
thin_pool_path = vg.CreateThinPool(
|
|
lv_meta_path, lv_data_path, -1, {})[0]
|
|
|
|
# Get thin pool client proxy
|
|
thin_pool = ClientProxy(self.bus, thin_pool_path)
|
|
|
|
return vg, thin_pool
|
|
|
|
def test_meta_lv_data_lv_props(self):
|
|
# Ensure that metadata lv and data lv for thin pools and cache pools
|
|
# point to a valid LV
|
|
(vg, thin_pool) = self._create_raid5_thin_pool()
|
|
|
|
# Check properties on thin pool
|
|
self.assertTrue(thin_pool.ThinPool.DataLv != '/')
|
|
self.assertTrue(thin_pool.ThinPool.MetaDataLv != '/')
|
|
|
|
(vg, cache_pool) = self._create_cache_pool(vg)
|
|
|
|
self.assertTrue(cache_pool.CachePool.DataLv != '/')
|
|
self.assertTrue(cache_pool.CachePool.MetaDataLv != '/')
|
|
|
|
# Cache the thin pool
|
|
cached_thin_pool_path = cache_pool.\
|
|
CachePool.CacheLv(thin_pool.object_path, -1, {})[0]
|
|
|
|
# Get object proxy for cached thin pool
|
|
cached_thin_pool_object = ClientProxy(self.bus, cached_thin_pool_path)
|
|
|
|
# Check properties on cache pool
|
|
self.assertTrue(cached_thin_pool_object.ThinPool.DataLv != '/')
|
|
self.assertTrue(cached_thin_pool_object.ThinPool.MetaDataLv != '/')
|
|
|
|
def _lookup(self, lvm_id):
|
|
return self.objs[MANAGER_INT][0].Manager.LookUpByLvmId(lvm_id)
|
|
|
|
def test_lookup_by_lvm_id(self):
|
|
# For the moment lets just lookup what we know about which is PVs
|
|
# When we start testing VGs and LVs we will test lookups for those
|
|
# during those unit tests
|
|
for p in self.objs[PV_INT]:
|
|
rc = self._lookup(p.Pv.Name)
|
|
self.assertTrue(rc is not None and rc != '/')
|
|
|
|
# Search for something which doesn't exist
|
|
rc = self._lookup('/dev/null')
|
|
self.assertTrue(rc == '/')
|
|
|
|
def test_vg_extend(self):
|
|
# Create a VG
|
|
self.assertTrue(len(self.objs[PV_INT]) >= 2)
|
|
|
|
if len(self.objs[PV_INT]) >= 2:
|
|
pv_initial = self.objs[PV_INT][0]
|
|
pv_next = self.objs[PV_INT][1]
|
|
|
|
vg = self._vg_create([pv_initial.object_path]).Vg
|
|
path = vg.Extend([pv_next.object_path], -1, {})
|
|
self.assertTrue(path == '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# noinspection PyUnresolvedReferences
|
|
def test_vg_reduce(self):
|
|
self.assertTrue(len(self.objs[PV_INT]) >= 2)
|
|
|
|
if len(self.objs[PV_INT]) >= 2:
|
|
vg = self._vg_create(
|
|
[self.objs[PV_INT][0].object_path,
|
|
self.objs[PV_INT][1].object_path]).Vg
|
|
|
|
path = vg.Reduce(False, [vg.Pvs[0]], -1, {})
|
|
self.assertTrue(path == '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# noinspection PyUnresolvedReferences
|
|
def test_vg_rename(self):
|
|
vg = self._vg_create().Vg
|
|
|
|
mgr = self.objs[MANAGER_INT][0].Manager
|
|
|
|
# Do a vg lookup
|
|
path = mgr.LookUpByLvmId(vg.Name)
|
|
|
|
vg_name_start = vg.Name
|
|
|
|
prev_path = path
|
|
self.assertTrue(path != '/', "%s" % (path))
|
|
|
|
# Create some LVs in the VG
|
|
for i in range(0, 5):
|
|
lv_t = self._create_lv(size=mib(4), vg=vg)
|
|
full_name = "%s/%s" % (vg_name_start, lv_t.LvCommon.Name)
|
|
lv_path = mgr.LookUpByLvmId(full_name)
|
|
self.assertTrue(lv_path == lv_t.object_path)
|
|
|
|
new_name = 'renamed_' + vg.Name
|
|
|
|
path = vg.Rename(new_name, -1, {})
|
|
self.assertTrue(path == '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# Do a vg lookup
|
|
path = mgr.LookUpByLvmId(new_name)
|
|
self.assertTrue(path != '/', "%s" % (path))
|
|
self.assertTrue(prev_path == path, "%s != %s" % (prev_path, path))
|
|
|
|
# Go through each LV and make sure it has the correct path back to the
|
|
# VG
|
|
vg.update()
|
|
|
|
lv_paths = vg.Lvs
|
|
self.assertTrue(len(lv_paths) == 5)
|
|
|
|
for l in lv_paths:
|
|
lv_proxy = ClientProxy(self.bus, l).LvCommon
|
|
self.assertTrue(lv_proxy.Vg == vg.object_path, "%s != %s" %
|
|
(lv_proxy.Vg, vg.object_path))
|
|
full_name = "%s/%s" % (new_name, lv_proxy.Name)
|
|
lv_path = mgr.LookUpByLvmId(full_name)
|
|
self.assertTrue(lv_path == lv_proxy.object_path, "%s != %s" %
|
|
(lv_path, lv_proxy.object_path))
|
|
|
|
def _verify_hidden_lookups(self, lv_common_object, vgname):
|
|
mgr = self.objs[MANAGER_INT][0].Manager
|
|
|
|
hidden_lv_paths = lv_common_object.HiddenLvs
|
|
|
|
for h in hidden_lv_paths:
|
|
h_lv = ClientProxy(self.bus, h).LvCommon
|
|
|
|
if len(h_lv.HiddenLvs) > 0:
|
|
self._verify_hidden_lookups(h_lv, vgname)
|
|
|
|
# print("Hidden check %s %s" % (h, h_lv.Name))
|
|
full_name = "%s/%s" % (vgname, h_lv.Name)
|
|
lookup_path = mgr.LookUpByLvmId(full_name)
|
|
self.assertTrue(lookup_path != '/')
|
|
self.assertTrue(lookup_path == h_lv.object_path)
|
|
|
|
def test_vg_rename_with_thin_pool(self):
|
|
|
|
(vg, thin_pool) = self._create_raid5_thin_pool()
|
|
|
|
vg_name_start = vg.Name
|
|
mgr = self.objs[MANAGER_INT][0].Manager
|
|
|
|
# noinspection PyTypeChecker
|
|
self._verify_hidden_lookups(thin_pool.LvCommon, vg_name_start)
|
|
|
|
for i in range(0, 5):
|
|
lv_name = rs(8, '_lv')
|
|
|
|
thin_lv_path = thin_pool.ThinPool.LvCreate(
|
|
lv_name, mib(16), -1, {})[0]
|
|
|
|
self.assertTrue(thin_lv_path != '/')
|
|
|
|
full_name = "%s/%s" % (vg_name_start, lv_name)
|
|
|
|
lookup_lv_path = mgr.LookUpByLvmId(full_name)
|
|
self.assertTrue(thin_lv_path == lookup_lv_path,
|
|
"%s != %s" % (thin_lv_path, lookup_lv_path))
|
|
|
|
# Rename the VG
|
|
new_name = 'renamed_' + vg.Name
|
|
|
|
path = vg.Rename(new_name, -1, {})
|
|
self.assertTrue(path == '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# Go through each LV and make sure it has the correct path back to the
|
|
# VG
|
|
vg.update()
|
|
thin_pool.update()
|
|
|
|
lv_paths = vg.Lvs
|
|
|
|
for l in lv_paths:
|
|
lv_proxy = ClientProxy(self.bus, l).LvCommon
|
|
self.assertTrue(lv_proxy.Vg == vg.object_path, "%s != %s" %
|
|
(lv_proxy.Vg, vg.object_path))
|
|
full_name = "%s/%s" % (new_name, lv_proxy.Name)
|
|
# print('Full Name %s' % (full_name))
|
|
lv_path = mgr.LookUpByLvmId(full_name)
|
|
self.assertTrue(lv_path == lv_proxy.object_path, "%s != %s" %
|
|
(lv_path, lv_proxy.object_path))
|
|
|
|
# noinspection PyTypeChecker
|
|
self._verify_hidden_lookups(thin_pool.LvCommon, new_name)
|
|
|
|
def _test_lv_create(self, method, params, vg):
|
|
lv = None
|
|
path = method(*params)[0]
|
|
|
|
self.assertTrue(vg)
|
|
|
|
if path:
|
|
lv = ClientProxy(self.bus, path)
|
|
# TODO verify object properties
|
|
|
|
# We are quick enough now that we can get VolumeType changes from
|
|
# 'I' to 'i' between the time it takes to create a RAID and it returns
|
|
# and when we refresh state here. Not sure how we can handle this as
|
|
# we cannot just sit and poll all the time for changes...
|
|
# self.assertEqual(self._refresh(), 0)
|
|
return lv
|
|
|
|
def test_lv_create(self):
|
|
vg = self._vg_create().Vg
|
|
self._test_lv_create(
|
|
vg.LvCreate,
|
|
(rs(8, '_lv'), mib(4),
|
|
dbus.Array([], '(ott)'), -1, {}), vg)
|
|
|
|
def test_lv_create_linear(self):
|
|
|
|
vg = self._vg_create().Vg
|
|
self._test_lv_create(
|
|
vg.LvCreateLinear,
|
|
(rs(8, '_lv'), mib(4), False, -1, {}), vg)
|
|
|
|
def test_lv_create_striped(self):
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
self._test_lv_create(
|
|
vg.LvCreateStriped,
|
|
(rs(8, '_lv'), mib(4), 2, 8, False,
|
|
-1, {}), vg)
|
|
|
|
def test_lv_create_mirror(self):
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
self._test_lv_create(vg.LvCreateMirror,
|
|
(rs(8, '_lv'), mib(4), 2, -1, {}), vg)
|
|
|
|
def test_lv_create_raid(self):
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
self._test_lv_create(vg.LvCreateRaid,
|
|
(rs(8, '_lv'), 'raid4',
|
|
mib(16), 2, 8, -1, {}), vg)
|
|
|
|
def _create_lv(self, thinpool=False, size=None, vg=None):
|
|
|
|
if not vg:
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
|
|
if size is None:
|
|
size = mib(4)
|
|
|
|
return self._test_lv_create(
|
|
vg.LvCreateLinear,
|
|
(rs(8, '_lv'), size, thinpool, -1, {}), vg)
|
|
|
|
def test_lv_create_rounding(self):
|
|
self._create_lv(size=(mib(2) + 13))
|
|
|
|
def test_lv_create_thin_pool(self):
|
|
self._create_lv(True)
|
|
|
|
def test_lv_rename(self):
|
|
# Rename a regular LV
|
|
lv = self._create_lv()
|
|
|
|
path = self.objs[MANAGER_INT][0].Manager.LookUpByLvmId(lv.LvCommon.Name)
|
|
prev_path = path
|
|
|
|
new_name = 'renamed_' + lv.LvCommon.Name
|
|
lv.Lv.Rename(new_name, -1, {})
|
|
|
|
path = self.objs[MANAGER_INT][0].Manager.LookUpByLvmId(new_name)
|
|
|
|
self.assertEqual(self._refresh(), 0)
|
|
self.assertTrue(prev_path == path, "%s != %s" % (prev_path, path))
|
|
|
|
def test_lv_thinpool_rename(self):
|
|
# Rename a thin pool
|
|
tp = self._create_lv(True)
|
|
self.assertTrue(THINPOOL_LV_PATH in tp.object_path,
|
|
"%s" % (tp.object_path))
|
|
|
|
new_name = 'renamed_' + tp.LvCommon.Name
|
|
tp.Lv.Rename(new_name, -1, {})
|
|
tp.update()
|
|
self.assertEqual(self._refresh(), 0)
|
|
self.assertEqual(new_name, tp.LvCommon.Name)
|
|
|
|
# noinspection PyUnresolvedReferences
|
|
def test_lv_on_thin_pool_rename(self):
|
|
# Rename a LV on a thin Pool
|
|
|
|
# This returns a LV with the LV interface, need to get a proxy for
|
|
# thinpool interface too
|
|
tp = self._create_lv(True)
|
|
|
|
thin_path = tp.ThinPool.LvCreate(
|
|
rs(10, '_thin_lv'), mib(8), -1, {})[0]
|
|
|
|
lv = ClientProxy(self.bus, thin_path)
|
|
rc = lv.Lv.Rename('rename_test' + lv.LvCommon.Name, -1, {})
|
|
self.assertTrue(rc == '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def test_lv_remove(self):
|
|
lv = self._create_lv().Lv
|
|
rc = lv.Remove(-1, {})
|
|
self.assertTrue(rc == '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def test_lv_snapshot(self):
|
|
lv_p = self._create_lv()
|
|
ss_name = 'ss_' + lv_p.LvCommon.Name
|
|
|
|
# Test waiting to complete
|
|
ss, job = lv_p.Lv.Snapshot(ss_name, 0, -1, {})
|
|
self.assertTrue(ss != '/')
|
|
self.assertTrue(job == '/')
|
|
|
|
snapshot = ClientProxy(self.bus, ss)
|
|
self.assertTrue(snapshot.LvCommon.Name == ss_name)
|
|
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# Test getting a job returned immediately
|
|
rc, job = lv_p.Lv.Snapshot('ss2_' + lv_p.LvCommon.Name, 0, 0, {})
|
|
self.assertTrue(rc == '/')
|
|
self.assertTrue(job != '/')
|
|
self._wait_for_job(job)
|
|
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# noinspection PyUnresolvedReferences
|
|
def _wait_for_job(self, j_path):
|
|
import time
|
|
rc = None
|
|
j = ClientProxy(self.bus, j_path).Job
|
|
|
|
while True:
|
|
j.update()
|
|
if j.Complete:
|
|
(ec, error_msg) = j.GetError
|
|
self.assertTrue(ec == 0, "%d :%s" % (ec, error_msg))
|
|
|
|
if ec == 0:
|
|
self.assertTrue(j.Percent == 100, "P= %f" % j.Percent)
|
|
|
|
rc = j.Result
|
|
j.Remove()
|
|
|
|
break
|
|
|
|
if j.Wait(1):
|
|
j.update()
|
|
self.assertTrue(j.Complete)
|
|
|
|
return rc
|
|
|
|
def test_lv_create_pv_specific(self):
|
|
vg = self._vg_create().Vg
|
|
|
|
pv = vg.Pvs
|
|
|
|
pv_proxy = ClientProxy(self.bus, pv[0])
|
|
|
|
self._test_lv_create(vg.LvCreate, (rs(8, '_lv'), mib(4),
|
|
dbus.Array([[pv_proxy.object_path, 0, (pv_proxy.Pv.PeCount - 1)]],
|
|
'(ott)'), -1, {}), vg)
|
|
|
|
def test_lv_resize(self):
|
|
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
lv = self._create_lv(vg=vg, size=mib(16))
|
|
|
|
for size in \
|
|
[
|
|
lv.LvCommon.SizeBytes + 4194304,
|
|
lv.LvCommon.SizeBytes - 4194304,
|
|
lv.LvCommon.SizeBytes + 2048,
|
|
lv.LvCommon.SizeBytes - 2048]:
|
|
|
|
pv_in_use = [i[0] for i in lv.LvCommon.Devices]
|
|
# Select a PV in the VG that isn't in use
|
|
pv_empty = [p for p in vg.Pvs if p not in pv_in_use]
|
|
|
|
prev = lv.LvCommon.SizeBytes
|
|
|
|
if len(pv_empty):
|
|
p = ClientProxy(self.bus, pv_empty[0])
|
|
rc = lv.Lv.Resize(
|
|
size,
|
|
dbus.Array([[p.object_path, 0, p.Pv.PeCount - 1]], '(oii)'),
|
|
-1, {})
|
|
else:
|
|
rc = lv.Lv.Resize(size, dbus.Array([], '(oii)'), -1, {})
|
|
|
|
self.assertEqual(rc, '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
lv.update()
|
|
|
|
if prev < size:
|
|
self.assertTrue(lv.LvCommon.SizeBytes > prev)
|
|
else:
|
|
# We are testing re-sizing to same size too...
|
|
self.assertTrue(lv.LvCommon.SizeBytes <= prev)
|
|
|
|
def test_lv_resize_same(self):
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
lv = self._create_lv(vg=vg)
|
|
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
lv.Lv.Resize(lv.LvCommon.SizeBytes, dbus.Array([], '(oii)'), -1, {})
|
|
|
|
def test_lv_move(self):
|
|
lv = self._create_lv()
|
|
|
|
pv_path_move = str(lv.LvCommon.Devices[0][0])
|
|
|
|
# Test moving a specific LV
|
|
job = lv.Lv.Move(pv_path_move, (0, 0), dbus.Array([], '(oii)'), 0, {})
|
|
self._wait_for_job(job)
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
lv.update()
|
|
new_pv = str(lv.LvCommon.Devices[0][0])
|
|
self.assertTrue(pv_path_move != new_pv, "%s == %s" %
|
|
(pv_path_move, new_pv))
|
|
|
|
def test_lv_activate_deactivate(self):
|
|
lv_p = self._create_lv()
|
|
lv_p.update()
|
|
|
|
lv_p.Lv.Deactivate(0, -1, {})
|
|
lv_p.update()
|
|
self.assertFalse(lv_p.LvCommon.Active)
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
lv_p.Lv.Activate(0, -1, {})
|
|
|
|
lv_p.update()
|
|
self.assertTrue(lv_p.LvCommon.Active)
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# Try control flags
|
|
for i in range(0, 5):
|
|
lv_p.Lv.Activate(1 << i, -1, {})
|
|
self.assertTrue(lv_p.LvCommon.Active)
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def test_move(self):
|
|
lv = self._create_lv()
|
|
|
|
# Test moving without being LV specific
|
|
vg = ClientProxy(self.bus, lv.LvCommon.Vg).Vg
|
|
pv_to_move = str(lv.LvCommon.Devices[0][0])
|
|
job = vg.Move(pv_to_move, (0, 0), dbus.Array([], '(oii)'), 0, {})
|
|
self._wait_for_job(job)
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# Test Vg.Move
|
|
# TODO Test this more!
|
|
vg.update()
|
|
lv.update()
|
|
|
|
location = lv.LvCommon.Devices[0][0]
|
|
|
|
dst = None
|
|
for p in vg.Pvs:
|
|
if p != location:
|
|
dst = p
|
|
|
|
# Fetch the destination
|
|
pv = ClientProxy(self.bus, dst).Pv
|
|
|
|
# Test range, move it to the middle of the new destination and blocking
|
|
# blocking for it to complete
|
|
job = vg.Move(
|
|
location, (0, 0), [(dst, pv.PeCount / 2, 0), ], -1, {})
|
|
self.assertEqual(job, '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def test_job_handling(self):
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg_name = rs(8, '_vg')
|
|
|
|
# Test getting a job right away
|
|
vg_path, vg_job = self.objs[MANAGER_INT][0].Manager.VgCreate(
|
|
vg_name, pv_paths,
|
|
0, {})
|
|
|
|
self.assertTrue(vg_path == '/')
|
|
self.assertTrue(vg_job and len(vg_job) > 0)
|
|
|
|
self._wait_for_job(vg_job)
|
|
|
|
def _test_expired_timer(self, num_lvs):
|
|
rc = False
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
# In small configurations lvm is pretty snappy, so lets create a VG
|
|
# add a number of LVs and then remove the VG and all the contained
|
|
# LVs which appears to consistently run a little slow.
|
|
|
|
vg_proxy = self._vg_create(pv_paths)
|
|
|
|
for i in range(0, num_lvs):
|
|
|
|
vg_proxy.update()
|
|
if vg_proxy.Vg.FreeCount > 0:
|
|
obj_path, job = vg_proxy.Vg.LvCreateLinear(
|
|
rs(8, "_lv"), mib(4), False, -1, {})
|
|
self.assertTrue(job == '/')
|
|
else:
|
|
# We ran out of space, test will probably fail
|
|
break
|
|
|
|
# Make sure that we are honoring the timeout
|
|
start = time.time()
|
|
|
|
remove_job = vg_proxy.Vg.Remove(1, {})
|
|
|
|
end = time.time()
|
|
|
|
tt_remove = float(end) - float(start)
|
|
|
|
self.assertTrue(tt_remove < 2.0, "remove time %s" % (str(tt_remove)))
|
|
|
|
# Depending on how long it took we could finish either way
|
|
if remove_job != '/':
|
|
# We got a job
|
|
result = self._wait_for_job(remove_job)
|
|
self.assertTrue(result == '/')
|
|
rc = True
|
|
else:
|
|
# It completed before timer popped
|
|
pass
|
|
|
|
return rc
|
|
|
|
def test_job_handling_timer(self):
|
|
|
|
yes = False
|
|
|
|
# This may not pass
|
|
for i in [48, 64, 128]:
|
|
yes = self._test_expired_timer(i)
|
|
if yes:
|
|
break
|
|
print('Attempt (%d) failed, trying again...' % (i))
|
|
|
|
self.assertTrue(yes)
|
|
|
|
def test_pv_tags(self):
|
|
pvs = []
|
|
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
|
|
# Get the PVs
|
|
for p in vg.Pvs:
|
|
pvs.append(ClientProxy(self.bus, p).Pv)
|
|
|
|
for tags_value in [['hello'], ['foo', 'bar']]:
|
|
rc = vg.PvTagsAdd(vg.Pvs, tags_value, -1, {})
|
|
self.assertTrue(rc == '/')
|
|
|
|
for p in pvs:
|
|
p.update()
|
|
self.assertTrue(sorted(tags_value) == p.Tags)
|
|
|
|
vg.PvTagsDel(vg.Pvs, tags_value, -1, {})
|
|
for p in pvs:
|
|
p.update()
|
|
self.assertTrue([] == p.Tags)
|
|
|
|
def test_vg_tags(self):
|
|
vg = self._vg_create().Vg
|
|
|
|
t = ['Testing', 'tags']
|
|
|
|
vg.TagsAdd(t, -1, {})
|
|
vg.update()
|
|
self.assertTrue(t == vg.Tags)
|
|
vg.TagsDel(t, -1, {})
|
|
vg.update()
|
|
self.assertTrue([] == vg.Tags)
|
|
|
|
def test_lv_tags(self):
|
|
vg = self._vg_create().Vg
|
|
lv = self._test_lv_create(
|
|
vg.LvCreateLinear,
|
|
(rs(8, '_lv'), mib(4), False, -1, {}),
|
|
vg)
|
|
|
|
t = ['Testing', 'tags']
|
|
|
|
lv.Lv.TagsAdd(t, -1, {})
|
|
lv.update()
|
|
self.assertTrue(t == lv.LvCommon.Tags)
|
|
lv.Lv.TagsDel(t, -1, {})
|
|
lv.update()
|
|
self.assertTrue([] == lv.LvCommon.Tags)
|
|
|
|
def test_vg_allocation_policy_set(self):
|
|
vg = self._vg_create().Vg
|
|
|
|
for p in ['anywhere', 'contiguous', 'cling', 'normal']:
|
|
rc = vg.AllocationPolicySet(p, -1, {})
|
|
self.assertEqual(rc, '/')
|
|
vg.update()
|
|
|
|
prop = getattr(vg, 'Alloc' + p.title())
|
|
self.assertTrue(prop)
|
|
|
|
def test_vg_max_pv(self):
|
|
vg = self._vg_create().Vg
|
|
|
|
# BZ: https://bugzilla.redhat.com/show_bug.cgi?id=1280496
|
|
# TODO: Add a test back for larger values here when bug is resolved
|
|
for p in [0, 1, 10, 100, 100, 1024, 2 ** 32 - 1]:
|
|
rc = vg.MaxPvSet(p, -1, {})
|
|
self.assertEqual(rc, '/')
|
|
vg.update()
|
|
self.assertTrue(vg.MaxPv == p, "Expected %s != Actual %s" %
|
|
(str(p), str(vg.MaxPv)))
|
|
|
|
def test_vg_max_lv(self):
|
|
vg = self._vg_create().Vg
|
|
|
|
# BZ: https://bugzilla.redhat.com/show_bug.cgi?id=1280496
|
|
# TODO: Add a test back for larger values here when bug is resolved
|
|
for p in [0, 1, 10, 100, 100, 1024, 2 ** 32 - 1]:
|
|
rc = vg.MaxLvSet(p, -1, {})
|
|
self.assertEqual(rc, '/')
|
|
vg.update()
|
|
self.assertTrue(vg.MaxLv == p, "Expected %s != Actual %s" %
|
|
(str(p), str(vg.MaxLv)))
|
|
|
|
def test_vg_uuid_gen(self):
|
|
# TODO renable test case when
|
|
# https://bugzilla.redhat.com/show_bug.cgi?id=1264169 gets fixed
|
|
# This was tested with lvmetad disabled and we passed
|
|
print("\nSkipping Vg.UuidGenerate until BZ: 1264169 resolved\n")
|
|
|
|
if False:
|
|
vg = self._vg_create().Vg
|
|
prev_uuid = vg.Uuid
|
|
rc = vg.UuidGenerate(-1, {})
|
|
self.assertEqual(rc, '/')
|
|
vg.update()
|
|
self.assertTrue(vg.Uuid != prev_uuid, "Expected %s != Actual %s" %
|
|
(vg.Uuid, prev_uuid))
|
|
|
|
def test_vg_activate_deactivate(self):
|
|
vg = self._vg_create().Vg
|
|
self._test_lv_create(
|
|
vg.LvCreateLinear,
|
|
(rs(8, '_lv'), mib(4), False, -1, {}),
|
|
vg)
|
|
|
|
vg.update()
|
|
|
|
vg.Deactivate(0, -1, {})
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
vg.Activate(0, -1, {})
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
# Try control flags
|
|
for i in range(0, 5):
|
|
vg.Activate(1 << i, -1, {})
|
|
|
|
def test_pv_resize(self):
|
|
|
|
self.assertTrue(len(self.objs[PV_INT]) > 0)
|
|
|
|
if len(self.objs[PV_INT]) > 0:
|
|
pv = ClientProxy(self.bus, self.objs[PV_INT][0].object_path).Pv
|
|
|
|
original_size = pv.SizeBytes
|
|
|
|
new_size = original_size / 2
|
|
|
|
pv.ReSize(new_size, -1, {})
|
|
self.assertEqual(self._refresh(), 0)
|
|
pv.update()
|
|
|
|
self.assertTrue(pv.SizeBytes != original_size)
|
|
pv.ReSize(0, -1, {})
|
|
self.assertEqual(self._refresh(), 0)
|
|
pv.update()
|
|
self.assertTrue(pv.SizeBytes == original_size)
|
|
|
|
def test_pv_allocation(self):
|
|
|
|
pv_paths = []
|
|
for pp in self.objs[PV_INT]:
|
|
pv_paths.append(pp.object_path)
|
|
|
|
vg = self._vg_create(pv_paths).Vg
|
|
|
|
pv = ClientProxy(self.bus, vg.Pvs[0]).Pv
|
|
|
|
pv.AllocationEnabled(False, -1, {})
|
|
pv.update()
|
|
self.assertFalse(pv.Allocatable)
|
|
|
|
pv.AllocationEnabled(True, -1, {})
|
|
pv.update()
|
|
self.assertTrue(pv.Allocatable)
|
|
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
def _get_devices(self):
|
|
context = pyudev.Context()
|
|
return context.list_devices(subsystem='block', MAJOR='8')
|
|
|
|
def test_pv_scan(self):
|
|
devices = self._get_devices()
|
|
|
|
mgr = self._manager().Manager
|
|
|
|
self.assertEqual(
|
|
mgr.PvScan(
|
|
False, True, dbus.Array([], 's'),
|
|
dbus.Array([], '(ii)'), -1, {}), '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
self.assertEqual(
|
|
mgr.PvScan(
|
|
False, False,
|
|
dbus.Array([], 's'),
|
|
dbus.Array([], '(ii)'), -1, {}), '/')
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
block_path = []
|
|
for d in devices:
|
|
block_path.append(d['DEVNAME'])
|
|
|
|
self.assertEqual(mgr.PvScan(False, True,
|
|
block_path,
|
|
dbus.Array([], '(ii)'), -1, {}), '/')
|
|
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
mm = []
|
|
for d in devices:
|
|
mm.append((int(d['MAJOR']), int(d['MINOR'])))
|
|
|
|
self.assertEqual(mgr.PvScan(False, True,
|
|
block_path,
|
|
mm, -1, {}), '/')
|
|
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
self.assertEqual(
|
|
mgr.PvScan(
|
|
False, True,
|
|
dbus.Array([], 's'),
|
|
mm, -1, {}), '/')
|
|
|
|
self.assertEqual(self._refresh(), 0)
|
|
|
|
@staticmethod
|
|
def _write_some_data(device_path, size):
|
|
blocks = int(size / 512)
|
|
block = bytearray(512)
|
|
for i in range(0, 512):
|
|
block[i] = i % 255
|
|
|
|
with open(device_path, mode='wb') as lv:
|
|
for i in range(0, blocks):
|
|
lv.write(block)
|
|
|
|
def test_snapshot_merge(self):
|
|
# Create a non-thin LV and merge it
|
|
ss_size = mib(8)
|
|
|
|
lv_p = self._create_lv(size=mib(16))
|
|
ss_name = lv_p.LvCommon.Name + '_snap'
|
|
snapshot_path = lv_p.Lv.Snapshot(ss_name, ss_size, -1, {})[0]
|
|
ss = ClientProxy(self.bus, snapshot_path)
|
|
|
|
# Write some data to snapshot so merge takes some time
|
|
TestDbusService._write_some_data(ss.LvCommon.Path, ss_size / 2)
|
|
|
|
job_path = ss.Snapshot.Merge(0, {})
|
|
|
|
self.assertTrue(job_path != '/')
|
|
self._wait_for_job(job_path)
|
|
|
|
def test_snapshot_merge_thin(self):
|
|
# Create a thin LV, snapshot it and merge it
|
|
tp = self._create_lv(True)
|
|
|
|
thin_path = tp.ThinPool.LvCreate(
|
|
rs(10, '_thin_lv'), mib(10), -1, {})[0]
|
|
|
|
lv_p = ClientProxy(self.bus, thin_path)
|
|
|
|
ss_name = lv_p.LvCommon.Name + '_snap'
|
|
snapshot_path = lv_p.Lv.Snapshot(ss_name, 0, -1, {})[0]
|
|
ss = ClientProxy(self.bus, snapshot_path)
|
|
job_path = ss.Snapshot.Merge(0, {})
|
|
self.assertTrue(job_path != '/')
|
|
self._wait_for_job(job_path)
|
|
|
|
def _create_cache_pool(self, vg=None):
|
|
|
|
if not vg:
|
|
vg = self._vg_create().Vg
|
|
|
|
md = self._create_lv(size=(mib(8)), vg=vg)
|
|
data = self._create_lv(size=(mib(8)), vg=vg)
|
|
|
|
cache_pool_path = vg.CreateCachePool(
|
|
md.object_path, data.object_path, -1, {})[0]
|
|
|
|
cp = ClientProxy(self.bus, cache_pool_path)
|
|
|
|
return vg, cp
|
|
|
|
def test_cache_pool_create(self):
|
|
|
|
vg, cache_pool = self._create_cache_pool()
|
|
|
|
self.assertTrue('/com/redhat/lvmdbus1/CachePool' in
|
|
cache_pool.object_path)
|
|
|
|
def test_cache_lv_create(self):
|
|
|
|
for destroy_cache in [True, False]:
|
|
vg, cache_pool = self._create_cache_pool()
|
|
|
|
lv_to_cache = self._create_lv(size=mib(8), vg=vg)
|
|
|
|
c_lv_path = cache_pool.CachePool.CacheLv(
|
|
lv_to_cache.object_path, -1, {})[0]
|
|
|
|
cached_lv = ClientProxy(self.bus, c_lv_path)
|
|
|
|
uncached_lv_path = \
|
|
cached_lv.CachedLv.DetachCachePool(destroy_cache, -1, {})[0]
|
|
|
|
self.assertTrue('/com/redhat/lvmdbus1/Lv' in
|
|
uncached_lv_path)
|
|
|
|
vg.Remove(-1, {})
|
|
|
|
def test_vg_change(self):
|
|
vg_proxy = self._vg_create()
|
|
result = vg_proxy.Vg.Change(-1, {'-a': 'ay'})
|
|
self.assertTrue(result == '/')
|
|
result = vg_proxy.Vg.Change(-1, {'-a': 'n'})
|
|
self.assertTrue(result == '/')
|
|
|
|
def _invalid_vg_lv_name_characters(self):
|
|
bad_vg_lv_set = set(string.printable) - \
|
|
set(string.ascii_letters + string.digits + '.-_+')
|
|
return ''.join(bad_vg_lv_set)
|
|
|
|
def test_invalid_names(self):
|
|
mgr = self.objs[MANAGER_INT][0].Manager
|
|
|
|
# Pv device path
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
mgr.PvCreate("/dev/space in name", -1, {})
|
|
|
|
# VG Name testing...
|
|
# Go through all bad characters
|
|
pv_paths = [self.objs[PV_INT][0].object_path]
|
|
bad_chars = self._invalid_vg_lv_name_characters()
|
|
for c in bad_chars:
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
mgr.VgCreate("name%s" % (c), pv_paths, -1, {})
|
|
|
|
# Bad names
|
|
for bad in [".", ".."]:
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
mgr.VgCreate(bad, pv_paths, -1, {})
|
|
|
|
# Exceed name length
|
|
for i in [128, 1024, 4096]:
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
mgr.VgCreate('T' * i, pv_paths, -1, {})
|
|
|
|
# Create a VG and try to create LVs with different bad names
|
|
vg_path = mgr.VgCreate("test", pv_paths, -1, {})[0]
|
|
vg_proxy = ClientProxy(self.bus, vg_path)
|
|
|
|
for c in bad_chars:
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
vg_proxy.Vg.LvCreateLinear(rs(8, '_lv') + c,
|
|
mib(4), False, -1, {})
|
|
|
|
for r in ("_cdata", "_cmeta", "_corig", "_mimage", "_mlog",
|
|
"_pmspare", "_rimage", "_rmeta", "_tdata", "_tmeta", "_vorigin"):
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
vg_proxy.Vg.LvCreateLinear(rs(8, '_lv') + r,
|
|
mib(4), False, -1, {})
|
|
|
|
for r in ("snapshot", "pvmove"):
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
vg_proxy.Vg.LvCreateLinear(r + rs(8, '_lv'),
|
|
mib(4), False, -1, {})
|
|
|
|
_ALLOWABLE_TAG_CH = string.ascii_letters + string.digits + "._-+/=!:&#"
|
|
|
|
def _invalid_tag_characters(self):
|
|
bad_tag_ch_set = set(string.printable) - set(self._ALLOWABLE_TAG_CH)
|
|
return ''.join(bad_tag_ch_set)
|
|
|
|
def test_invalid_tags(self):
|
|
mgr = self.objs[MANAGER_INT][0].Manager
|
|
pv_paths = [self.objs[PV_INT][0].object_path]
|
|
|
|
vg_path = mgr.VgCreate("test", pv_paths, -1, {})[0]
|
|
vg_proxy = ClientProxy(self.bus, vg_path)
|
|
|
|
for c in self._invalid_tag_characters():
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
vg_proxy.Vg.TagsAdd([c], -1, {})
|
|
|
|
for c in self._invalid_tag_characters():
|
|
with self.assertRaises(dbus.exceptions.DBusException):
|
|
vg_proxy.Vg.TagsAdd(["a%sb" % (c)], -1, {})
|
|
|
|
def test_tag_names(self):
|
|
mgr = self.objs[MANAGER_INT][0].Manager
|
|
pv_paths = [self.objs[PV_INT][0].object_path]
|
|
|
|
vg_path = mgr.VgCreate("test", pv_paths, -1, {})[0]
|
|
vg_proxy = ClientProxy(self.bus, vg_path)
|
|
|
|
for i in range(1, 64):
|
|
tag = rs(i, "", self._ALLOWABLE_TAG_CH)
|
|
r = vg_proxy.Vg.TagsAdd([tag], -1, {})
|
|
self.assertTrue(r == '/')
|
|
vg_proxy.update()
|
|
|
|
self.assertTrue(tag in vg_proxy.Vg.Tags, "%s not in %s" %
|
|
(tag, str(vg_proxy.Vg.Tags)))
|
|
|
|
self.assertEqual(i, len(vg_proxy.Vg.Tags), "%d != %d" %
|
|
(i, len(vg_proxy.Vg.Tags)))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# Test forking & exec new each time
|
|
test_shell = os.getenv('LVM_DBUS_TEST_SHELL', 0)
|
|
|
|
set_execution(False)
|
|
|
|
if test_shell == 0:
|
|
unittest.main(exit=True)
|
|
else:
|
|
unittest.main(exit=False)
|
|
# Test lvm shell
|
|
print('\n *** Testing lvm shell *** \n')
|
|
set_execution(True)
|
|
unittest.main()
|