/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 at recompile
  • Date: 2020-07-04 08:59:37 UTC
  • Revision ID: teddy@recompile.se-20200704085937-r495zb32nfng9lnt
In initramfs-tools boots, only use setsid when available

* initramfs-tools-script: If available, use setsid to start
  mandos-to-cryptroot-unlock.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
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*")))); -*-
 
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
3
#
4
4
# Mandos Monitor - Control and monitor the Mandos server
5
5
#
46
46
import tempfile
47
47
import contextlib
48
48
 
49
 
import dbus as dbus_python
 
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
try:
 
65
    import dbussy
 
66
    import ravel
 
67
except ImportError:
 
68
    try:
 
69
        import pydbus
 
70
        import gi
 
71
    except ImportError:
 
72
        import dbus as dbus_python
 
73
 
50
74
 
51
75
# Show warnings by default
52
76
if not sys.warnoptions:
60
84
logging.captureWarnings(True)   # Show warnings via the logging system
61
85
 
62
86
if sys.version_info.major == 2:
63
 
    str = unicode
64
87
    import StringIO
65
88
    io.StringIO = StringIO.StringIO
66
89
 
67
90
locale.setlocale(locale.LC_ALL, "")
68
91
 
69
 
version = "1.8.3"
 
92
version = "1.8.11"
70
93
 
71
94
 
72
95
def main():
81
104
    if options.debug:
82
105
        log.setLevel(logging.DEBUG)
83
106
 
84
 
    bus = dbus_python_adapter.CachingBus(dbus_python)
 
107
    if dbussy is not None and ravel is not None:
 
108
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
 
109
    elif pydbus is not None:
 
110
        bus = pydbus_adapter.CachingBus(pydbus)
 
111
    else:
 
112
        bus = dbus_python_adapter.CachingBus(dbus_python)
85
113
 
86
114
    try:
87
115
        all_clients = bus.get_clients_and_properties()
235
263
def rfc3339_duration_to_delta(duration):
236
264
    """Parse an RFC 3339 "duration" and return a datetime.timedelta
237
265
 
238
 
    >>> rfc3339_duration_to_delta("P7D")
239
 
    datetime.timedelta(7)
240
 
    >>> rfc3339_duration_to_delta("PT60S")
241
 
    datetime.timedelta(0, 60)
242
 
    >>> rfc3339_duration_to_delta("PT60M")
243
 
    datetime.timedelta(0, 3600)
244
 
    >>> rfc3339_duration_to_delta("P60M")
245
 
    datetime.timedelta(1680)
246
 
    >>> rfc3339_duration_to_delta("PT24H")
247
 
    datetime.timedelta(1)
248
 
    >>> rfc3339_duration_to_delta("P1W")
249
 
    datetime.timedelta(7)
250
 
    >>> rfc3339_duration_to_delta("PT5M30S")
251
 
    datetime.timedelta(0, 330)
252
 
    >>> rfc3339_duration_to_delta("P1DT3M20S")
253
 
    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
254
283
    >>> # Can not be empty:
255
284
    >>> rfc3339_duration_to_delta("")
256
285
    Traceback (most recent call last):
366
395
    """Parse an interval string as documented by Mandos before 1.6.1,
367
396
    and return a datetime.timedelta
368
397
 
369
 
    >>> parse_pre_1_6_1_interval('7d')
370
 
    datetime.timedelta(7)
371
 
    >>> parse_pre_1_6_1_interval('60s')
372
 
    datetime.timedelta(0, 60)
373
 
    >>> parse_pre_1_6_1_interval('60m')
374
 
    datetime.timedelta(0, 3600)
375
 
    >>> parse_pre_1_6_1_interval('24h')
376
 
    datetime.timedelta(1)
377
 
    >>> parse_pre_1_6_1_interval('1w')
378
 
    datetime.timedelta(7)
379
 
    >>> parse_pre_1_6_1_interval('5m 30s')
