/mandos/release

To get this branch, use:
bzr branch http://bzr.recompile.se/loggerhead/mandos/release

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-03-31 09:13:31 UTC
  • mto: This revision was merged to the branch mainline in revision 382.
  • Revision ID: teddy@recompile.se-20190331091331-avb4odg8uz1yb5tm
mandos-ctl: Add support for D-Bus module "pydbus"

* mandos-ctl: Try to import pydbus and gi modules.
  (main): Use pydbus if available, fall back to dbus-python.
  (pydbus_adapter): New.
  (Test_pydbus_adapter_SystemBus): - '' -
  (Test_pydbus_adapter_CachingBus): - '' -

Show diffs side-by-side

added added

removed removed

Lines of Context:
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 -*-
3
 
#
4
 
# Mandos Control - Control or query the Mandos server
5
 
#
6
 
# Copyright © 2008-2020 Teddy Hogeborn
7
 
# Copyright © 2008-2020 Björn Påhlsson
 
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
#
 
4
# Mandos Monitor - Control and monitor the Mandos server
 
5
#
 
6
# Copyright © 2008-2019 Teddy Hogeborn
 
7
# Copyright © 2008-2019 Björn Påhlsson
8
8
#
9
9
# This file is part of Mandos.
10
10
#
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.11"
 
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
 
500
 
        def call_method(self, methodname, busname, objectpath,
501
 
                        interface, *args):
502
 
            raise NotImplementedError()
503
 
 
504
486
 
505
487
    class MandosBus(SystemBus):
506
488
        busname_domain = "se.recompile"
538
520
        pass
539
521
 
540
522
 
541
 
class dbus_python_adapter:
 
523
class dbus_python_adapter(object):
542
524
 
543
525
    class SystemBus(dbus.MandosBus):
544
526
        """Use dbus-python"""
589
571
                        for key, subval in value.items()}
590
572
            return value
591
573
 
592
 
        def set_client_property(self, objectpath, key, value):
593
 
            if key == "Secret":
594
 
                if not isinstance(value, bytes):
595
 
                    value = value.encode("utf-8")
596
 
                value = self.dbus_python.ByteArray(value)
597
 
            return self.set_property(self.busname, objectpath,
598
 
                                     self.client_interface, key,
599
 
                                     value)
600
574
 
601
 
    class SilenceLogger:
 
575
    class SilenceLogger(object):
602
576
        "Simple context manager to silence a particular logger"
603
577
        def __init__(self, loggername):
604
578
            self.logger = logging.getLogger(loggername)
633
607
                return new_object
634
608
 
635
609
 
636
 
class pydbus_adapter:
 
610
class pydbus_adapter(object):
637
611
    class SystemBus(dbus.MandosBus):
638
612
        def __init__(self, module=pydbus):
639
613
            self.pydbus = module
696
670
                return new_object
697
671
 
698
672
 
699
 
class dbussy_adapter:
700
 
    class SystemBus(dbus.SystemBus):
701
 
        """Use DBussy"""
702
 
 
703
 
        def __init__(self, dbussy, ravel):
704
 
            self.dbussy = dbussy
705
 
            self.ravel = ravel
706
 
            self.bus = ravel.system_bus()
707
 
 
708
 
        @contextlib.contextmanager
709
 
        def convert_exception(self, exception_class=dbus.Error):
710
 
            try:
711
 
                yield
712
 
            except self.dbussy.DBusError as e:
713
 
                # This does what "raise from" would do
714
 
                exc = exception_class(*e.args)
715
 
                exc.__cause__ = e
716
 
                raise exc
717
 
 
718
 
        def call_method(self, methodname, busname, objectpath,
719
 
                        interface, *args):
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
 
            # list of one element with the return value
730
 
            if value:
731
 
                return self.type_filter(value[0])
732
 
 
733
 
        def get_object(self, busname, objectpath):
734
 
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
735
 
                      busname, objectpath)
736
 
            with self.convert_exception(dbus.ConnectFailed):
737
 
                return self.bus[busname][objectpath]
738
 
 
739
 
        def type_filter(self, value):
740
 
            """Convert the most bothersome types to Python types"""
741
 
            # A D-Bus Variant value is represented as the Python type
742
 
            # Tuple[dbussy.DBUS.Signature, Any]
743
 
            if isinstance(value, tuple):
