To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk
« back to all changes in this revision
Viewing changes to mandos-ctl
- browse files at revision 1097
- compare with another revision
- download diff
- download tarball
- view history from revision 1097
- Committer: Teddy Hogeborn
- Date: 2019-04-08 21:53:22 UTC
- Revision ID: teddy@recompile.se-20190408215322-y3hmfxzdgs9t84l1
plugin-runner: Fix minor memory leak
* plugin-runner.c (free_plugin): Even if argv[0] is NULL, as for the
pseudo-plugin for global options,
free all other arguments by simply
starting at argv[1] and freeing
plugin_node->name (which is always
argv[0]) separately.
* plugin-runner.c (free_plugin): Even if argv[0] is NULL, as for the
pseudo-plugin for global options,
free all other arguments by simply
starting at argv[1] and freeing
plugin_node->name (which is always
argv[0]) separately.
- files removed:
- debian/po/POTFILES.in
- debian/tests
- dracut-module
- files modified:
- .bzrignore
added
removed
mandos-ctl
1
#!/usr/bin/python3 -bbI
2
# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-
1
#!/usr/bin/python
2
# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-
3
3
#
4
4
# Mandos Monitor - Control and monitor the Mandos server
5
5
#
46
46
import tempfile
47
47
import contextlib
48
48
49
if sys.version_info.major == 2:
50
__metaclass__ = type
51
str = unicode
52
53
class gi:
54
"""Dummy gi module, for the tests"""
55
class repository:
56
class GLib:
57
class Error(Exception):
58
pass
59
dbussy = None
60
ravel = None
61
dbus_python = None
62
pydbus = None
63
64
49
try:
65
import dbussy
66
import ravel
50
import pydbus
51
import gi
52
dbus_python = None
67
53
except ImportError:
68
try:
69
import pydbus
70
import gi
71
except ImportError:
72
import dbus as dbus_python
73
54
import dbus as dbus_python
55
pydbus = None
56
class gi(object):
57
"""Dummy gi module, for the tests"""
58
class repository(object):
59
class GLib(object):
60
class Error(Exception):
61
pass
74
62
75
63
# Show warnings by default
76
64
if not sys.warnoptions:
84
72
logging.captureWarnings(True) # Show warnings via the logging system
85
73
86
74
if sys.version_info.major == 2:
75
str = unicode
87
76
import StringIO
88
77
io.StringIO = StringIO.StringIO
89
78
90
79
locale.setlocale(locale.LC_ALL, "")
91
80
92
version = "1.8.9"
81
version = "1.8.3"
93
82
94
83
95
84
def main():
104
93
if options.debug:
105
94
log.setLevel(logging.DEBUG)
106
95
107
if dbussy is not None and ravel is not None:
108
bus = dbussy_adapter.CachingBus(dbussy, ravel)
109
elif pydbus is not None:
96
if pydbus is not None:
110
97
bus = pydbus_adapter.CachingBus(pydbus)
111
98
else:
112
99
bus = dbus_python_adapter.CachingBus(dbus_python)
263
250
def rfc3339_duration_to_delta(duration):
264
251
"""Parse an RFC 3339 "duration" and return a datetime.timedelta
265
252
266
>>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)
267
True
268
>>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)
269
True
270
>>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)
271
True
272
>>> # 60 months
273
>>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)
274
True
275
>>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)
276
True
277
>>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)
278
True
279
>>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)
280
True
281
>>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)
282
True
253
>>> rfc3339_duration_to_delta("P7D")
254
datetime.timedelta(7)
255
>>> rfc3339_duration_to_delta("PT60S")
256
datetime.timedelta(0, 60)
257
>>> rfc3339_duration_to_delta("PT60M")
258
datetime.timedelta(0, 3600)
259
>>> rfc3339_duration_to_delta("P60M")
260
datetime.timedelta(1680)
261
>>> rfc3339_duration_to_delta("PT24H")
262
datetime.timedelta(1)
263
>>> rfc3339_duration_to_delta("P1W")
264
datetime.timedelta(7)
265
>>> rfc3339_duration_to_delta("PT5M30S")
266
datetime.timedelta(0, 330)
267
>>> rfc3339_duration_to_delta("P1DT3M20S")
268
datetime.timedelta(1, 200)
283
269
>>> # Can not be empty:
284
270
>>> rfc3339_duration_to_delta("")
285
271
Traceback (most recent call last):
395
381
"""Parse an interval string as documented by Mandos before 1.6.1,
396
382
and return a datetime.timedelta
397
383
398
>>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)
399
True
400
>>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)
401
True
402
>>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)
403
True
404
>>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)
405
True
406
>>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)
407
True
408
>>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)
409
True
410
>>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)
411
True
384
>>> parse_pre_1_6_1_interval('7d')
385
datetime.timedelta(7)
386
>>> parse_pre_1_6_1_interval('60s')
387
datetime.timedelta(0, 60)
388
>>> parse_pre_1_6_1_interval('60m')
389
datetime.timedelta(0, 3600)
390
>>> parse_pre_1_6_1_interval('24h')
391
datetime.timedelta(1)
392
>>> parse_pre_1_6_1_interval('1w')
393
datetime.timedelta(7)
394
>>> parse_pre_1_6_1_interval('5m 30s')
395
datetime.timedelta(0, 330)
396
>>> parse_pre_1_6_1_interval('')
397
datetime.timedelta(0)
412
398
>>> # Ignore unknown characters, allow any order and repetitions
413
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
414
True
399
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
400
datetime.timedelta(2, 480, 18000)
415
401
416
402
"""
417
403
480
466
parser.error("--remove can only be combined with --deny")
481
467
482
468
483
class dbus:
469
class dbus(object):
484
470
485
class SystemBus:
471
class SystemBus(object):
486
472
487
473
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
488
474
def get_managed_objects(self, busname, objectpath):
497
483
self.properties_iface, interface, key,
498
484
value)
499
485
486
500
487
class MandosBus(SystemBus):
501
488
busname_domain = "se.recompile"
502
489
busname = busname_domain + ".Mandos"
533
520
pass
534
521
535
522
536
class dbus_python_adapter:
523
class dbus_python_adapter(object):
537
524
538
525
class SystemBus(dbus.MandosBus):
539
526
"""Use dbus-python"""
584
571
for key, subval in value.items()}
585
572
return value
586
573
587
def set_client_property(self, objectpath, key, value):
588
if key == "Secret":
589
if not isinstance(value, bytes):
590
value = value.encode("utf-8")
591
value = self.dbus_python.ByteArray(value)
592
return self.set_property(self.busname, objectpath,
593
self.client_interface, key,
594
value)
595
574
596
class SilenceLogger:
575
class SilenceLogger(object):
597
576
"Simple context manager to silence a particular logger"
598
577
def __init__(self, loggername):
599
578
self.logger = logging.getLogger(loggername)
628
607
return new_object
629
608
630
609
631
class pydbus_adapter:
610
class pydbus_adapter(object):
632
611
class SystemBus(dbus.MandosBus):
633
612
def __init__(self, module=pydbus):
634
613
self.pydbus = module
691
670
return new_object
692
671
693
672
694
class dbussy_adapter:
695
class SystemBus(dbus.SystemBus):
696
"""Use DBussy"""
697
698
def __init__(self, dbussy, ravel):
699
self.dbussy = dbussy
700
self.ravel = ravel
701
self.bus = ravel.system_bus()
702
703
@contextlib.contextmanager
704
def convert_exception(self, exception_class=dbus.Error):
705
try:
706
yield
707
except self.dbussy.DBusError as e:
708
# This does what "raise from" would do
709
exc = exception_class(*e.args)
710
exc.__cause__ = e
711
raise exc
712
713
def call_method(self, methodname, busname, objectpath,
714
interface, *args):
715
proxy_object = self.get_object(busname, objectpath)
716
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
717
interface, methodname,
718
", ".join(repr(a) for a in args))
719
iface = proxy_object.get_interface(interface)
720
method = getattr(iface, methodname)
721
with self.convert_exception(dbus.Error):
722
value = method(*args)
723
# DBussy returns values either as an empty list or as a
724
# tuple: (signature, value)
725
if value:
726
return self.type_filter(value[0])
727
728
def get_object(self, busname, objectpath):
729
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
730
busname, objectpath)
731
with self.convert_exception(dbus.ConnectFailed):
732
return self.bus[busname][objectpath]
733
734
def type_filter(self, value):
735
"""Convert the most bothersome types to Python types"""
736
if isinstance(value, tuple):
737
if (len(value) == 2
738
and isinstance(value[0],
739
self.dbussy.DBUS.Signature)):
740
return self.type_filter(value[1])
741
elif isinstance(value, self.dbussy.DBUS.ObjectPath):
742
return str(value)
743
# Also recurse into dictionaries
744
elif isinstance(value, dict):
745
return {self.type_filter(key):
746
self.type_filter(subval)
747
for key, subval in value.items()}
748
return value
749
750
def set_property(self, busname, objectpath, interface, key,
751
value):
752
proxy_object = self.get_object(busname, objectpath)
753
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
754
objectpath, self.properties_iface, interface,
755
key, value)
756
if key == "Secret":
757
# DBussy wants a Byte Array to be a sequence of
758
# values, not a byte string
759
value = tuple(value)
760
setattr(proxy_object.get_interface(interface), key, value)
761
762
class MandosBus(SystemBus, dbus.MandosBus):
763
pass
764
765
class CachingBus(MandosBus):
766
"""A caching layer for dbussy_adapter.MandosBus"""
767
def __init__(self, *args, **kwargs):
768
self.object_cache = {}
769
super(dbussy_adapter.CachingBus, self).__init__(*args,
770
**kwargs)
771
def get_object(self, busname, objectpath):
772
try:
773
return self.object_cache[(busname, objectpath)]
774
except KeyError:
775
new_object = super(
776
dbussy_adapter.CachingBus,
777
self).get_object(busname, objectpath)
778
self.object_cache[(busname, objectpath)] = new_object
779
return new_object
780
781
782
673
def commands_from_options(options):
783
674
784
675
commands = list(options.commands)
812
703
return commands
813
704
814
705
815
class command:
706
class command(object):
816
707
"""A namespace for command classes"""
817
708
818
class Base:
709
class Base(object):
819
710
"""Abstract base class for commands"""
820
711
def run(self, clients, bus=None):
821
712
"""Normal commands should implement run_on_one_client(),
884
775
keywords = self.all_keywords
885
776
print(self.TableOfClients(clients.values(), keywords))
886
777
887
class TableOfClients:
778
class TableOfClients(object):
888
779
tableheaders = {
889
780
"Name": "Name",
890
781
"Enabled": "Enabled",
1121
1012
"output"))
1122
1013
1123
1014
1124
class Unique:
1015
class Unique(object):
1125
1016
"""Class for objects which exist only to be unique objects, since
1126
1017
unittest.mock.sentinel only exists in Python 3.3"""
1127
1018
1411
1302
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1412
1303
1413
1304
def MockDBusPython_func(self, func):
1414
class mock_dbus_python:
1305
class mock_dbus_python(object):
1415
1306
"""mock dbus-python module"""
1416
class exceptions:
1307
class exceptions(object):
1417
1308
"""Pseudo-namespace"""
1418
1309
class DBusException(Exception):
1419
1310
pass
1420
class SystemBus:
1311
class SystemBus(object):
1421
1312
@staticmethod
1422
1313
def get_object(busname, objectpath):
1423
1314
DBusObject = collections.namedtuple(
1424
"DBusObject", ("methodname", "Set"))
1315
"DBusObject", ("methodname",))
1425
1316
def method(*args, **kwargs):
1426
1317
self.assertEqual({"dbus_interface":
1427
1318
"interface"},
1428
1319
kwargs)
1429
1320
return func(*args)
1430
def set_property(interface, key, value,
1431
dbus_interface=None):
1432
self.assertEqual(
1433
"org.freedesktop.DBus.Properties",
1434
dbus_interface)
1435
self.assertEqual("Secret", key)
1436
return func(interface, key, value,
1437
dbus_interface=dbus_interface)
1438
return DBusObject(methodname=method,
1439
Set=set_property)
1440
class Boolean:
1321
return DBusObject(methodname=method)
1322
class Boolean(object):
1441
1323
def __init__(self, value):
1442
1324
self.value = bool(value)
1443
1325
def __bool__(self):
1448
1330
pass
1449
1331
class Dictionary(dict):
1450
1332
pass
1451
class ByteArray(bytes):
1452
pass
1453
1333
return mock_dbus_python
1454
1334
1455
1335
def call_method(self, bus, methodname, busname, objectpath,
1632
1512
# Make sure the dbus logger was suppressed
1633
1513
self.assertEqual(0, counting_handler.count)
1634
1514
1635
def test_Set_Secret_sends_bytearray(self):
1636
ret = [None]
1637
def func(*args, **kwargs):
1638
ret[0] = (args, kwargs)
1639
mock_dbus_python = self.MockDBusPython_func(func)
1640
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1641
bus.set_client_property("objectpath", "Secret", "value")
1642
expected_call = (("se.recompile.Mandos.Client", "Secret",
1643
mock_dbus_python.ByteArray(b"value")),
1644
{"dbus_interface":
1645
"org.freedesktop.DBus.Properties"})
1646
self.assertEqual(expected_call, ret[0])
1647
if sys.version_info.major == 2:
1648
self.assertIsInstance(ret[0][0][-1],
1649
mock_dbus_python.ByteArray)
1650
1651
1515
def test_get_object_converts_to_correct_exception(self):
1652
1516
bus = dbus_python_adapter.SystemBus(
1653
1517
self.fake_dbus_python_raises_exception_on_connect)
1655
1519
self.call_method(bus, "methodname", "busname",
1656
1520
"objectpath", "interface")
1657
1521
1658
class fake_dbus_python_raises_exception_on_connect:
1522
class fake_dbus_python_raises_exception_on_connect(object):
1659
1523
"""fake dbus-python module"""
1660
class exceptions:
1524
class exceptions(object):
1661
1525
"""Pseudo-namespace"""
1662
1526
class DBusException(Exception):
1663
1527
pass
1671
1535
1672
1536
1673
1537
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1674
class mock_dbus_python:
1538
class mock_dbus_python(object):
1675
1539
"""mock dbus-python modules"""
1676
class SystemBus:
1540
class SystemBus(object):
1677
1541
@staticmethod
1678
1542
def get_object(busname, objectpath):
1679
1543
return Unique()
1725
1589
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1726
1590
1727
1591
def Stub_pydbus_func(self, func):
1728
class stub_pydbus:
1592
class stub_pydbus(object):
1729
1593
"""stub pydbus module"""
1730
class SystemBus:
1594
class SystemBus(object):
1731
1595
@staticmethod
1732
1596
def get(busname, objectpath):
1733
1597
DBusObject = collections.namedtuple(
1779
1643
self.call_method(bus, "methodname", "busname",
1780
1644
"objectpath", "interface")
1781
1645
1782
class fake_pydbus_raises_exception_on_connect:
1646
class fake_pydbus_raises_exception_on_connect(object):
1783
1647
"""fake dbus-python module"""
1784
1648
@classmethod
1785
1649
def SystemBus(cls):
1789
1653
return Bus(get=get)
1790
1654
1791
1655
def test_set_property_uses_setattr(self):
1792
class Object:
1656
class Object(object):
1793
1657
pass
1794
1658
obj = Object()
1795
class pydbus_spy:
1796
class SystemBus:
1659
class pydbus_spy(object):
1660
class SystemBus(object):
1797
1661
@staticmethod
1798
1662
def get(busname, objectpath):
1799
1663
return {"interface": obj}
1806
1670
def test_get_suppresses_xml_deprecation_warning(self):
1807
1671
if sys.version_info.major >= 3:
1808
1672
return
1809
class stub_pydbus_get:
1810
class SystemBus:
1673
class stub_pydbus_get(object):
1674
class SystemBus(object):
1811
1675
@staticmethod
1812
1676
def get(busname, objectpath):
1813
1677
warnings.warn_explicit(
1821
1685
1822
1686
1823
1687
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1824
class stub_pydbus:
1688
class stub_pydbus(object):
1825
1689
"""stub pydbus module"""
1826
class SystemBus:
1690
class SystemBus(object):
1827
1691
@staticmethod
1828
1692
def get(busname, objectpath):
1829
1693
return Unique()
1871
1735
self.assertIs(obj1, obj1b)
1872
1736
1873
1737
1874
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
1875
1876
class dummy_dbussy:
1877
class DBUS:
1878
class ObjectPath(str):
1879
pass
1880
class DBusError(Exception):
1881
pass
1882
1883
def fake_ravel_func(self, func):
1884
class fake_ravel:
1885
@staticmethod
1886
def system_bus():
1887
class DBusInterfaceProxy:
1888
@staticmethod
1889
def methodname(*args):
1890
return [func(*args)]
1891
class DBusObject:
1892
@staticmethod
1893
def get_interface(interface):
1894
if interface == "interface":
1895
return DBusInterfaceProxy()
1896
return {"busname": {"objectpath": DBusObject()}}
1897
return fake_ravel
1898
1899
def call_method(self, bus, methodname, busname, objectpath,
1900
interface, *args):
1901
with self.assertLogs(log, logging.DEBUG):
1902
return bus.call_method(methodname, busname, objectpath,
1903
interface, *args)
1904
1905
def test_call_method_returns(self):
1906
expected_method_return = Unique()
1907
method_args = (Unique(), Unique())
1908
def func(*args):
1909
self.assertEqual(len(method_args), len(args))
1910
for marg, arg in zip(method_args, args):
1911
self.assertIs(marg, arg)
1912
return expected_method_return
1913
fake_ravel = self.fake_ravel_func(func)
1914
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1915
ret = self.call_method(bus, "methodname", "busname",
1916
"objectpath", "interface",
1917
*method_args)
1918
self.assertIs(ret, expected_method_return)
1919
1920
def test_call_method_filters_objectpath(self):
1921
def func():
1922
return method_return
1923
fake_ravel = self.fake_ravel_func(func)
1924
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1925
method_return = (self.dummy_dbussy.DBUS
1926
.ObjectPath("objectpath"))
1927
ret = self.call_method(bus, "methodname", "busname",
1928
"objectpath", "interface")
1929
self.assertEqual("objectpath", ret)
1930
self.assertNotIsInstance(ret,
1931
self.dummy_dbussy.DBUS.ObjectPath)
1932
1933
def test_call_method_filters_objectpaths_in_dict(self):
1934
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1935
def func():
1936
return method_return
1937
fake_ravel = self.fake_ravel_func(func)
1938
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1939
method_return = {
1940
ObjectPath("objectpath_key_1"):
1941
ObjectPath("objectpath_value_1"),
1942
ObjectPath("objectpath_key_2"):
1943
ObjectPath("objectpath_value_2"),
1944
}
1945
ret = self.call_method(bus, "methodname", "busname",
1946
"objectpath", "interface")
1947
expected_method_return = {str(key): str(value)
1948
for key, value in
1949
method_return.items()}
1950
for key, value in ret.items():
1951
self.assertNotIsInstance(key, ObjectPath)
1952
self.assertNotIsInstance(value, ObjectPath)
1953
self.assertEqual(expected_method_return, ret)
1954
self.assertIsInstance(ret, dict)
1955
1956
def test_call_method_filters_objectpaths_in_dict_in_dict(self):
1957
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1958
def func():
1959
return method_return
1960
fake_ravel = self.fake_ravel_func(func)
1961
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1962
method_return = {
1963
ObjectPath("key1"): {
1964
ObjectPath("key11"): ObjectPath("value11"),
1965
ObjectPath("key12"): ObjectPath("value12"),
1966
},
1967
ObjectPath("key2"): {
1968
ObjectPath("key21"): ObjectPath("value21"),
1969
ObjectPath("key22"): ObjectPath("value22"),
1970
},
1971
}
1972
ret = self.call_method(bus, "methodname", "busname",
1973
"objectpath", "interface")
1974
expected_method_return = {
1975
"key1": {"key11": "value11",
1976
"key12": "value12"},
1977
"key2": {"key21": "value21",
1978
"key22": "value22"},
1979
}
1980
self.assertEqual(expected_method_return, ret)
1981
for key, value in ret.items():
1982
self.assertIsInstance(value, dict)
1983
self.assertEqual(expected_method_return[key], value)
1984
self.assertNotIsInstance(key, ObjectPath)
1985
for inner_key, inner_value in value.items():
1986
self.assertIsInstance(value, dict)
1987
self.assertEqual(
1988
expected_method_return[key][inner_key],
1989
inner_value)
1990
self.assertNotIsInstance(key, ObjectPath)
1991
1992
def test_call_method_filters_objectpaths_in_dict_three_deep(self):
1993
ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1994
def func():
1995
return method_return
1996
fake_ravel = self.fake_ravel_func(func)
1997
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1998
method_return = {
1999
ObjectPath("key1"): {
2000
ObjectPath("key2"): {
2001
ObjectPath("key3"): ObjectPath("value"),
2002
},
2003
},
2004
}
2005
ret = self.call_method(bus, "methodname", "busname",
2006
"objectpath", "interface")
2007
expected_method_return = {"key1": {"key2": {"key3": "value"}}}
2008
self.assertEqual(expected_method_return, ret)
2009
self.assertIsInstance(ret, dict)
2010
self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
2011
self.assertIsInstance(ret["key1"], dict)
2012
self.assertNotIsInstance(next(iter(ret["key1"].keys())),
2013
ObjectPath)
2014
self.assertIsInstance(ret["key1"]["key2"], dict)
2015
self.assertNotIsInstance(
2016
next(iter(ret["key1"]["key2"].keys())),
2017
ObjectPath)
2018
self.assertEqual("value", ret["key1"]["key2"]["key3"])
2019
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
2020
self.dummy_dbussy.DBUS.ObjectPath)
2021
2022
def test_call_method_handles_exception(self):
2023
def func():
2024
raise self.dummy_dbussy.DBusError()
2025
2026
fake_ravel = self.fake_ravel_func(func)
2027
bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2028
2029
with self.assertRaises(dbus.Error) as e:
2030
self.call_method(bus, "methodname", "busname",
2031
"objectpath", "interface")
2032
2033
self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
2034
2035
def test_get_object_converts_to_correct_exception(self):
2036
class fake_ravel_raises_exception_on_connect:
2037
@staticmethod
2038
def system_bus():
2039
class Bus:
2040
@staticmethod
2041
def __getitem__(key):
2042
if key == "objectpath":
2043
raise self.dummy_dbussy.DBusError()
2044
raise Exception(key)
2045
return {"busname": Bus()}
2046
def func():
2047
raise self.dummy_dbussy.DBusError()
2048
bus = dbussy_adapter.SystemBus(
2049
self.dummy_dbussy,
2050
fake_ravel_raises_exception_on_connect)
2051
with self.assertRaises(dbus.ConnectFailed):
2052
self.call_method(bus, "methodname", "busname",
2053
"objectpath", "interface")
2054
2055
2056
1738
class Test_commands_from_options(unittest.TestCase):
2057
1739
2058
1740
def setUp(self):
2063
1745
self.assert_command_from_args(["--is-enabled", "client"],
2064
1746
command.IsEnabled)
2065
1747
2066
def assert_command_from_args(self, args, command_cls, length=1,
2067
clients=None, **cmd_attrs):
1748
def assert_command_from_args(self, args, command_cls,
1749
**cmd_attrs):
2068
1750
"""Assert that parsing ARGS should result in an instance of
2069
1751
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
2070
1752
options = self.parser.parse_args(args)
2071
1753
check_option_syntax(self.parser, options)
2072
1754
commands = commands_from_options(options)
2073
self.assertEqual(length, len(commands))
2074
for command in commands:
2075
if isinstance(command, command_cls):
2076
break
2077
else:
2078
self.assertIsInstance(command, command_cls)
2079
if clients is not None:
2080
self.assertEqual(clients, options.client)
1755
self.assertEqual(1, len(commands))
1756
command = commands[0]
1757
self.assertIsInstance(command, command_cls)
2081
1758
for key, value in cmd_attrs.items():
2082
1759
self.assertEqual(value, getattr(command, key))
2083
1760
2084
def assert_commands_from_args(self, args, commands, clients=None):
2085
for cmd in commands:
2086
self.assert_command_from_args(args, cmd,
2087
length=len(commands),
2088
clients=clients)
2089
2090
1761
def test_is_enabled_short(self):
2091
1762
self.assert_command_from_args(["-V", "client"],
2092
1763
command.IsEnabled)
2283
1954
verbose=True)
2284
1955
2285
1956
2286
def test_manual_page_example_1(self):
2287
self.assert_command_from_args("",
2288
command.PrintTable,
2289
clients=[],
2290
verbose=False)
2291
2292
def test_manual_page_example_2(self):
2293
self.assert_command_from_args(
2294
"--verbose foo1.example.org foo2.example.org".split(),
2295
command.PrintTable, clients=["foo1.example.org",
2296
"foo2.example.org"],
2297
verbose=True)
2298
2299
def test_manual_page_example_3(self):
2300
self.assert_command_from_args("--enable --all".split(),
2301
command.Enable,
2302
clients=[])
2303
2304
def test_manual_page_example_4(self):
2305
self.assert_commands_from_args(
2306
("--timeout=PT5M --interval=PT1M foo1.example.org"
2307
" foo2.example.org").split(),
2308
[command.SetTimeout, command.SetInterval],
2309
clients=["foo1.example.org", "foo2.example.org"])
2310
2311
def test_manual_page_example_5(self):
2312
self.assert_command_from_args("--approve --all".split(),
2313
command.Approve,
2314
clients=[])
2315
2316
2317
1957
class TestCommand(unittest.TestCase):
2318
1958
"""Abstract class for tests of command classes"""
2319
1959
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0