380
 
    datetime.timedelta(0, 330)
381
 
    >>> parse_pre_1_6_1_interval('')
382
 
    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
383
412
    >>> # Ignore unknown characters, allow any order and repetitions
384
 
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
385
 
    datetime.timedelta(2, 480, 18000)
 
413
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
 
414
    True
386
415
 
387
416
    """
388
417
 
451
480
        parser.error("--remove can only be combined with --deny")
452
481
 
453
482
 
454
 
class dbus(object):
 
483
class dbus:
455
484
 
456
 
    class SystemBus(object):
 
485
    class SystemBus:
457
486
 
458
487
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
459
488
        def get_managed_objects(self, busname, objectpath):
468
497
                             self.properties_iface, interface, key,
469
498
                             value)
470
499
 
 
500
        def call_method(self, methodname, busname, objectpath,
 
501
                        interface, *args):
 
502
            raise NotImplementedError()
 
503
 
471
504
 
472
505
    class MandosBus(SystemBus):
473
506
        busname_domain = "se.recompile"
505
538
        pass
506
539
 
507
540
 
508
 
class dbus_python_adapter(object):
 
541
class dbus_python_adapter:
509
542
 
510
543
    class SystemBus(dbus.MandosBus):
511
544
        """Use dbus-python"""
556
589
                        for key, subval in value.items()}
557
590
            return value
558
591
 
 
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)
559
600
 
560
 
    class SilenceLogger(object):
 
601
    class SilenceLogger:
561
602
        "Simple context manager to silence a particular logger"
562
603
        def __init__(self, loggername):
563
604
            self.logger = logging.getLogger(loggername)
592
633
                return new_object
593
634
 
594
635
 
 
636
class pydbus_adapter:
 
637
    class SystemBus(dbus.MandosBus):
 
638
        def __init__(self, module=pydbus):
 
639
            self.pydbus = module
 
640
            self.bus = self.pydbus.SystemBus()
 
641
 
 
642
        @contextlib.contextmanager
 
643
        def convert_exception(self, exception_class=dbus.Error):
 
644
            try:
 
645
                yield
 
646
            except gi.repository.GLib.Error as e:
 
647
                # This does what "raise from" would do
 
648
                exc = exception_class(*e.args)
 
649
                exc.__cause__ = e
 
650
                raise exc
 
651
 
 
652
        def call_method(self, methodname, busname, objectpath,
 
653
                        interface, *args):
 
654
            proxy_object = self.get(busname, objectpath)
 
655
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
656
                      interface, methodname,
 
657
                      ", ".join(repr(a) for a in args))
 
658
            method = getattr(proxy_object[interface], methodname)
 
659
            with self.convert_exception():
 
660
                return method(*args)
 
661
 
 
662
        def get(self, busname, objectpath):
 
663
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
664
                      busname, objectpath)
 
665
            with self.convert_exception(dbus.ConnectFailed):
 
666
                if sys.version_info.major <= 2:
 
667
                    with warnings.catch_warnings():
 
668
                        warnings.filterwarnings(
 
669
                            "ignore", "", DeprecationWarning,
 
670
                            r"^xml\.etree\.ElementTree$")
 
671
                        return self.bus.get(busname, objectpath)
 
672
                else:
 
673
                    return self.bus.get(busname, objectpath)
 
674
 
 
675
        def set_property(self, busname, objectpath, interface, key,
 
676
                         value):
 
677
            proxy_object = self.get(busname, objectpath)
 
678
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
 
679
                      objectpath, self.properties_iface, interface,
 
680
                      key, value)
 
681
            setattr(proxy_object[interface], key, value)
 
682
 
 
683
    class CachingBus(SystemBus):
 
684
        """A caching layer for pydbus_adapter.SystemBus"""
 
685
        def __init__(self, *args, **kwargs):
 
686
            self.object_cache = {}
 
687
            super(pydbus_adapter.CachingBus,
 
688
                  self).__init__(*args, **kwargs)
 
689
        def get(self, busname, objectpath):
 
690
            try:
 
691
                return self.object_cache[(busname, objectpath)]
 
692
            except KeyError:
 
693
                new_object = (super(pydbus_adapter.CachingBus, self)
 
694
                              .get(busname, objectpath))
 
695
                self.object_cache[(busname, objectpath)]  = new_object
 
696
                return new_object
 
697
 
 
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
            # tuple: (signature, 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
            if isinstance(value, tuple):
 
742
                if (len(value) == 2
 
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):
 
747
                return str(value)
 
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()}
 
753
            return value
 
754
 
 
755
        def set_property(self, busname, objectpath, interface, key,
 
756
                         value):
 
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,
 
760
                      key, value)
 
761
            if key == "Secret":
 
762
                # DBussy wants a Byte Array to be a sequence of
 
763
                # values, not a byte string
 
764
                value = tuple(value)
 
765
            setattr(proxy_object.get_interface(interface), key, value)
 
766
 
 
767
    class MandosBus(SystemBus, dbus.MandosBus):
 
768
        pass
 
769
 
 
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,
 
775
                                                            **kwargs)
 
776
        def get_object(self, busname, objectpath):
 
777
            try:
 
778
                return self.object_cache[(busname, objectpath)]
 
779
            except KeyError:
 
780
                new_object = super(
 
781
                    dbussy_adapter.CachingBus,
 
782
                    self).get_object(busname, objectpath)
 
783
                self.object_cache[(busname, objectpath)]  = new_object
 
784
                return new_object
 
785
 
 
786
 
595
787
def commands_from_options(options):
596
788
 
597
789
    commands = list(options.commands)
625
817
    return commands
626
818
 
627
819
 
628
 
class command(object):
 
820
class command:
629
821
    """A namespace for command classes"""
630
822
 
631
 
    class Base(object):
 
823
    class Base:
632
824
        """Abstract base class for commands"""
633
825
        def run(self, clients, bus=None):
634
826
            """Normal commands should implement run_on_one_client(),
