/mandos/trunk

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

  • Committer: Teddy Hogeborn
  • Date: 2021-02-03 08:33:43 UTC
  • Revision ID: teddy@recompile.se-20210203083343-spdbhxgd5v8e3lv9
Update copyright year

* debian/copyright: Update copyright year to 2021.
* dracut-module/password-agent.c: - '' -
* dracut-module/password-agent.xml: Update copyright year to 2020.
* intro.xml: - '' -
* plugin-helpers/mandos-client-iprouteadddel.c: Update copyright year
  to 2021.
* plugin-runner.c: - '' -
* plugins.d/askpass-fifo.c: - '' -
* plugins.d/mandos-client.c: - '' -
* plugins.d/password-prompt.c: - '' -
* plugins.d/plymouth.c: - '' -
* plugins.d/splashy.c: - '' -
* plugins.d/usplash.c: - '' -

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
1
 
#!/usr/bin/python
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 --test" (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 Monitor - Control and monitor the Mandos server
5
 
#
6
 
# Copyright © 2008-2019 Teddy Hogeborn
7
 
# Copyright © 2008-2019 Björn Påhlsson
 
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
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
 
49
64
try:
50
 
    import pydbus
51
 
    import gi
52
 
    dbus_python = None
 
65
    import dbussy
 
66
    import ravel
53
67
except ImportError:
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
 
68
    try:
 
69
        import pydbus
 
70
        import gi
 
71
    except ImportError:
 
72
        import dbus as dbus_python
 
73
 
62
74
 
63
75
# Show warnings by default
64
76
if not sys.warnoptions:
72
84
logging.captureWarnings(True)   # Show warnings via the logging system
73
85
 
74
86
if sys.version_info.major == 2:
75
 
    str = unicode
76
87
    import StringIO
77
88
    io.StringIO = StringIO.StringIO
78
89
 
79
90
locale.setlocale(locale.LC_ALL, "")
80
91
 
81
 
version = "1.8.4"
 
92
version = "1.8.13"
82
93
 
83
94
 
84
95
def main():
93
104
    if options.debug:
94
105
        log.setLevel(logging.DEBUG)
95
106
 
96
 
    if pydbus is not None:
 
107
    if dbussy is not None and ravel is not None:
 
108
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
 
109
    elif pydbus is not None:
97
110
        bus = pydbus_adapter.CachingBus(pydbus)
98
111
    else:
99
112
        bus = dbus_python_adapter.CachingBus(dbus_python)
250
263
def rfc3339_duration_to_delta(duration):
251
264
    """Parse an RFC 3339 "duration" and return a datetime.timedelta
252
265
 
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)
 
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
269
283
    >>> # Can not be empty:
270
284
    >>> rfc3339_duration_to_delta("")
271
285
    Traceback (most recent call last):
381
395
    """Parse an interval string as documented by Mandos before 1.6.1,
382
396
    and return a datetime.timedelta
383
397
 
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)
 
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
398
412
    >>> # Ignore unknown characters, allow any order and repetitions
399
 
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
400
 
    datetime.timedelta(2, 480, 18000)
 
413
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
 
414
    True
401
415
 
402
416
    """
403
417
 
466
480
        parser.error("--remove can only be combined with --deny")
467
481
 
468
482
 
469
 
class dbus(object):
 
483
class dbus:
470
484
 
471
 
    class SystemBus(object):
 
485
    class SystemBus:
472
486
 
473
487
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
474
488
        def get_managed_objects(self, busname, objectpath):
483
497
                             self.properties_iface, interface, key,
484
498
                             value)
485
499
 
 
500
        def call_method(self, methodname, busname, objectpath,
 
501
                        interface, *args):
 
502
            raise NotImplementedError()
 
503
 
486
504
 
487
505
    class MandosBus(SystemBus):
488
506
        busname_domain = "se.recompile"
520
538
        pass
521
539
 
522
540
 
523
 
class dbus_python_adapter(object):
 
541
class dbus_python_adapter:
524
542
 
525
543
    class SystemBus(dbus.MandosBus):
526
544
        """Use dbus-python"""
580
598
                                     self.client_interface, key,
581
599
                                     value)
582
600
 
583
 
    class SilenceLogger(object):
 
601
    class SilenceLogger:
584
602
        "Simple context manager to silence a particular logger"
585
603
        def __init__(self, loggername):
586
604
            self.logger = logging.getLogger(loggername)
615
633
                return new_object
616
634
 
617
635
 
618
 
class pydbus_adapter(object):
 
636
class pydbus_adapter:
619
637
    class SystemBus(dbus.MandosBus):
620
638
        def __init__(self, module=pydbus):
621
639
            self.pydbus = module
678
696
                return new_object
679
697
 
680
698
 
 
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
 
681
789
def commands_from_options(options):
682
790
 
683
791
    commands = list(options.commands)
711
819
    return commands
712
820
 
713
821
 
714
 
class command(object):
 
822
class command:
715
823
    """A namespace for command classes"""
716
824
 
717
 
    class Base(object):
 
825
    class Base:
718
826
        """Abstract base class for commands"""
719
827
        def run(self, clients, bus=None):
720
828
            """Normal commands should implement run_on_one_client(),
783
891
                keywords = self.all_keywords
784
892
            print(self.TableOfClients(clients.values(), keywords))
785
893
 
786
 
        class TableOfClients(object):
 
894
        class TableOfClients:
787
895
            tableheaders = {
788
896
                "Name": "Name",
789
897
                "Enabled": "Enabled",
1020
1128
                                                     "output"))
1021
1129
 
1022
1130
 
1023
 
class Unique(object):
 
