/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: 2019-03-31 04:39:25 UTC
  • Revision ID: teddy@recompile.se-20190331043925-0j9pdspo3hux5nka
mandos-ctl: Refactor: Move command list generation into argparse

* mandos-ctl (add_command_line_options): Don't simply set flags for
                                         each option; instead, add
                                         command objects to a
                                         "commands" list.
  (check_option_syntax): Rework to deal with an already-generated
                         command list.
  (commands_from_options): Remove all code to generate command list,
                           but add more code to move command.Deny
                           ahead of command.Remove in command list.
  (command.PropertySetterValue.argparse): New; used in
                                          add_command_line_options.
  (Test_check_option_syntax): Refactor to pass actual string arguments
                              to parse_args() instead of modifying an
                              argparse.Namespace.

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 -*-
 
1
#!/usr/bin/python
 
2
# -*- mode: python; coding: utf-8; after-save-hook: (lambda () (let ((command (if (and (boundp 'tramp-file-name-structure) (string-match (car tramp-file-name-structure) (buffer-file-name))) (tramp-file-name-localname (tramp-dissect-file-name (buffer-file-name))) (buffer-file-name)))) (if (= (shell-command (format "%s --check" (shell-quote-argument command)) "*Test*") 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w)) (kill-buffer "*Test*")) (display-buffer "*Test*")))); -*-
3
3
#
4
4
# Mandos Monitor - Control and monitor the Mandos server
5
5
#
45
45
import io
46
46
import tempfile
47
47
import contextlib
48
 
 
49
 
if sys.version_info.major == 2:
50
 
    __metaclass__ = type
51
 
    str = unicode
52
 
 
53
 
try:
54
 
    import pydbus
55
 
    import gi
56
 
    dbus_python = None
57
 
except ImportError:
58
 
    import dbus as dbus_python
59
 
    pydbus = None
60
 
    class gi:
61
 
        """Dummy gi module, for the tests"""
62
 
        class repository:
63
 
            class GLib:
64
 
                class Error(Exception):
65
 
                    pass
 
48
import abc
 
49
 
 
50
import dbus as dbus_python
66
51
 
67
52
# Show warnings by default
68
53
if not sys.warnoptions:
76
61
logging.captureWarnings(True)   # Show warnings via the logging system
77
62
 
78
63
if sys.version_info.major == 2:
 
64
    str = unicode
79
65
    import StringIO
80
66
    io.StringIO = StringIO.StringIO
81
67
 
82
68
locale.setlocale(locale.LC_ALL, "")
83
69
 
84
 
version = "1.8.9"
 
70
version = "1.8.3"
85
71
 
86
72
 
87
73
def main():
96
82
    if options.debug:
97
83
        log.setLevel(logging.DEBUG)
98
84
 
99
 
    if pydbus is not None:
100
 
        bus = pydbus_adapter.CachingBus(pydbus)
101
 
    else:
102
 
        bus = dbus_python_adapter.CachingBus(dbus_python)
 
85
    bus = dbus_python_adapter.CachingBus(dbus_python)
103
86
 
104
87
    try:
105
88
        all_clients = bus.get_clients_and_properties()
253
236
def rfc3339_duration_to_delta(duration):
254
237
    """Parse an RFC 3339 "duration" and return a datetime.timedelta
255
238
 
256
 
    >>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)
257
 
    True
258
 
    >>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)
259
 
    True
260
 
    >>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)
261
 
    True
262
 
    >>> # 60 months
263
 
    >>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)
264
 
    True
265
 
    >>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)
266
 
    True
267
 
    >>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)
268
 
    True
269
 
    >>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)
270
 
    True
271
 
    >>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)
272
 
    True
 
239
    >>> rfc3339_duration_to_delta("P7D")
 
240
    datetime.timedelta(7)
 
241
    >>> rfc3339_duration_to_delta("PT60S")
 
242
    datetime.timedelta(0, 60)
 
243
    >>> rfc3339_duration_to_delta("PT60M")
 
244
    datetime.timedelta(0, 3600)
 
245
    >>> rfc3339_duration_to_delta("P60M")
 
246
    datetime.timedelta(1680)
 
247
    >>> rfc3339_duration_to_delta("PT24H")
 
248
    datetime.timedelta(1)
 
249
    >>> rfc3339_duration_to_delta("P1W")
 
250
    datetime.timedelta(7)
 
251
    >>> rfc3339_duration_to_delta("PT5M30S")
 
252
    datetime.timedelta(0, 330)
 
253
    >>> rfc3339_duration_to_delta("P1DT3M20S")
 
254
    datetime.timedelta(1, 200)
273
255
    >>> # Can not be empty:
274
256
    >>> rfc3339_duration_to_delta("")
275
257
    Traceback (most recent call last):
385
367
    """Parse an interval string as documented by Mandos before 1.6.1,