697
889
                keywords = self.all_keywords
698
890
            print(self.TableOfClients(clients.values(), keywords))
699
891
 
700
 
        class TableOfClients(object):
 
892
        class TableOfClients:
701
893
            tableheaders = {
702
894
                "Name": "Name",
703
895
                "Enabled": "Enabled",
934
1126
                                                     "output"))
935
1127
 
936
1128
 
937
 
class Unique(object):
 
1129
class Unique:
938
1130
    """Class for objects which exist only to be unique objects, since
939
1131
unittest.mock.sentinel only exists in Python 3.3"""
940
1132
 
1224
1416
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1225
1417
 
1226
1418
    def MockDBusPython_func(self, func):
1227
 
        class mock_dbus_python(object):
 
1419
        class mock_dbus_python:
1228
1420
            """mock dbus-python module"""
1229
 
            class exceptions(object):
 
1421
            class exceptions:
1230
1422
                """Pseudo-namespace"""
1231
1423
                class DBusException(Exception):
1232
1424
                    pass
1233
 
            class SystemBus(object):
 
1425
            class SystemBus:
1234
1426
                @staticmethod
1235
1427
                def get_object(busname, objectpath):
1236
1428
                    DBusObject = collections.namedtuple(
1237
 
                        "DBusObject", ("methodname",))
 
1429
                        "DBusObject", ("methodname", "Set"))
1238
1430
                    def method(*args, **kwargs):
1239
1431
                        self.assertEqual({"dbus_interface":
1240
1432
                                          "interface"},
1241
1433
                                         kwargs)
1242
1434
                        return func(*args)
1243
 
                    return DBusObject(methodname=method)
1244
 
            class Boolean(object):
 
1435
                    def set_property(interface, key, value,
 
1436
                                     dbus_interface=None):
 