1131
class Unique:
1024
1132
    """Class for objects which exist only to be unique objects, since
1025
1133
unittest.mock.sentinel only exists in Python 3.3"""
1026
1134
 
1310
1418
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1311
1419
 
1312
1420
    def MockDBusPython_func(self, func):
1313
 
        class mock_dbus_python(object):
 
1421
        class mock_dbus_python:
1314
1422
            """mock dbus-python module"""
1315
 
            class exceptions(object):
 
1423
            class exceptions:
1316
1424
                """Pseudo-namespace"""
1317
1425
                class DBusException(Exception):
1318
1426
                    pass
1319
 
            class SystemBus(object):
 
1427
            class SystemBus:
1320
1428
                @staticmethod
1321
1429
                def get_object(busname, objectpath):
1322
1430
                    DBusObject = collections.namedtuple(
1336
1444
                                    dbus_interface=dbus_interface)
1337
1445
                    return DBusObject(methodname=method,
1338
1446
                                      Set=set_property)
1339
 
            class Boolean(object):
 
1447
            class Boolean:
1340
1448
                def __init__(self, value):
1341
1449
                    self.value = bool(value)
1342
1450
                def __bool__(self):
1526
1634
        finally:
1527
1635
            dbus_logger.removeFilter(counting_handler)
1528
1636
 
1529
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1637
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1530
1638
 
1531
1639
        # Make sure the dbus logger was suppressed
1532
1640
        self.assertEqual(0, counting_handler.count)
1554
1662
            self.call_method(bus, "methodname", "busname",
1555
1663
                             "objectpath", "interface")
1556
1664
 
1557
 
    class fake_dbus_python_raises_exception_on_connect(object):
 
1665
    class fake_dbus_python_raises_exception_on_connect:
1558
1666
        """fake dbus-python module"""
1559
 
        class exceptions(object):
 
1667
        class exceptions:
1560
1668
            """Pseudo-namespace"""
1561
1669
            class DBusException(Exception):
1562
1670
                pass
1570
1678
 
1571
1679
 
1572
1680
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1573
 
    class mock_dbus_python(object):
 
1681
    class mock_dbus_python:
1574
1682
        """mock dbus-python modules"""
1575
 
        class SystemBus(object):
 
1683
        class SystemBus:
1576
1684
            @staticmethod
1577
1685
            def get_object(busname, objectpath):
1578
1686
                return Unique()
1624
1732
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1625
1733
 
1626
1734
    def Stub_pydbus_func(self, func):
1627
 
        class stub_pydbus(object):
 
1735
        class stub_pydbus:
1628
1736
            """stub pydbus module"""
1629
 
            class SystemBus(object):
 
1737
            class SystemBus:
1630
1738
                @staticmethod
1631
1739
                def get(busname, objectpath):
1632
1740
                    DBusObject = collections.namedtuple(
1669
1777
            self.call_method(bus, "methodname", "busname",
1670
1778
                             "objectpath", "interface")
1671
1779
 
1672
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1780
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1673
1781
 
1674
1782
    def test_get_converts_to_correct_exception(self):
1675
1783
        bus = pydbus_adapter.SystemBus(
1678
1786
            self.call_method(bus, "methodname", "busname",
1679
1787
                             "objectpath", "interface")
1680
1788
 
1681
 
    class fake_pydbus_raises_exception_on_connect(object):
 
1789
    class fake_pydbus_raises_exception_on_connect:
1682
1790
        """fake dbus-python module"""
1683
1791
        @classmethod
1684
1792
        def SystemBus(cls):
1688
1796
            return Bus(get=get)
1689
1797
 
1690
1798
    def test_set_property_uses_setattr(self):
1691
 
        class Object(object):
 
1799
        class Object:
1692
1800
            pass
1693
1801
        obj = Object()
1694
 
        class pydbus_spy(object):
1695
 
            class SystemBus(object):
 
1802
        class pydbus_spy:
 
1803
            class SystemBus:
1696
1804
                @staticmethod
1697
1805
                def get(busname, objectpath):
1698
1806
                    return {"interface": obj}
1705
1813
    def test_get_suppresses_xml_deprecation_warning(self):
1706
1814
        if sys.version_info.major >= 3:
1707
1815
            return
1708
 
        class stub_pydbus_get(object):
1709
 
            class SystemBus(object):
 
1816
        class stub_pydbus_get:
 
1817
            class SystemBus:
1710
1818
                @staticmethod
1711
1819
                def get(busname, objectpath):
1712
1820
                    warnings.warn_explicit(
1720
1828
 
1721
1829
 
1722
1830
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1723
 
    class stub_pydbus(object):
 
1831
    class stub_pydbus:
1724
1832
        """stub pydbus module"""
1725
 
        class SystemBus(object):
 
1833
        class SystemBus:
1726
1834
            @staticmethod
1727
1835
            def get(busname, objectpath):
1728
1836
                return Unique()
1770
1878
        self.assertIs(obj1, obj1b)
1771
1879
 
1772
1880
 
 
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
 
1773
2063
class Test_commands_from_options(unittest.TestCase):
1774
2064
 
1775
2065
    def setUp(self):
2001
2291
 
2002
2292
 
2003
2293
    def test_manual_page_example_1(self):
2004
 
        self.assert_command_from_args("--verbose".split(),
 
2294
        self.assert_command_from_args("",
2005
2295
                                      command.PrintTable,
2006
2296
                                      clients=[],
2007
 
                                      verbose=True)
 
2297
                                      verbose=False)
2008
2298
 
2009
2299
    def test_manual_page_example_2(self):
2010
2300
        self.assert_command_from_args(