744
 
                if (len(value) == 2
745
 
                    and isinstance(value[0],
746
 
                                   self.dbussy.DBUS.Signature)):
747
 
                    return self.type_filter(value[1])
748
 
            elif isinstance(value, self.dbussy.DBUS.ObjectPath):
749
 
                return str(value)
750
 
            # Also recurse into dictionaries
751
 
            elif isinstance(value, dict):
752
 
                return {self.type_filter(key):
753
 
                        self.type_filter(subval)
754
 
                        for key, subval in value.items()}
755
 
            return value
756
 
 
757
 
        def set_property(self, busname, objectpath, interface, key,
758
 
                         value):
759
 
            proxy_object = self.get_object(busname, objectpath)
760
 
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
761
 
                      objectpath, self.properties_iface, interface,
762
 
                      key, value)
763
 
            if key == "Secret":
764
 
                # DBussy wants a Byte Array to be a sequence of
765
 
                # values, not a byte string
766
 
                value = tuple(value)
767
 
            setattr(proxy_object.get_interface(interface), key, value)
768
 
 
769
 
    class MandosBus(SystemBus, dbus.MandosBus):
770
 
        pass
771
 
 
772
 
    class CachingBus(MandosBus):
773
 
        """A caching layer for dbussy_adapter.MandosBus"""
774
 
        def __init__(self, *args, **kwargs):
775
 
            self.object_cache = {}
776
 
            super(dbussy_adapter.CachingBus, self).__init__(*args,
777
 
                                                            **kwargs)
778
 
        def get_object(self, busname, objectpath):
779
 
            try:
780
 
                return self.object_cache[(busname, objectpath)]
781
 
            except KeyError:
782
 
                new_object = super(
783
 
                    dbussy_adapter.CachingBus,
784
 
                    self).get_object(busname, objectpath)
785
 
                self.object_cache[(busname, objectpath)]  = new_object
786
 
                return new_object
787
 
 
788
 
 
789
673
def commands_from_options(options):
790
674
 
791
675
    commands = list(options.commands)
819
703
    return commands
820
704
 
821
705
 
822
 
class command:
 
706
class command(object):
823
707
    """A namespace for command classes"""
824
708
 
825
 
    class Base:
 
709
    class Base(object):
826
710
        """Abstract base class for commands"""
827
711
        def run(self, clients, bus=None):
828
712
            """Normal commands should implement run_on_one_client(),
891
775
                keywords = self.all_keywords
892
776
            print(self.TableOfClients(clients.values(), keywords))
893
777
 
894
 
        class TableOfClients:
 
778
        class TableOfClients(object):
895
779
            tableheaders = {
896
780
                "Name": "Name",
897
781
                "Enabled": "Enabled",
1128
1012
                                                     "output"))
1129
1013
 
1130
1014
 
1131
 
class Unique:
 
1015
class Unique(object):
1132
1016
    """Class for objects which exist only to be unique objects, since