386
368
    and return a datetime.timedelta
387
369
 
388
 
    >>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)
389
 
    True
390
 
    >>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)
391
 
    True
392
 
    >>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)
393
 
    True
394
 
    >>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)
395
 
    True
396
 
    >>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)
397
 
    True
398
 
    >>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)
399
 
    True
400
 
    >>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)
401
 
    True
 
370
    >>> parse_pre_1_6_1_interval('7d')
 
371
    datetime.timedelta(7)
 
372
    >>> parse_pre_1_6_1_interval('60s')
 
373
    datetime.timedelta(0, 60)
 
374
    >>> parse_pre_1_6_1_interval('60m')
 
375
    datetime.timedelta(0, 3600)
 
376
    >>> parse_pre_1_6_1_interval('24h')
 
377
    datetime.timedelta(1)
 
378
    >>> parse_pre_1_6_1_interval('1w')
 
379
    datetime.timedelta(7)
 
380
    >>> parse_pre_1_6_1_interval('5m 30s')
 
381
    datetime.timedelta(0, 330)
 
382
    >>> parse_pre_1_6_1_interval('')
 
383
    datetime.timedelta(0)
402
384
    >>> # Ignore unknown characters, allow any order and repetitions
403
 
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
404
 
    True
 
385
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
 
386
    datetime.timedelta(2, 480, 18000)
405
387
 
406
388
    """
407
389
 
470
452
        parser.error("--remove can only be combined with --deny")
471
453
 
472
454
 
473
 
class dbus:
 
455
class dbus(object):
474
456
 
475
 
    class SystemBus:
 
457
    class SystemBus(object):
476
458
 
477
459
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
478
460
        def get_managed_objects(self, busname, objectpath):
524
506
        pass
525
507
 
526
508
 
527
 
class dbus_python_adapter:
 
509
class dbus_python_adapter(object):
528
510
 
529
511
    class SystemBus(dbus.MandosBus):
530
512
        """Use dbus-python"""
575
557
                        for key, subval in value.items()}
576
558
            return value
577
559
 
578
 
        def set_client_property(self, objectpath, key, value):
579
 
            if key == "Secret":
580
 
                if not isinstance(value, bytes):
581
 
                    value = value.encode("utf-8")
582
 
                value = self.dbus_python.ByteArray(value)
583
 
            return self.set_property(self.busname, objectpath,
584
 
                                     self.client_interface, key,
585
 
                                     value)
586
560
 
587
 
    class SilenceLogger:
 
561
    class SilenceLogger(object):
588
562
        "Simple context manager to silence a particular logger"
589
563
        def __init__(self, loggername):
590
564
            self.logger = logging.getLogger(loggername)
619
593
                return new_object
620
594
 
621
595
 
622
 
class pydbus_adapter:
623
 
    class SystemBus(dbus.MandosBus):
624
 
        def __init__(self, module=pydbus):
625
 
            self.pydbus = module
626
 
            self.bus = self.pydbus.SystemBus()
627
 
 
628
 
        @contextlib.contextmanager
629
 
        def convert_exception(self, exception_class=dbus.Error):
630
 
            try:
631
 
                yield
632
 
            except gi.repository.GLib.Error as e:
633
 
                # This does what "raise from" would do
634
 
                exc = exception_class(*e.args)
635
 
                exc.__cause__ = e
636
 
                raise exc
637
 
 
638
 
        def call_method(self, methodname, busname, objectpath,
639
 
                        interface, *args):
640
 
            proxy_object = self.get(busname, objectpath)
641
 
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
642
 
                      interface, methodname,
643
 
                      ", ".join(repr(a) for a in args))
644
 
            method = getattr(proxy_object[interface], methodname)
645
 
            with self.convert_exception():
646
 
                return method(*args)
647
 
 
648
 
        def get(self, busname, objectpath):
649
 
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
650
 
                      busname, objectpath)
651
 
            with self.convert_exception(dbus.ConnectFailed):
652
 
                if sys.version_info.major <= 2:
653
 
                    with warnings.catch_warnings():
654
 
                        warnings.filterwarnings(
655
 
                            "ignore", "", DeprecationWarning,
656
 
                            r"^xml\.etree\.ElementTree$")
657
 
                        return self.bus.get(busname, objectpath)
658
 
                else:
659
 
                    return self.bus.get(busname, objectpath)
660
 
 
661
 
        def set_property(self, busname, objectpath, interface, key,
662
 
                         value):
663
 
            proxy_object = self.get(busname, objectpath)
664
 
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
665
 
                      objectpath, self.properties_iface, interface,
666
 
                      key, value)
667
 
            setattr(proxy_object[interface], key, value)
668
 
 
669
 
    class CachingBus(SystemBus):
670
 
        """A caching layer for pydbus_adapter.SystemBus"""
671
 
        def __init__(self, *args, **kwargs):
672
 
            self.object_cache = {}
673
 
            super(pydbus_adapter.CachingBus,
674
 
                  self).__init__(*args, **kwargs)
675
 
        def get(self, busname, objectpath):
676
 
            try:
677
 
                return self.object_cache[(busname, objectpath)]
678
 
            except KeyError:
679
 
                new_object = (super(pydbus_adapter.CachingBus, self)
680
 
                              .get(busname, objectpath))
681
 
                self.object_cache[(busname, objectpath)]  = new_object
682
 
                return new_object
683
 
 
684
 
 
685
596
def commands_from_options(options):
686
597
 
687
598
    commands = list(options.commands)
715
626
    return commands
716
627
 
717
628
 
718
 
class command:
 
629
class command(object):
719
630
    """A namespace for command classes"""
720
631
 
721
 
    class Base:
 
632
    class Base(object):
722
633
        """Abstract base class for commands"""
723
634
        def run(self, clients, bus=None):
724
635
            """Normal commands should implement run_on_one_client(),