1437
                        self.assertEqual(
 
1438
                            "org.freedesktop.DBus.Properties",
 
1439
                            dbus_interface)
 
1440
                        self.assertEqual("Secret", key)
 
1441
                        return func(interface, key, value,
 
1442
                                    dbus_interface=dbus_interface)
 
1443
                    return DBusObject(methodname=method,
 
1444
                                      Set=set_property)
 
1445
            class Boolean:
1245
1446
                def __init__(self, value):
1246
1447
                    self.value = bool(value)
1247
1448
                def __bool__(self):
1252
1453
                pass
1253
1454
            class Dictionary(dict):
1254
1455
                pass
 
1456
            class ByteArray(bytes):
 
1457
                pass
1255
1458
        return mock_dbus_python
1256
1459
 
1257
1460
    def call_method(self, bus, methodname, busname, objectpath,
1429
1632
        finally:
1430
1633
            dbus_logger.removeFilter(counting_handler)
1431
1634
 
1432
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1635
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1433
1636
 
1434
1637
        # Make sure the dbus logger was suppressed
1435
1638
        self.assertEqual(0, counting_handler.count)
1436
1639
 
 
1640
    def test_Set_Secret_sends_bytearray(self):
 
1641
        ret = [None]
 
1642
        def func(*args, **kwargs):
 
1643
            ret[0] = (args, kwargs)
 
1644
        mock_dbus_python = self.MockDBusPython_func(func)
 
1645
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
 
1646
        bus.set_client_property("objectpath", "Secret", "value")
 
1647
        expected_call = (("se.recompile.Mandos.Client", "Secret",
 
1648
                          mock_dbus_python.ByteArray(b"value")),
 
1649
                         {"dbus_interface":
 
1650
                          "org.freedesktop.DBus.Properties"})
 
1651
        self.assertEqual(expected_call, ret[0])
 
1652
        if sys.version_info.major == 2:
 
1653
            self.assertIsInstance(ret[0][0][-1],
 
1654
                                  mock_dbus_python.ByteArray)
 
1655
 
1437
1656
    def test_get_object_converts_to_correct_exception(self):
1438
1657
        bus = dbus_python_adapter.SystemBus(
1439
1658
            self.fake_dbus_python_raises_exception_on_connect)
1441
1660
            self.call_method(bus, "methodname", "busname",
1442
1661
                             "objectpath", "interface")
1443
1662
 
1444
 
    class fake_dbus_python_raises_exception_on_connect(object):
 
1663
    class fake_dbus_python_raises_exception_on_connect:
1445
1664
        """fake dbus-python module"""
1446
 
        class exceptions(object):
 
1665
        class exceptions:
1447
1666
            """Pseudo-namespace"""
1448
1667
            class DBusException(Exception):
1449
1668
                pass
1457
1676
 
1458
1677
 
1459
1678
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1460
 
    class mock_dbus_python(object):
 
1679
    class mock_dbus_python:
1461
1680
        """mock dbus-python modules"""
1462
 
        class SystemBus(object):
 
1681
        class SystemBus:
1463
1682
            @staticmethod
1464
1683
            def get_object(busname, objectpath):
1465
1684
                return Unique()
1508
1727
        self.assertIs(obj1, obj1b)
1509
1728
 
1510
1729
 
 
1730
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
 
1731
 
 
1732
    def Stub_pydbus_func(self, func):
 
1733
        class stub_pydbus:
 
1734
            """stub pydbus module"""
 
1735
            class SystemBus:
 
1736
                @staticmethod
 
1737
                def get(busname, objectpath):
 
1738
                    DBusObject = collections.namedtuple(
 
1739
                        "DBusObject", ("methodname",))
 
1740
                    return {"interface":
 
1741
                            DBusObject(methodname=func)}
 
1742
        return stub_pydbus
 
1743
 
 
1744
    def call_method(self, bus, methodname, busname, objectpath,
 
1745
                    interface, *args):
 
1746
        with self.assertLogs(log, logging.DEBUG):
 
1747
            return bus.call_method(methodname, busname, objectpath,
 
1748
                                   interface, *args)
 
1749
 
 
1750
    def test_call_method_returns(self):
 
1751
        expected_method_return = Unique()
 
1752
        method_args = (Unique(), Unique())
 
1753
        def func(*args):
 
1754
            self.assertEqual(len(method_args), len(args))
 
1755
            for marg, arg in zip(method_args, args):
 
1756
                self.assertIs(marg, arg)
 
1757
            return expected_method_return
 
1758
        stub_pydbus = self.Stub_pydbus_func(func)
 
1759
        bus = pydbus_adapter.SystemBus(stub_pydbus)
 
1760
        ret = self.call_method(bus, "methodname", "busname",
 
1761
                               "objectpath", "interface",
 
1762
                               *method_args)
 
1763
        self.assertIs(ret, expected_method_return)
 
1764
 
 
1765
    def test_call_method_handles_exception(self):
 
1766
        dbus_logger = logging.getLogger("dbus.proxies")
 
1767
 
 
1768
        def func():
 
1769
            raise gi.repository.GLib.Error()
 
1770
 
 
1771
        stub_pydbus = self.Stub_pydbus_func(func)
 
1772
        bus = pydbus_adapter.SystemBus(stub_pydbus)
 
1773
 
 
1774
        with self.assertRaises(dbus.Error) as e:
 
1775
            self.call_method(bus, "methodname", "busname",
 
1776
                             "objectpath", "interface")
 
1777
 
 
1778
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
1779
 
 
1780
    def test_get_converts_to_correct_exception(self):
 
1781
        bus = pydbus_adapter.SystemBus(
 
1782
            self.fake_pydbus_raises_exception_on_connect)
 
1783
        with self.assertRaises(dbus.ConnectFailed):
 
1784
            self.call_method(bus, "methodname", "busname",
 
1785
                             "objectpath", "interface")
 
1786
 
 
1787
    class fake_pydbus_raises_exception_on_connect:
 
1788
        """fake dbus-python module"""
 
1789
        @classmethod
 
1790
        def SystemBus(cls):
 
1791
            def get(busname, objectpath):
 
1792
                raise gi.repository.GLib.Error()
 
1793
            Bus = collections.namedtuple("Bus", ["get"])
 
1794
            return Bus(get=get)
 
1795
 
 
1796
    def test_set_property_uses_setattr(self):
 
1797
        class Object:
 
1798
            pass
 
1799
        obj = Object()
 
1800
        class pydbus_spy:
 
1801
            class SystemBus:
 
1802
                @staticmethod
 
1803
                def get(busname, objectpath):
 
1804
                    return {"interface": obj}
 
1805
        bus = pydbus_adapter.SystemBus(pydbus_spy)
 
1806
        value = Unique()
 
1807
        bus.set_property("busname", "objectpath", "interface", "key",
 
1808
                         value)
 
1809
        self.assertIs(value, obj.key)
 
1810
 
 
1811
    def test_get_suppresses_xml_deprecation_warning(self):
 
1812
        if sys.version_info.major >= 3:
 
1813
            return
 
1814
        class stub_pydbus_get:
 
1815
            class SystemBus:
 
1816
                @staticmethod
 
1817
                def get(busname, objectpath):
 
1818
                    warnings.warn_explicit(
 
1819
                        "deprecated", DeprecationWarning,
 
1820
                        "xml.etree.ElementTree", 0)
 
1821
        bus = pydbus_adapter.SystemBus(stub_pydbus_get)
 
1822
        with warnings.catch_warnings(record=True) as w:
 
1823
            warnings.simplefilter("always")
 
1824
            bus.get("busname", "objectpath")
 
1825
            self.assertEqual(0, len(w))
 
1826
 
 
1827
 
 
1828
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
 
1829
    class stub_pydbus:
 
1830
        """stub pydbus module"""
 
1831
        class SystemBus:
 
1832
            @staticmethod
 
1833
            def get(busname, objectpath):
 
1834
                return Unique()
 
1835
 
 
1836
    def setUp(self):
 
1837
        self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
 
1838
 
 
1839
    def test_returns_distinct_objectpaths(self):
 
1840
        obj1 = self.bus.get("busname", "objectpath1")
 
1841
        self.assertIsInstance(obj1, Unique)
 
1842
        obj2 = self.bus.get("busname", "objectpath2")
 
1843
        self.assertIsInstance(obj2, Unique)
 
1844
        self.assertIsNot(obj1, obj2)
 
1845
 
 
1846
    def test_returns_distinct_busnames(self):
 
1847
        obj1 = self.bus.get("busname1", "objectpath")
 
1848
        self.assertIsInstance(obj1, Unique)
 
1849
        obj2 = self.bus.get("busname2", "objectpath")
 
1850
        self.assertIsInstance(obj2, Unique)
 
1851
        self.assertIsNot(obj1, obj2)
 
1852
 
 
1853
    def test_returns_distinct_both(self):
 
1854
        obj1 = self.bus.get("busname1", "objectpath")
 
1855
        self.assertIsInstance(obj1, Unique)
 
1856
        obj2 = self.bus.get("busname2", "objectpath")
 
1857
        self.assertIsInstance(obj2, Unique)
 
1858
        self.assertIsNot(obj1, obj2)
 
1859
 
 
1860
    def test_returns_same(self):
 
1861
        obj1 = self.bus.get("busname", "objectpath")
 
1862
        self.assertIsInstance(obj1, Unique)
 
1863
        obj2 = self.bus.get("busname", "objectpath")
 
1864
        self.assertIsInstance(obj2, Unique)
 
1865
        self.assertIs(obj1, obj2)
 
1866
 
 
1867
    def test_returns_same_old(self):
 
1868
        obj1 = self.bus.get("busname1", "objectpath1")
 
1869
        self.assertIsInstance(obj1, Unique)
 
1870
        obj2 = self.bus.get("busname2", "objectpath2")
 
1871
        self.assertIsInstance(obj2, Unique)
 
1872
        obj1b = self.bus.get("busname1", "objectpath1")
 
1873
        self.assertIsInstance(obj1b, Unique)
 
1874
        self.assertIsNot(obj1, obj2)
 
1875
        self.assertIsNot(obj2, obj1b)
 
1876
        self.assertIs(obj1, obj1b)
 
1877
 
 
1878
 
 
1879
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
 
1880
 
 
1881
    class dummy_dbussy:
 
1882
        class DBUS:
 
1883
            class ObjectPath(str):
 
1884
                pass
 
1885
        class DBusError(Exception):
 
1886
            pass
 
1887
 
 
1888
    def fake_ravel_func(self, func):
 
1889
        class fake_ravel:
 
1890
            @staticmethod
 
1891
            def system_bus():
 
1892
                class DBusInterfaceProxy:
 
1893
                    @staticmethod
 
1894
                    def methodname(*args):
 
1895
                        return [func(*args)]
 
1896
                class DBusObject:
 
1897
                    @staticmethod
 
1898
                    def get_interface(interface):
 
1899
                        if interface == "interface":
 
1900
                            return DBusInterfaceProxy()
 
1901
                return {"busname": {"objectpath": DBusObject()}}
 
1902
        return fake_ravel
 
1903
 
 
1904
    def call_method(self, bus, methodname, busname, objectpath,
 
1905
                    interface, *args):
 
1906
        with self.assertLogs(log, logging.DEBUG):
 
1907
            return bus.call_method(methodname, busname, objectpath,
 
1908
                                   interface, *args)
 
1909
 
 
1910
    def test_call_method_returns(self):
 
1911
        expected_method_return = Unique()
 
1912
        method_args = (Unique(), Unique())
 
1913
        def func(*args):
 
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",
 
1922
                               *method_args)
 