1133
1017
unittest.mock.sentinel only exists in Python 3.3"""
1134
1018
 
1418
1302
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1419
1303
 
1420
1304
    def MockDBusPython_func(self, func):
1421
 
        class mock_dbus_python:
 
1305
        class mock_dbus_python(object):
1422
1306
            """mock dbus-python module"""
1423
 
            class exceptions:
 
1307
            class exceptions(object):
1424
1308
                """Pseudo-namespace"""
1425
1309
                class DBusException(Exception):
1426
1310
                    pass
1427
 
            class SystemBus:
 
1311
            class SystemBus(object):
1428
1312
                @staticmethod
1429
1313
                def get_object(busname, objectpath):
1430
1314
                    DBusObject = collections.namedtuple(
1431
 
                        "DBusObject", ("methodname", "Set"))
 
1315
                        "DBusObject", ("methodname",))
1432
1316
                    def method(*args, **kwargs):
1433
1317
                        self.assertEqual({"dbus_interface":
1434
1318
                                          "interface"},
1435
1319
                                         kwargs)
1436
1320
                        return func(*args)
1437
 
                    def set_property(interface, key, value,
1438
 
                                     dbus_interface=None):
1439
 
                        self.assertEqual(
1440
 
                            "org.freedesktop.DBus.Properties",
1441
 
                            dbus_interface)
1442
 
                        self.assertEqual("Secret", key)
1443
 
                        return func(interface, key, value,
1444
 
                                    dbus_interface=dbus_interface)
1445
 
                    return DBusObject(methodname=method,
1446
 
                                      Set=set_property)
1447
 
            class Boolean:
 
1321
                    return DBusObject(methodname=method)
 
1322
            class Boolean(object):
1448
1323
                def __init__(self, value):
1449
1324
                    self.value = bool(value)
1450
1325
                def __bool__(self):
1455
1330
                pass
1456
1331
            class Dictionary(dict):
1457
1332
                pass
1458
 
            class ByteArray(bytes):
1459
 
                pass
1460
1333
        return mock_dbus_python
1461
1334
 
1462
1335
    def call_method(self, bus, methodname, busname, objectpath,
1634
1507
        finally:
1635
1508
            dbus_logger.removeFilter(counting_handler)
1636
1509
 
1637
 
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1510
        self.assertNotIsInstance(e, dbus.ConnectFailed)
1638
1511
 
1639
1512
        # Make sure the dbus logger was suppressed
1640
1513
        self.assertEqual(0, counting_handler.count)
1641
1514
 
1642
 
    def test_Set_Secret_sends_bytearray(self):
1643
 
        ret = [None]
1644
 
        def func(*args, **kwargs):
1645
 
            ret[0] = (args, kwargs)
1646
 
        mock_dbus_python = self.MockDBusPython_func(func)
1647
 
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1648
 
        bus.set_client_property("objectpath", "Secret", "value")
1649
 
        expected_call = (("se.recompile.Mandos.Client", "Secret",
1650
 
                          mock_dbus_python.ByteArray(b"value")),
1651
 
                         {"dbus_interface":
1652
 
                          "org.freedesktop.DBus.Properties"})
1653
 
        self.assertEqual(expected_call, ret[0])
1654
 
        if sys.version_info.major == 2:
1655
 
            self.assertIsInstance(ret[0][0][-1],
1656
 
                                  mock_dbus_python.ByteArray)
1657
 
 
1658
1515
    def test_get_object_converts_to_correct_exception(self):
1659
1516
        bus = dbus_python_adapter.SystemBus(
1660
1517
            self.fake_dbus_python_raises_exception_on_connect)
1662
1519
            self.call_method(bus, "methodname", "busname",
1663
1520
                             "objectpath", "interface")
1664
1521
 
1665
 
    class fake_dbus_python_raises_exception_on_connect:
 
1522
    class fake_dbus_python_raises_exception_on_connect(object):
1666
1523
        """fake dbus-python module"""
1667
 
        class exceptions:
 
1524
        class exceptions(object):
1668
1525
            """Pseudo-namespace"""
1669
1526
            class DBusException(Exception):
1670
1527
                pass
1678
1535
 
1679
1536
 
1680
1537
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1681
 
    class mock_dbus_python:
 
1538
    class mock_dbus_python(object):
1682
1539
        """mock dbus-python modules"""
1683
 
        class SystemBus:
 
1540
        class SystemBus(object):
1684
1541
            @staticmethod
1685
1542
            def get_object(busname, objectpath):
1686
1543
                return Unique()
1732
1589
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1733
1590
 
1734
1591
    def Stub_pydbus_func(self, func):
1735
 
        class stub_pydbus:
 
1592
        class stub_pydbus(object):
1736
1593
            """stub pydbus module"""
1737
 
            class SystemBus:
 
1594
            class SystemBus(object):
1738
1595
                @staticmethod
1739
1596
                def get(busname, objectpath):
1740
1597
                    DBusObject = collections.namedtuple(
1777
1634
            self.call_method(bus, "methodname", "busname",
1778
1635
                             "objectpath", "interface")
1779
1636
 
1780
 
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1637
        self.assertNotIsInstance(e, dbus.ConnectFailed)
1781
1638
 
1782
1639
    def test_get_converts_to_correct_exception(self):
1783
1640
        bus = pydbus_adapter.SystemBus(
1786
1643
            self.call_method(bus, "methodname", "busname",
1787
1644
                             "objectpath", "interface")
1788
1645
 
1789
 
    class fake_pydbus_raises_exception_on_connect:
 
1646
    class fake_pydbus_raises_exception_on_connect(object):
1790
1647
        """fake dbus-python module"""
1791
1648
        @classmethod
1792
1649
        def SystemBus(cls):
1796
1653
            return Bus(get=get)
1797
1654
 
1798
1655
    def test_set_property_uses_setattr(self):
1799
 
        class Object:
 
1656
        class Object(object):
1800
1657
            pass
1801
1658
        obj = Object()
1802
 
        class pydbus_spy:
1803
 
            class SystemBus:
 
1659
        class pydbus_spy(object):
 
1660
            class SystemBus(object):
1804
1661
                @staticmethod
1805
1662
                def get(busname, objectpath):
1806
1663
                    return {"interface": obj}
1813
1670
    def test_get_suppresses_xml_deprecation_warning(self):
1814
1671
        if sys.version_info.major >= 3:
1815
1672
            return
1816
 
        class stub_pydbus_get:
1817
 
            class SystemBus:
 
1673
        class stub_pydbus_get(object):
 
1674
            class SystemBus(object):
1818
1675
                @staticmethod
1819
1676
                def get(busname, objectpath):
1820
1677
                    warnings.warn_explicit(
1828
1685
 
1829
1686
 
1830
1687
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1831
 
    class stub_pydbus:
 
1688
    class stub_pydbus(object):
1832
1689
        """stub pydbus module"""
1833
 
        class SystemBus:
 
1690
        class SystemBus(object):
1834
1691
            @staticmethod
1835
1692
            def get(busname, objectpath):
1836
1693
                return Unique()
1878
1735
        self.assertIs(obj1, obj1b)
1879
1736
 
1880
1737
 
1881
 
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
1882
 
 
1883
 
    class dummy_dbussy:
1884
 
        class DBUS:
1885
 
            class ObjectPath(str):
1886
 
                pass
1887
 
        class DBusError(Exception):
1888
 
            pass
1889
 
 
1890
 
    def fake_ravel_func(self, func):
1891
 
        class fake_ravel:
1892
 
            @staticmethod
1893
 
            def system_bus():
1894
 
                class DBusInterfaceProxy:
1895
 
                    @staticmethod
1896
 
                    def methodname(*args):
1897
 
                        return [func(*args)]
1898
 
                class DBusObject:
1899
 
                    @staticmethod
1900
 
                    def get_interface(interface):
1901
 
                        if interface == "interface":
1902
 
                            return DBusInterfaceProxy()
1903
 
                return {"busname": {"objectpath": DBusObject()}}
1904
 
        return fake_ravel
1905
 
 
1906
 
    def call_method(self, bus, methodname, busname, objectpath,
1907
 
                    interface, *args):
1908
 
        with self.assertLogs(log, logging.DEBUG):
1909
 
            return bus.call_method(methodname, busname, objectpath,
1910
 
                                   interface, *args)
1911
 
 
1912
 
    def test_call_method_returns(self):
1913
 
        expected_method_return = Unique()
1914
 
        method_args = (Unique(), Unique())
1915
 
        def func(*args):
1916
 
            self.assertEqual(len(method_args), len(args))
1917
 
            for marg, arg in zip(method_args, args):
1918
 
                self.assertIs(marg, arg)
1919
 
            return expected_method_return
1920
 
        fake_ravel = self.fake_ravel_func(func)
1921
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1922
 
        ret = self.call_method(bus, "methodname", "busname",
1923
 
                               "objectpath", "interface",
1924
 
                               *method_args)
1925
 
        self.assertIs(ret, expected_method_return)
1926
 
 
1927
 
    def test_call_method_filters_objectpath(self):
1928
 
        def func():
1929
 
            return method_return
1930
 
        fake_ravel = self.fake_ravel_func(func)
1931
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1932
 
        method_return = (self.dummy_dbussy.DBUS
1933
 
                         .ObjectPath("objectpath"))
1934
 
        ret = self.call_method(bus, "methodname", "busname",
1935
 
                               "objectpath", "interface")
1936
 
        self.assertEqual("objectpath", ret)
1937
 
        self.assertNotIsInstance(ret,
1938
 
                                 self.dummy_dbussy.DBUS.ObjectPath)
1939
 
 
1940
 
    def test_call_method_filters_objectpaths_in_dict(self):
1941
 
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1942
 
        def func():
1943
 
            return method_return
1944
 
        fake_ravel = self.fake_ravel_func(func)
1945
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1946
 
        method_return = {
1947
 
            ObjectPath("objectpath_key_1"):
1948
 
            ObjectPath("objectpath_value_1"),
1949
 
            ObjectPath("objectpath_key_2"):
1950
 
            ObjectPath("objectpath_value_2"),
1951
 
        }
1952
 
        ret = self.call_method(bus, "methodname", "busname",
1953
 
                               "objectpath", "interface")
1954
 
        expected_method_return = {str(key): str(value)
1955
 
                                  for key, value in
1956
 
                                  method_return.items()}
1957
 
        for key, value in ret.items():
1958
 
            self.assertNotIsInstance(key, ObjectPath)
1959
 
            self.assertNotIsInstance(value, ObjectPath)
1960
 
        self.assertEqual(expected_method_return, ret)
1961
 
        self.assertIsInstance(ret, dict)
1962
 
 
1963
 
    def test_call_method_filters_objectpaths_in_dict_in_dict(self):
1964
 
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
1965
 
        def func():
1966
 
            return method_return
1967
 
        fake_ravel = self.fake_ravel_func(func)
1968
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
1969
 
        method_return = {
1970
 
            ObjectPath("key1"): {
1971
 
                ObjectPath("key11"): ObjectPath("value11"),
1972
 
                ObjectPath("key12"): ObjectPath("value12"),
1973
 
            },
1974
 
            ObjectPath("key2"): {
1975
 
                ObjectPath("key21"): ObjectPath("value21"),
1976
 
                ObjectPath("key22"): ObjectPath("value22"),
1977
 
            },
1978
 
        }
1979
 
        ret = self.call_method(bus, "methodname", "busname",
1980
 
                               "objectpath", "interface")
1981
 
        expected_method_return = {
1982
 
            "key1": {"key11": "value11",
1983
 
                     "key12": "value12"},
1984
 
            "key2": {"key21": "value21",
1985
 
                     "key22": "value22"},
1986
 
        }
1987
 
        self.assertEqual(expected_method_return, ret)
1988
 
        for key, value in ret.items():
1989
 
            self.assertIsInstance(value, dict)
1990
 
            self.assertEqual(expected_method_return[key], value)
1991
 
            self.assertNotIsInstance(key, ObjectPath)
1992
 
            for inner_key, inner_value in value.items():
1993
 
                self.assertIsInstance(value, dict)
1994
 
                self.assertEqual(
1995
 
                    expected_method_return[key][inner_key],
1996
 
                    inner_value)
1997
 
                self.assertNotIsInstance(key, ObjectPath)
1998
 
 
1999
 
    def test_call_method_filters_objectpaths_in_dict_three_deep(self):
2000
 
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
2001
 
        def func():
2002
 
            return method_return
2003
 
        fake_ravel = self.fake_ravel_func(func)
2004
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2005
 
        method_return = {
2006
 
            ObjectPath("key1"): {
2007
 
                ObjectPath("key2"): {
2008
 
                    ObjectPath("key3"): ObjectPath("value"),
2009
 
                },
2010
 
            },
2011
 
        }
2012
 
        ret = self.call_method(bus, "methodname", "busname",
2013
 
                               "objectpath", "interface")
2014
 
        expected_method_return = {"key1": {"key2": {"key3": "value"}}}
2015
 
        self.assertEqual(expected_method_return, ret)
2016
 
        self.assertIsInstance(ret, dict)
2017
 
        self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
2018
 
        self.assertIsInstance(ret["key1"], dict)
2019
 
        self.assertNotIsInstance(next(iter(ret["key1"].keys())),
2020
 
                                 ObjectPath)
2021
 
        self.assertIsInstance(ret["key1"]["key2"], dict)
2022
 
        self.assertNotIsInstance(
2023
 
            next(iter(ret["key1"]["key2"].keys())),
2024
 
            ObjectPath)
2025
 
        self.assertEqual("value", ret["key1"]["key2"]["key3"])
2026
 
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
2027
 
                                 self.dummy_dbussy.DBUS.ObjectPath)
2028
 
 
2029
 
    def test_call_method_handles_exception(self):
2030
 
        def func():
2031
 
            raise self.dummy_dbussy.DBusError()
2032
 
 
2033
 
        fake_ravel = self.fake_ravel_func(func)
2034
 
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
2035
 
 
2036
 
        with self.assertRaises(dbus.Error) as e:
2037
 
            self.call_method(bus, "methodname", "busname",
2038
 
                             "objectpath", "interface")
2039
 
 
2040
 
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
2041
 
 
2042
 
    def test_get_object_converts_to_correct_exception(self):
2043
 
        class fake_ravel_raises_exception_on_connect:
2044
 
            @staticmethod
2045
 
            def system_bus():
2046
 
                class Bus:
2047
 
                    @staticmethod
2048
 
                    def __getitem__(key):
2049
 
                        if key == "objectpath":
2050
 
                            raise self.dummy_dbussy.DBusError()
2051
 
                        raise Exception(key)
2052
 
                return {"busname": Bus()}
2053
 
        def func():
2054
 
            raise self.dummy_dbussy.DBusError()
2055
 
        bus = dbussy_adapter.SystemBus(
2056
 
            self.dummy_dbussy,
2057
 
            fake_ravel_raises_exception_on_connect)
2058
 
        with self.assertRaises(dbus.ConnectFailed):
2059
 
            self.call_method(bus, "methodname", "busname",
2060
 
                             "objectpath", "interface")
2061
 
 
2062
 
 
2063
1738
class Test_commands_from_options(unittest.TestCase):
2064
1739
 
2065
1740
    def setUp(self):
2070
1745
        self.assert_command_from_args(["--is-enabled", "client"],
2071
1746
                                      command.IsEnabled)
2072
1747
 
2073
 
    def assert_command_from_args(self, args, command_cls, length=1,
2074
 
                                 clients=None, **cmd_attrs):
 
1748
    def assert_command_from_args(self, args, command_cls,
 
1749
                                 **cmd_attrs):
2075
1750
        """Assert that parsing ARGS should result in an instance of