787
698
                keywords = self.all_keywords
788
699
            print(self.TableOfClients(clients.values(), keywords))
789
700
 
790
 
        class TableOfClients:
 
701
        class TableOfClients(object):
791
702
            tableheaders = {
792
703
                "Name": "Name",
793
704
                "Enabled": "Enabled",
1024
935
                                                     "output"))
1025
936
 
1026
937
 
1027
 
class Unique:
 
938
class Unique(object):
1028
939
    """Class for objects which exist only to be unique objects, since
1029
940
unittest.mock.sentinel only exists in Python 3.3"""
1030
941
 
1314
1225
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1315
1226
 
1316
1227
    def MockDBusPython_func(self, func):
1317
 
        class mock_dbus_python:
 
1228
        class mock_dbus_python(object):
1318
1229
            """mock dbus-python module"""
1319
 
            class exceptions:
 
1230
            class exceptions(object):
1320
1231
                """Pseudo-namespace"""
1321
1232
                class DBusException(Exception):
1322
1233
                    pass
1323
 
            class SystemBus:
 
1234
            class SystemBus(object):
1324
1235
                @staticmethod
1325
1236
                def get_object(busname, objectpath):
1326
1237
                    DBusObject = collections.namedtuple(
1327
 
                        "DBusObject", ("methodname", "Set"))
 
1238
                        "DBusObject", ("methodname",))
1328
1239
                    def method(*args, **kwargs):
1329
1240
                        self.assertEqual({"dbus_interface":
1330
1241
                                          "interface"},
1331
1242
                                         kwargs)
1332
1243
                        return func(*args)
1333
 
                    def set_property(interface, key, value,
1334
 
                                     dbus_interface=None):
1335
 
                        self.assertEqual(
1336
 
                            "org.freedesktop.DBus.Properties",
1337
 
                            dbus_interface)
1338
 
                        self.assertEqual("Secret", key)
1339
 
                        return func(interface, key, value,
1340
 
                                    dbus_interface=dbus_interface)
1341
 
                    return DBusObject(methodname=method,
1342
 
                                      Set=set_property)
1343
 
            class Boolean:
 
1244
                    return DBusObject(methodname=method)
 
1245
            class Boolean(object):
1344
1246
                def __init__(self, value):
1345
1247
                    self.value = bool(value)
1346
1248
                def __bool__(self):
1351
1253
                pass