1923
        self.assertIs(ret, expected_method_return)
 
1924
 
 
1925
    def test_call_method_filters_objectpath(self):
 
1926
        def func():
 
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)
 
1937
 
 
1938
    def test_call_method_filters_objectpaths_in_dict(self):
 
1939
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1940
        def func():
 
1941
            return method_return
 
1942
        fake_ravel = self.fake_ravel_func(func)
 
1943
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1944
        method_return = {
 
1945
            ObjectPath("objectpath_key_1"):
 
1946
            ObjectPath("objectpath_value_1"),
 
1947
            ObjectPath("objectpath_key_2"):
 
1948
            ObjectPath("objectpath_value_2"),
 
1949
        }
 
1950
        ret = self.call_method(bus, "methodname", "busname",
 
1951
                               "objectpath", "interface")
 
1952
        expected_method_return = {str(key): str(value)
 
1953
                                  for key, value in
 
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)
 
1960
 
 
1961
    def test_call_method_filters_objectpaths_in_dict_in_dict(self):
 
1962
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1963
        def func():
 
1964
            return method_return
 
1965
        fake_ravel = self.fake_ravel_func(func)
 
1966
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1967
        method_return = {
 
1968
            ObjectPath("key1"): {
 
1969
                ObjectPath("key11"): ObjectPath("value11"),
 
1970
                ObjectPath("key12"): ObjectPath("value12"),
 
1971
            },
 
1972
            ObjectPath("key2"): {
 
1973
                ObjectPath("key21"): ObjectPath("value21"),
 
1974
                ObjectPath("key22"): ObjectPath("value22"),
 
1975
            },
 
1976
        }
 
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"},
 
