696
682
return new_object
699
class dbussy_adapter:
700
class SystemBus(dbus.SystemBus):
703
def __init__(self, dbussy, ravel):
706
self.bus = ravel.system_bus()
708
@contextlib.contextmanager
709
def convert_exception(self, exception_class=dbus.Error):
712
except self.dbussy.DBusError as e:
713
# This does what "raise from" would do
714
exc = exception_class(*e.args)
718
def call_method(self, methodname, busname, objectpath,
720
proxy_object = self.get_object(busname, objectpath)
721
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
722
interface, methodname,
723
", ".join(repr(a) for a in args))
724
iface = proxy_object.get_interface(interface)
725
method = getattr(iface, methodname)
726
with self.convert_exception(dbus.Error):
727
value = method(*args)
728
# DBussy returns values either as an empty list or as a
729
# tuple: (signature, value)
731
return self.type_filter(value[0])
733
def get_object(self, busname, objectpath):
734
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
736
with self.convert_exception(dbus.ConnectFailed):
737
return self.bus[busname][objectpath]
739
def type_filter(self, value):
740
"""Convert the most bothersome types to Python types"""
741
if isinstance(value, tuple):
743
and isinstance(value[0],
744
self.dbussy.DBUS.Signature)):
745
return self.type_filter(value[1])
746
elif isinstance(value, self.dbussy.DBUS.ObjectPath):
748
# Also recurse into dictionaries
749
elif isinstance(value, dict):
750
return {self.type_filter(key):
751
self.type_filter(subval)
752
for key, subval in value.items()}
755
def set_property(self, busname, objectpath, interface, key,
757
proxy_object = self.get_object(busname, objectpath)
758
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
759
objectpath, self.properties_iface, interface,
762
# DBussy wants a Byte Array to be a sequence of
763
# values, not a byte string
765
setattr(proxy_object.get_interface(interface), key, value)
767
class MandosBus(SystemBus, dbus.MandosBus):
770
class CachingBus(MandosBus):
771
"""A caching layer for dbussy_adapter.MandosBus"""
772
def __init__(self, *args, **kwargs):
773
self.object_cache = {}
774
super(dbussy_adapter.CachingBus, self).__init__(*args,
776
def get_object(self, busname, objectpath):
778
return self.object_cache[(busname, objectpath)]
781
dbussy_adapter.CachingBus,
782
self).get_object(busname, objectpath)
783
self.object_cache[(busname, objectpath)] = new_object
787
685
def commands_from_options(options):
789
687
commands = list(options.commands)
1876
1774
self.assertIs(obj1, obj1b)
1879
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
1883
class ObjectPath(str):
1885
class DBusError(Exception):
1888
def fake_ravel_func(self, func):
1892
class DBusInterfaceProxy:
1894
def methodname(*args):
1895
return [func(*args)]
1898
def get_interface(interface):
1899
if interface == "interface":
1900
return DBusInterfaceProxy()
1901
return {"busname": {"objectpath": DBusObject()}}
1904
def call_method(self, bus, methodname, busname, objectpath,
1906
with self.assertLogs(log, logging.DEBUG):
1907
return bus.call_method(methodname, busname, objectpath,
1910
def test_call_method_returns(self):
1911
expected_method_return = Unique()
1912
method_args = (Unique(), Unique())
1914
self.assertEqual(len(method_args), len(args))
1915
for marg, arg in zip(method_args, args):
1916
self.assertIs(marg, arg)
1917
return expected_method_return
1918
fake_ravel = self.fake_ravel_func(func)
1919
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1920
ret = self.call_method(bus, "methodname", "busname",
1921
"objectpath", "interface",
1923
self.assertIs(ret, expected_method_return)
1925
def test_call_method_filters_objectpath(self):
1927
return method_return
1928
fake_ravel = self.fake_ravel_func(func)
1929
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1930
method_return = (self.dummy_dbussy.DBUS
1931
.ObjectPath("objectpath"))
1932
ret = self.call_method(bus, "methodname", "busname",
1933
"objectpath", "interface")
1934
self.assertEqual("objectpath", ret)
1935
self.assertNotIsInstance(ret,
1936
self.dummy_dbussy.DBUS.ObjectPath)
1938
def test_call_method_filters_objectpaths_in_dict(self):
1939
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1941
return method_return
1942
fake_ravel = self.fake_ravel_func(func)
1943
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1945
ObjectPath("objectpath_key_1"):
1946
ObjectPath("objectpath_value_1"),
1947
ObjectPath("objectpath_key_2"):
1948
ObjectPath("objectpath_value_2"),
1950
ret = self.call_method(bus, "methodname", "busname",
1951
"objectpath", "interface")
1952
expected_method_return = {str(key): str(value)
1954
method_return.items()}
1955
for key, value in ret.items():
1956
self.assertNotIsInstance(key, ObjectPath)
1957
self.assertNotIsInstance(value, ObjectPath)
1958
self.assertEqual(expected_method_return, ret)
1959
self.assertIsInstance(ret, dict)
1961
def test_call_method_filters_objectpaths_in_dict_in_dict(self):
1962
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1964
return method_return
1965
fake_ravel = self.fake_ravel_func(func)
1966
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1968
ObjectPath("key1"): {
1969
ObjectPath("key11"): ObjectPath("value11"),
1970
ObjectPath("key12"): ObjectPath("value12"),
1972
ObjectPath("key2"): {
1973
ObjectPath("key21"): ObjectPath("value21"),
1974
ObjectPath("key22"): ObjectPath("value22"),
1977
ret = self.call_method(bus, "methodname", "busname",
1978
"objectpath", "interface")
1979
expected_method_return = {
1980
"key1": {"key11": "value11",
1981
"key12": "value12"},
1982
"key2": {"key21": "value21",
1983
"key22": "value22"},
1985
self.assertEqual(expected_method_return, ret)
1986
for key, value in ret.items():
1987
self.assertIsInstance(value, dict)
1988
self.assertEqual(expected_method_return[key], value)
1989
self.assertNotIsInstance(key, ObjectPath)
1990
for inner_key, inner_value in value.items():
1991
self.assertIsInstance(value, dict)
1993
expected_method_return[key][inner_key],
1995
self.assertNotIsInstance(key, ObjectPath)
1997
def test_call_method_filters_objectpaths_in_dict_three_deep(self):
1998
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
2000
return method_return
2001
fake_ravel = self.fake_ravel_func(func)
2002
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2004
ObjectPath("key1"): {
2005
ObjectPath("key2"): {
2006
ObjectPath("key3"): ObjectPath("value"),
2010
ret = self.call_method(bus, "methodname", "busname",
2011
"objectpath", "interface")
2012
expected_method_return = {"key1": {"key2": {"key3": "value"}}}
2013
self.assertEqual(expected_method_return, ret)
2014
self.assertIsInstance(ret, dict)
2015
self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
2016
self.assertIsInstance(ret["key1"], dict)
2017
self.assertNotIsInstance(next(iter(ret["key1"].keys())),
2019
self.assertIsInstance(ret["key1"]["key2"], dict)
2020
self.assertNotIsInstance(
2021
next(iter(ret["key1"]["key2"].keys())),
2023
self.assertEqual("value", ret["key1"]["key2"]["key3"])
2024
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
2025
self.dummy_dbussy.DBUS.ObjectPath)
2027
def test_call_method_handles_exception(self):
2029
raise self.dummy_dbussy.DBusError()
2031
fake_ravel = self.fake_ravel_func(func)
2032
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2034
with self.assertRaises(dbus.Error) as e:
2035
self.call_method(bus, "methodname", "busname",
2036
"objectpath", "interface")
2038
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
2040
def test_get_object_converts_to_correct_exception(self):
2041
class fake_ravel_raises_exception_on_connect:
2046
def __getitem__(key):
2047
if key == "objectpath":
2048
raise self.dummy_dbussy.DBusError()
2049
raise Exception(key)
2050
return {"busname": Bus()}
2052
raise self.dummy_dbussy.DBusError()
2053
bus = dbussy_adapter.SystemBus(
2055
fake_ravel_raises_exception_on_connect)
2056
with self.assertRaises(dbus.ConnectFailed):
2057
self.call_method(bus, "methodname", "busname",
2058
"objectpath", "interface")
2061
1777
class Test_commands_from_options(unittest.TestCase):
2063
1779
def setUp(self):