2076
1751
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
2077
1752
        options = self.parser.parse_args(args)
2078
1753
        check_option_syntax(self.parser, options)
2079
1754
        commands = commands_from_options(options)
2080
 
        self.assertEqual(length, len(commands))
2081
 
        for command in commands:
2082
 
            if isinstance(command, command_cls):
2083
 
                break
2084
 
        else:
2085
 
            self.assertIsInstance(command, command_cls)
2086
 
        if clients is not None:
2087
 
            self.assertEqual(clients, options.client)
 
1755
        self.assertEqual(1, len(commands))
 
1756
        command = commands[0]
 
1757
        self.assertIsInstance(command, command_cls)
2088
1758
        for key, value in cmd_attrs.items():
2089
1759
            self.assertEqual(value, getattr(command, key))
2090
1760
 
2091
 
    def assert_commands_from_args(self, args, commands, clients=None):
2092
 
        for cmd in commands:
2093
 
            self.assert_command_from_args(args, cmd,
2094
 
                                          length=len(commands),
2095
 
                                          clients=clients)
2096
 
 
2097
1761
    def test_is_enabled_short(self):
2098
1762
        self.assert_command_from_args(["-V", "client"],
2099
1763
                                      command.IsEnabled)
2290
1954
                                      verbose=True)