1984
        }
 
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)
 
1992
                self.assertEqual(
 
1993
                    expected_method_return[key][inner_key],
 
1994
                    inner_value)
 
1995
                self.assertNotIsInstance(key, ObjectPath)
 
1996
 
 
1997
    def test_call_method_filters_objectpaths_in_dict_three_deep(self):
 
1998
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1999
        def func():
 
2000
            return method_return
 
2001
        fake_ravel = self.fake_ravel_func(func)
 
2002
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
2003
        method_return = {
 
2004
            ObjectPath("key1"): {
 
2005
                ObjectPath("key2"): {
 
2006
                    ObjectPath("key3"): ObjectPath("value"),
 
2007
                },
 
2008
            },
 
2009
        }
 
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())),
 
2018
                                 ObjectPath)
 
2019
        self.assertIsInstance(ret["key1"]["key2"], dict)
 
2020
        self.assertNotIsInstance(
 
2021
            next(iter(ret["key1"]["key2"].keys())),
 
2022
            ObjectPath)
 
2023
        self.assertEqual("value", ret["key1"]["key2"]["key3"])
 
2024
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
 
2025
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
2026
 
 
2027
    def test_call_method_handles_exception(self):
 
2028
        def func():
 
2029
            raise self.dummy_dbussy.DBusError()
 