1352
1254
            class Dictionary(dict):
1353
1255
                pass
1354
 
            class ByteArray(bytes):
1355
 
                pass
1356
1256
        return mock_dbus_python
1357
1257
 
1358
1258
    def call_method(self, bus, methodname, busname, objectpath,
1535
1435
        # Make sure the dbus logger was suppressed
1536
1436
        self.assertEqual(0, counting_handler.count)
1537
1437
 
1538
 
    def test_Set_Secret_sends_bytearray(self):
1539
 
        ret = [None]
1540
 
        def func(*args, **kwargs):
1541
 
            ret[0] = (args, kwargs)
1542
 
        mock_dbus_python = self.MockDBusPython_func(func)
1543
 
        bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1544
 
        bus.set_client_property("objectpath", "Secret", "value")
1545
 
        expected_call = (("se.recompile.Mandos.Client", "Secret",
1546
 
                          mock_dbus_python.ByteArray(b"value")),
1547
 
                         {"dbus_interface":
1548
 
                          "org.freedesktop.DBus.Properties"})
1549
 
        self.assertEqual(expected_call, ret[0])
1550
 
        if sys.version_info.major == 2:
1551
 
            self.assertIsInstance(ret[0][0][-1],
1552
 
                                  mock_dbus_python.ByteArray)
1553
 
 
1554
1438
    def test_get_object_converts_to_correct_exception(self):
1555
1439
        bus = dbus_python_adapter.SystemBus(
1556
1440
            self.fake_dbus_python_raises_exception_on_connect)
1558
1442
            self.call_method(bus, "methodname", "busname",
1559
1443
                             "objectpath", "interface")
1560
1444
 
1561
 
    class fake_dbus_python_raises_exception_on_connect:
 
1445
    class fake_dbus_python_raises_exception_on_connect(object):
1562
1446
        """fake dbus-python module"""
1563
 
        class exceptions:
 
1447
        class exceptions(object):
1564
1448
            """Pseudo-namespace"""
1565
1449
            class DBusException(Exception):
1566
1450
                pass
1574
1458
 
1575
1459
 
1576
1460
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1577
 
    class mock_dbus_python:
 
1461
    class mock_dbus_python(object):
1578
1462
        """mock dbus-python modules"""
1579
 
        class SystemBus:
 
1463
        class SystemBus(object):
1580
1464
            @staticmethod
1581
1465
            def get_object(busname, objectpath):
1582
1466
                return Unique()
1625
1509
        self.assertIs(obj1, obj1b)
1626
1510
 
1627
1511
 
1628
 
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1629
 
 
1630
 
    def Stub_pydbus_func(self, func):
1631
 
        class stub_pydbus:
1632
 
            """stub pydbus module"""
1633
 
            class SystemBus:
1634
 
                @staticmethod
1635
 
                def get(busname, objectpath):
1636
 
                    DBusObject = collections.namedtuple(
1637
 
                        "DBusObject", ("methodname",))
1638
 
                    return {"interface":
1639
 
                            DBusObject(methodname=func)}
1640
 
        return stub_pydbus
1641
 
 
1642
 
    def call_method(self, bus, methodname, busname, objectpath,
1643
 
                    interface, *args):
1644
 
        with self.assertLogs(log, logging.DEBUG):
1645
 
            return bus.call_method(methodname, busname, objectpath,
1646
 
                                   interface, *args)
1647
 
 
1648
 
    def test_call_method_returns(self):
1649
 
        expected_method_return = Unique()
1650
 
        method_args = (Unique(), Unique())
1651
 
        def func(*args):
1652
 
            self.assertEqual(len(method_args), len(args))
1653
 
            for marg, arg in zip(method_args, args):
1654
 
                self.assertIs(marg, arg)
1655
 
            return expected_method_return
1656
 
        stub_pydbus = self.Stub_pydbus_func(func)
1657
 
        bus = pydbus_adapter.SystemBus(stub_pydbus)
1658
 
        ret = self.call_method(bus, "methodname", "busname",
1659
 
                               "objectpath", "interface",
1660
 
                               *method_args)
1661
 
        self.assertIs(ret, expected_method_return)
1662
 
 
1663
 
    def test_call_method_handles_exception(self):
1664
 
        dbus_logger = logging.getLogger("dbus.proxies")
1665
 
 
1666
 
        def func():
1667
 
            raise gi.repository.GLib.Error()
1668
 
 
1669
 
        stub_pydbus = self.Stub_pydbus_func(func)
1670
 
        bus = pydbus_adapter.SystemBus(stub_pydbus)
1671
 
 
1672
 
        with self.assertRaises(dbus.Error) as e:
1673
 
            self.call_method(bus, "methodname", "busname",
1674
 
                             "objectpath", "interface")
1675
 
 
1676
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
1677
 
 
1678
 
    def test_get_converts_to_correct_exception(self):
1679
 
        bus = pydbus_adapter.SystemBus(
1680
 
            self.fake_pydbus_raises_exception_on_connect)
1681
 
        with self.assertRaises(dbus.ConnectFailed):
1682
 
            self.call_method(bus, "methodname", "busname",
1683
 
                             "objectpath", "interface")
1684
 
 
1685
 
    class fake_pydbus_raises_exception_on_connect:
1686
 
        """fake dbus-python module"""
1687
 
        @classmethod
1688
 
        def SystemBus(cls):
1689
 
            def get(busname, objectpath):
1690
 
                raise gi.repository.GLib.Error()
1691
 
            Bus = collections.namedtuple("Bus", ["get"])
1692
 
            return Bus(get=get)
1693
 
 
1694
 
    def test_set_property_uses_setattr(self):
1695
 
        class Object:
1696
 
            pass
1697
 
        obj = Object()
1698
 
        class pydbus_spy:
1699
 
            class SystemBus:
1700
 
                @staticmethod
1701
 
                def get(busname, objectpath):
1702
 
                    return {"interface": obj}
1703
 
        bus = pydbus_adapter.SystemBus(pydbus_spy)
1704
 
        value = Unique()
1705
 
        bus.set_property("busname", "objectpath", "interface", "key",
1706
 
                         value)
1707
 
        self.assertIs(value, obj.key)
1708
 
 
1709
 
    def test_get_suppresses_xml_deprecation_warning(self):
1710
 
        if sys.version_info.major >= 3:
1711
 
            return
1712
 
        class stub_pydbus_get:
1713
 
            class SystemBus:
1714
 
                @staticmethod
1715
 
                def get(busname, objectpath):
1716
 
                    warnings.warn_explicit(
1717
 
                        "deprecated", DeprecationWarning,
1718
 
                        "xml.etree.ElementTree", 0)
1719
 
        bus = pydbus_adapter.SystemBus(stub_pydbus_get)
1720
 
        with warnings.catch_warnings(record=True) as w:
1721
 
            warnings.simplefilter("always")
1722
 
            bus.get("busname", "objectpath")
1723
 
            self.assertEqual(0, len(w))
1724
 
 
1725
 
 
1726
 
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1727
 
    class stub_pydbus:
1728
 
        """stub pydbus module"""
1729
 
        class SystemBus:
1730
 
            @staticmethod
1731
 
            def get(busname, objectpath):
1732
 
                return Unique()
1733
 
 
1734
 
    def setUp(self):
1735
 
        self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
1736
 
 
1737
 
    def test_returns_distinct_objectpaths(self):
1738
 
        obj1 = self.bus.get("busname", "objectpath1")
1739
 
        self.assertIsInstance(obj1, Unique)
1740
 
        obj2 = self.bus.get("busname", "objectpath2")
1741
 
        self.assertIsInstance(obj2, Unique)
1742
 
        self.assertIsNot(obj1, obj2)
1743
 
 
1744
 
    def test_returns_distinct_busnames(self):
1745
 
        obj1 = self.bus.get("busname1", "objectpath")
1746
 
        self.assertIsInstance(obj1, Unique)
1747
 
        obj2 = self.bus.get("busname2", "objectpath")
1748
 
        self.assertIsInstance(obj2, Unique)
1749
 
        self.assertIsNot(obj1, obj2)
1750
 
 
1751
 
    def test_returns_distinct_both(self):
1752
 
        obj1 = self.bus.get("busname1", "objectpath")
1753
 
        self.assertIsInstance(obj1, Unique)
1754
 
        obj2 = self.bus.get("busname2", "objectpath")
1755
 
        self.assertIsInstance(obj2, Unique)
1756
 
        self.assertIsNot(obj1, obj2)
1757
 
 
1758
 
    def test_returns_same(self):
1759
 
        obj1 = self.bus.get("busname", "objectpath")
1760
 
        self.assertIsInstance(obj1, Unique)
1761
 
        obj2 = self.bus.get("busname", "objectpath")
1762
 
        self.assertIsInstance(obj2, Unique)
1763
 
        self.assertIs(obj1, obj2)
1764
 
 
1765
 
    def test_returns_same_old(self):
1766
 
        obj1 = self.bus.get("busname1", "objectpath1")
1767
 
        self.assertIsInstance(obj1, Unique)
1768
 
        obj2 = self.bus.get("busname2", "objectpath2")
1769
 
        self.assertIsInstance(obj2, Unique)
1770
 
        obj1b = self.bus.get("busname1", "objectpath1")
1771
 
        self.assertIsInstance(obj1b, Unique)
1772
 
        self.assertIsNot(obj1, obj2)
1773
 
        self.assertIsNot(obj2, obj1b)
1774
 
        self.assertIs(obj1, obj1b)
1775
 
 
1776
 
 
1777
1512
class Test_commands_from_options(unittest.TestCase):
1778
1513
 
1779
1514
    def setUp(self):
1784
1519
        self.assert_command_from_args(["--is-enabled", "client"],
1785
1520
                                      command.IsEnabled)
1786
1521
 
1787
 
    def assert_command_from_args(self, args, command_cls, length=1,
1788
 
                                 clients=None, **cmd_attrs):
 
1522
    def assert_command_from_args(self, args, command_cls,
 
1523
                                 **cmd_attrs):
1789
1524
        """Assert that parsing ARGS should result in an instance of
1790
1525
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1791
1526
        options = self.parser.parse_args(args)
1792
1527
        check_option_syntax(self.parser, options)
1793
1528
        commands = commands_from_options(options)
1794
 
        self.assertEqual(length, len(commands))
1795
 
        for command in commands:
1796
 
            if isinstance(command, command_cls):
1797
 
                break
1798
 
        else:
1799
 
            self.assertIsInstance(command, command_cls)
1800
 
        if clients is not None:
1801
 
            self.assertEqual(clients, options.client)
 
1529
        self.assertEqual(1, len(commands))
 
1530
        command = commands[0]
 
1531
        self.assertIsInstance(command, command_cls)
1802
1532
        for key, value in cmd_attrs.items():
1803
1533
            self.assertEqual(value, getattr(command, key))
1804
1534
 
1805
 
    def assert_commands_from_args(self, args, commands, clients=None):
1806
 
        for cmd in commands:
1807
 
            self.assert_command_from_args(args, cmd,
1808
 
                                          length=len(commands),
1809
 
                                          clients=clients)
1810
 
 
1811
1535
    def test_is_enabled_short(self):
1812
1536
        self.assert_command_from_args(["-V", "client"],
1813
1537
                                      command.IsEnabled)
2004
1728
                                      verbose=True)
2005
1729
 
2006
1730
 
2007
 
    def test_manual_page_example_1(self):
2008
 
        self.assert_command_from_args("",
2009
 
                                      command.PrintTable,
2010
 
                                      clients=[],
2011
 
                                      verbose=False)
2012
 
 
2013
 
    def test_manual_page_example_2(self):
2014
 
        self.assert_command_from_args(
2015
 
            "--verbose foo1.example.org foo2.example.org".split(),
2016
 
            command.PrintTable, clients=["foo1.example.org",
2017
 
                                         "foo2.example.org"],
2018
 
            verbose=True)
2019
 
 
2020
 
    def test_manual_page_example_3(self):
2021
 
        self.assert_command_from_args("--enable --all".split(),
2022
 
                                      command.Enable,
2023
 
                                      clients=[])
2024
 
 
2025
 
    def test_manual_page_example_4(self):
2026
 
        self.assert_commands_from_args(
2027
 
            ("--timeout=PT5M --interval=PT1M foo1.example.org"
2028
 
             " foo2.example.org").split(),
2029
 
            [command.SetTimeout, command.SetInterval],
2030
 
            clients=["foo1.example.org", "foo2.example.org"])
2031
 
 
2032
 
    def test_manual_page_example_5(self):
2033
 
        self.assert_command_from_args("--approve --all".split(),
2034
 
                                      command.Approve,
2035
 
                                      clients=[])
2036
 
 
2037
 
 
2038
1731
class TestCommand(unittest.TestCase):
2039
1732
    """Abstract class for tests of command classes"""
2040
1733