2291
1955
 
2292
1956
 
2293
 
    def test_manual_page_example_1(self):
2294
 
        self.assert_command_from_args("",
2295
 
                                      command.PrintTable,
2296
 
                                      clients=[],
2297
 
                                      verbose=False)
2298
 
 
2299
 
    def test_manual_page_example_2(self):
2300
 
        self.assert_command_from_args(
2301
 
            "--verbose foo1.example.org foo2.example.org".split(),
2302
 
            command.PrintTable, clients=["foo1.example.org",
2303
 
                                         "foo2.example.org"],
2304
 
            verbose=True)
2305
 
 
2306
 
    def test_manual_page_example_3(self):
2307
 
        self.assert_command_from_args("--enable --all".split(),
2308
 
                                      command.Enable,
2309
 
                                      clients=[])
2310
 
 
2311
 
    def test_manual_page_example_4(self):
2312
 
        self.assert_commands_from_args(
2313
 
            ("--timeout=PT5M --interval=PT1M foo1.example.org"
2314
 
             " foo2.example.org").split(),
2315
 
            [command.SetTimeout, command.SetInterval],
2316
 
            clients=["foo1.example.org", "foo2.example.org"])
2317
 
 
2318
 
    def test_manual_page_example_5(self):
2319
 
        self.assert_command_from_args("--approve --all".split(),
2320
 
                                      command.Approve,
2321
 
                                      clients=[])
2322
 
 
2323
 
 
2324
1957
class TestCommand(unittest.TestCase):
2325
1958
    """Abstract class for tests of command classes"""
2326
1959