2030
 
 
2031
        fake_ravel = self.fake_ravel_func(func)
 
2032
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
2033
 
 
2034
        with self.assertRaises(dbus.Error) as e:
 
2035
            self.call_method(bus, "methodname", "busname",
 
2036
                             "objectpath", "interface")
 
2037
 
 
2038
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
2039
 
 
2040
    def test_get_object_converts_to_correct_exception(self):
 
2041
        class fake_ravel_raises_exception_on_connect:
 
2042
            @staticmethod
 
2043
            def system_bus():
 
2044
                class Bus:
 
2045
                    @staticmethod
 
2046
                    def __getitem__(key):
 
2047
                        if key == "objectpath":
 
2048
                            raise self.dummy_dbussy.DBusError()
 
2049
                        raise Exception(key)
 
2050
                return {"busname": Bus()}
 
2051
        def func():
 
2052
            raise self.dummy_dbussy.DBusError()
 
2053
        bus = dbussy_adapter.SystemBus(
 
2054
            self.dummy_dbussy,
 
2055
            fake_ravel_raises_exception_on_connect)
 
2056
        with self.assertRaises(dbus.ConnectFailed):
 
2057
            self.call_method(bus, "methodname", "busname",
 
2058
                             "objectpath", "interface")
 
2059
 
 
2060
 
