diff options
author | Tony Asleson <tasleson@redhat.com> | 2016-02-25 16:11:08 -0600 |
---|---|---|
committer | Tony Asleson <tasleson@redhat.com> | 2016-02-25 16:43:55 -0600 |
commit | 0d620e681fe42d0a1b2a5666677c36753838c6fd (patch) | |
tree | ac5f821b0f6f5768c5c8129fb41909fff8cd609f | |
parent | f1bc68beb4e493a9c1a52750e99d50f874ca89d9 (diff) | |
download | lvm2-0d620e681fe42d0a1b2a5666677c36753838c6fd.tar.gz |
lvmdbustest.py: Move helper code to testlib.py
This will allow us to re-use in other client programs for testing.
Signed-off-by: Tony Asleson <tasleson@redhat.com>
-rwxr-xr-x | test/dbus/lvmdbustest.py | 199 | ||||
-rw-r--r-- | test/dbus/testlib.py | 174 |
2 files changed, 195 insertions, 178 deletions
diff --git a/test/dbus/lvmdbustest.py b/test/dbus/lvmdbustest.py index 92f5d70ac..4c58552d2 100755 --- a/test/dbus/lvmdbustest.py +++ b/test/dbus/lvmdbustest.py @@ -14,35 +14,10 @@ import dbus from dbus.mainloop.glib import DBusGMainLoop import unittest import sys -import random -import string -import functools import time import pyudev import os -import xml.etree.ElementTree as Et -from collections import OrderedDict - -BUSNAME = "com.redhat.lvmdbus1" -MANAGER_INT = BUSNAME + '.Manager' -MANAGER_OBJ = '/' + BUSNAME.replace('.', '/') + 'Manager' -PV_INT = BUSNAME + ".Pv" -VG_INT = BUSNAME + ".Vg" -LV_INT = BUSNAME + ".Lv" -THINPOOL_INT = BUSNAME + ".ThinPool" -SNAPSHOT_INT = BUSNAME + ".Snapshot" -LV_COMMON_INT = BUSNAME + ".LvCommon" -JOB_INT = BUSNAME + ".Job" -CACHE_POOL_INT = BUSNAME + ".CachePool" -CACHE_LV_INT = BUSNAME + ".CachedLv" - -THINPOOL_LV_PATH = '/' + THINPOOL_INT.replace('.', '/') - - -def rs(length, suffix, character_set=string.ascii_lowercase): - return ''.join(random.choice(character_set) - for _ in range(length)) + suffix - +from testlib import * use_session = os.getenv('LVMDBUSD_USE_SESSION', False) @@ -52,144 +27,6 @@ else: bus = dbus.SystemBus(mainloop=DBusGMainLoop()) -def mib(s): - return 1024 * 1024 * s - - -class DbusIntrospection(object): - @staticmethod - def introspect(xml_representation): - interfaces = {} - - root = Et.fromstring(xml_representation) - - for c in root: - if c.tag == "interface": - in_f = c.attrib['name'] - interfaces[in_f] = \ - dict(methods=OrderedDict(), properties={}) - for nested in c: - if nested.tag == "method": - mn = nested.attrib['name'] - interfaces[in_f]['methods'][mn] = OrderedDict() - - for arg in nested: - if arg.tag == 'arg': - arg_dir = arg.attrib['direction'] - if arg_dir == 'in': - n = arg.attrib['name'] - else: - n = None - - arg_type = arg.attrib['type'] - - if n: - v = dict( - name=mn, - a_dir=arg_dir, - a_type=arg_type) - interfaces[in_f]['methods'][mn][n] = v - - elif nested.tag == 'property': - pn = nested.attrib['name'] - p_access = nested.attrib['access'] - p_type = nested.attrib['type'] - - interfaces[in_f]['properties'][pn] = \ - dict(p_access=p_access, p_type=p_type) - else: - pass - - # print('Interfaces...') - # for k, v in list(interfaces.items()): - # print('Interface %s' % k) - # if v['methods']: - # for m, args in list(v['methods'].items()): - # print(' method: %s' % m) - # for a, aa in args.items(): - # print(' method arg: %s' % (a)) - # if v['properties']: - # for p, d in list(v['properties'].items()): - # print(' Property: %s' % (p)) - # print('End interfaces') - - return interfaces - - -class RemoteObject(object): - def _set_props(self, props=None): - # print 'Fetching properties' - if not props: - # prop_fetch = dbus.Interface(self.bus.get_object( - # BUSNAME, self.object_path), 'org.freedesktop.DBus.Properties') - - for i in range(0, 3): - try: - prop_fetch = dbus.Interface(self.bus.get_object( - BUSNAME, self.object_path), - 'org.freedesktop.DBus.Properties') - props = prop_fetch.GetAll(self.interface) - break - except dbus.exceptions.DBusException as dbe: - if "GetAll" not in str(dbe): - raise dbe - if props: - for kl, vl in list(props.items()): - setattr(self, kl, vl) - - def __init__(self, specified_bus, object_path, interface, properties=None): - self.object_path = object_path - self.interface = interface - self.bus = specified_bus - - self.dbus_method = dbus.Interface(specified_bus.get_object( - BUSNAME, self.object_path), self.interface) - - self._set_props(properties) - - def __getattr__(self, item): - if hasattr(self.dbus_method, item): - return functools.partial(self._wrapper, item) - else: - return functools.partial(self, item) - - def _wrapper(self, _method_name, *args, **kwargs): - return getattr(self.dbus_method, _method_name)(*args, **kwargs) - - def update(self): - self._set_props() - - -class ClientProxy(object): - @staticmethod - def _intf_short_name(nm): - return nm.split('.')[-1:][0] - - def __init__(self, specified_bus, object_path, interface=None, props=None): - i = dbus.Interface(specified_bus.get_object( - BUSNAME, object_path), 'org.freedesktop.DBus.Introspectable') - self.intro_spect = DbusIntrospection.introspect(i.Introspect()) - - for k in self.intro_spect.keys(): - sn = ClientProxy._intf_short_name(k) - # print('Client proxy has interface: %s %s' % (k, sn)) - - if interface and interface == k and props is not None: - ro = RemoteObject(specified_bus, object_path, k, props) - else: - ro = RemoteObject(specified_bus, object_path, k) - - setattr(self, sn, ro) - - self.object_path = object_path - - def update(self): - # Go through all interfaces and update them - for int_f in self.intro_spect.keys(): - sn = ClientProxy._intf_short_name(int_f) - getattr(self, sn).update() - - def get_objects(): rc = {MANAGER_INT: [], PV_INT: [], VG_INT: [], LV_INT: [], THINPOOL_INT: [], JOB_INT: [], SNAPSHOT_INT: [], LV_COMMON_INT: [], @@ -692,11 +529,12 @@ class TestDbusService(unittest.TestCase): 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]: + 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 @@ -1052,13 +890,16 @@ class TestDbusService(unittest.TestCase): mgr = self._manager().Manager - self.assertEqual(mgr.PvScan(False, True, - dbus.Array([], 's'), - dbus.Array([], '(ii)'), -1, {}), '/') + 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( + mgr.PvScan( + False, False, + dbus.Array([], 's'), + dbus.Array([], '(ii)'), -1, {}), '/') self.assertEqual(self._refresh(), 0) block_path = [] @@ -1081,9 +922,11 @@ class TestDbusService(unittest.TestCase): self.assertEqual(self._refresh(), 0) - self.assertEqual(mgr.PvScan(False, True, - dbus.Array([], 's'), - mm, -1, {}), '/') + self.assertEqual( + mgr.PvScan( + False, True, + dbus.Array([], 's'), + mm, -1, {}), '/') self.assertEqual(self._refresh(), 0) diff --git a/test/dbus/testlib.py b/test/dbus/testlib.py new file mode 100644 index 000000000..550b0bf67 --- /dev/null +++ b/test/dbus/testlib.py @@ -0,0 +1,174 @@ +#!/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 string +import random +import functools +import xml.etree.ElementTree as Et +from collections import OrderedDict +import dbus + +BUSNAME = "com.redhat.lvmdbus1" +MANAGER_INT = BUSNAME + '.Manager' +MANAGER_OBJ = '/' + BUSNAME.replace('.', '/') + '/Manager' +PV_INT = BUSNAME + ".Pv" +VG_INT = BUSNAME + ".Vg" +LV_INT = BUSNAME + ".Lv" +THINPOOL_INT = BUSNAME + ".ThinPool" +SNAPSHOT_INT = BUSNAME + ".Snapshot" +LV_COMMON_INT = BUSNAME + ".LvCommon" +JOB_INT = BUSNAME + ".Job" +CACHE_POOL_INT = BUSNAME + ".CachePool" +CACHE_LV_INT = BUSNAME + ".CachedLv" + +THINPOOL_LV_PATH = '/' + THINPOOL_INT.replace('.', '/') + + +def rs(length, suffix, character_set=string.ascii_lowercase): + return ''.join(random.choice(character_set) + for _ in range(length)) + suffix + + +def mib(s): + return 1024 * 1024 * s + + +class DbusIntrospection(object): + @staticmethod + def introspect(xml_representation): + interfaces = {} + + root = Et.fromstring(xml_representation) + + for c in root: + if c.tag == "interface": + in_f = c.attrib['name'] + interfaces[in_f] = \ + dict(methods=OrderedDict(), properties={}) + for nested in c: + if nested.tag == "method": + mn = nested.attrib['name'] + interfaces[in_f]['methods'][mn] = OrderedDict() + + for arg in nested: + if arg.tag == 'arg': + arg_dir = arg.attrib['direction'] + if arg_dir == 'in': + n = arg.attrib['name'] + else: + n = None + + arg_type = arg.attrib['type'] + + if n: + v = dict( + name=mn, + a_dir=arg_dir, + a_type=arg_type) + interfaces[in_f]['methods'][mn][n] = v + + elif nested.tag == 'property': + pn = nested.attrib['name'] + p_access = nested.attrib['access'] + p_type = nested.attrib['type'] + + interfaces[in_f]['properties'][pn] = \ + dict(p_access=p_access, p_type=p_type) + else: + pass + + # print('Interfaces...') + # for k, v in list(interfaces.items()): + # print('Interface %s' % k) + # if v['methods']: + # for m, args in list(v['methods'].items()): + # print(' method: %s' % m) + # for a, aa in args.items(): + # print(' method arg: %s' % (a)) + # if v['properties']: + # for p, d in list(v['properties'].items()): + # print(' Property: %s' % (p)) + # print('End interfaces') + + return interfaces + + +class RemoteObject(object): + def _set_props(self, props=None): + # print 'Fetching properties' + if not props: + # prop_fetch = dbus.Interface(self.bus.get_object( + # BUSNAME, self.object_path), 'org.freedesktop.DBus.Properties') + + for i in range(0, 3): + try: + prop_fetch = dbus.Interface(self.bus.get_object( + BUSNAME, self.object_path), + 'org.freedesktop.DBus.Properties') + props = prop_fetch.GetAll(self.interface) + break + except dbus.exceptions.DBusException as dbe: + if "GetAll" not in str(dbe): + raise dbe + if props: + for kl, vl in list(props.items()): + setattr(self, kl, vl) + + def __init__(self, specified_bus, object_path, interface, properties=None): + self.object_path = object_path + self.interface = interface + self.bus = specified_bus + + self.dbus_method = dbus.Interface(specified_bus.get_object( + BUSNAME, self.object_path), self.interface) + + self._set_props(properties) + + def __getattr__(self, item): + if hasattr(self.dbus_method, item): + return functools.partial(self._wrapper, item) + else: + return functools.partial(self, item) + + def _wrapper(self, _method_name, *args, **kwargs): + return getattr(self.dbus_method, _method_name)(*args, **kwargs) + + def update(self): + self._set_props() + + +class ClientProxy(object): + @staticmethod + def _intf_short_name(nm): + return nm.split('.')[-1:][0] + + def __init__(self, specified_bus, object_path, interface=None, props=None): + i = dbus.Interface(specified_bus.get_object( + BUSNAME, object_path), 'org.freedesktop.DBus.Introspectable') + self.intro_spect = DbusIntrospection.introspect(i.Introspect()) + + for k in self.intro_spect.keys(): + sn = ClientProxy._intf_short_name(k) + # print('Client proxy has interface: %s %s' % (k, sn)) + + if interface and interface == k and props is not None: + ro = RemoteObject(specified_bus, object_path, k, props) + else: + ro = RemoteObject(specified_bus, object_path, k) + + setattr(self, sn, ro) + + self.object_path = object_path + + def update(self): + # Go through all interfaces and update them + for int_f in self.intro_spect.keys(): + sn = ClientProxy._intf_short_name(int_f) + getattr(self, sn).update() |