1511
2061
class Test_commands_from_options(unittest.TestCase):
1512
2062
 
1513
2063
    def setUp(self):
1518
2068
        self.assert_command_from_args(["--is-enabled", "client"],
1519
2069
                                      command.IsEnabled)
1520
2070
 
1521
 
    def assert_command_from_args(self, args, command_cls,
1522
 
                                 **cmd_attrs):
 
2071
    def assert_command_from_args(self, args, command_cls, length=1,
 
2072
                                 clients=None, **cmd_attrs):
1523
2073
        """Assert that parsing ARGS should result in an instance of
1524
2074
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1525
2075
        options = self.parser.parse_args(args)
1526
2076
        check_option_syntax(self.parser, options)
1527
2077
        commands = commands_from_options(options)
1528
 
        self.assertEqual(1, len(commands))
1529
 
        command = commands[0]
1530
 
        self.assertIsInstance(command, command_cls)
 
2078
        self.assertEqual(length, len(commands))
 
2079
        for command in commands:
 
2080
            if isinstance(command, command_cls):
 
2081
                break
 
2082
        else:
 
2083
            self.assertIsInstance(command, command_cls)
 
2084
        if clients is not None:
 
2085
            self.assertEqual(clients, options.client)
1531
2086
        for key, value in cmd_attrs.items():
1532
2087
            self.assertEqual(value, getattr(command, key))
1533
2088
 
 
2089
    def assert_commands_from_args(self, args, commands, clients=None):
 
2090
        for cmd in commands:
 
2091
            self.assert_command_from_args(args, cmd,
 
2092
                                          length=len(commands),
 
2093
                                          clients=clients)
 
2094
 
1534
2095
    def test_is_enabled_short(self):
1535
2096
        self.assert_command_from_args(["-V", "client"],
1536
2097
                                      command.IsEnabled)
1727
2288
                                      verbose=True)
1728
2289
 
1729
2290
 
 
2291
    def test_manual_page_example_1(self):
 
2292
        self.assert_command_from_args("",
 
2293
                                      command.PrintTable,
 
2294
                                      clients=[],
 
2295
                                      verbose=False)
 
2296
 
 
2297
    def test_manual_page_example_2(self):
 
2298
        self.assert_command_from_args(
 
2299
            "--verbose foo1.example.org foo2.example.org".split(),
 
2300
            command.PrintTable, clients=["foo1.example.org",
 
2301
                                         "foo2.example.org"],
 
2302
            verbose=True)
 
2303
 
 
2304
    def test_manual_page_example_3(self):
 
2305
        self.assert_command_from_args("--enable --all".split(),
 
2306
                                      command.Enable,
 
2307
                                      clients=[])
 
2308
 
 
2309
    def test_manual_page_example_4(self):
 
2310
        self.assert_commands_from_args(
 
2311
            ("--timeout=PT5M --interval=PT1M foo1.example.org"
 
2312
             " foo2.example.org").split(),
 
2313
            [command.SetTimeout, command.SetInterval],
 
2314
            clients=["foo1.example.org", "foo2.example.org"])
 
2315
 
 
2316
    def test_manual_page_example_5(self):
 
2317
        self.assert_command_from_args("--approve --all".split(),
 
2318
                                      command.Approve,
 
2319
                                      clients=[])
 
2320
 
 
2321
 
1730
2322
class TestCommand(unittest.TestCase):
1731
2323
    """Abstract class for tests of command classes"""
1732
2324