/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: 2024-11-22 20:30:34 UTC
  • Revision ID: teddy@recompile.se-20241122203034-92q7483wxv29tqev
mandos-monitor: avoid deprecation warnings from urwid

The urwid.FlowWidget class is deprecated.  Replace it with inheriting
from urwid.Widget, and setting the appropriate class attribute.

* mandos-monitor: Replace any mentions of urwid.FlowWidget with
  urwid.Widget.
  (MandosClientWidget): Inherit from "urwid.Widget" instead of
  "urwid.FlowWidget".  Also, move MandosClientPropertyCache to first
  in inheritance list so that its __init__ method can absorb the
  "proxy_object" and "properties" keyword arguments.
  (MandosClientWidget._sizing): New class attribute; set to
  "frozenset(["flow"])".

(Thanks to an anonymous contributor for reporting this.)

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 --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 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
# -*- coding: utf-8; lexical-binding: t -*-
 
3
#
 
4
# Mandos Control - Control or query the Mandos server
 
5
#
 
6
# Copyright © 2008-2022 Teddy Hogeborn
 
7
# Copyright © 2008-2022 Björn Påhlsson
8
8
#
9
9
# This file is part of Mandos.
10
10
#
23
23
#
24
24
# Contact the authors at <mandos@recompile.se>.
25
25
#
26
 
 
27
26
from __future__ import (division, absolute_import, print_function,
28
27
                        unicode_literals)
29
28
 
33
32
    pass
34
33
 
35
34
import sys
 
35
import unittest
36
36
import argparse
 
37
import logging
 
38
import os
37
39
import locale
38
40
import datetime
39
41
import re
40
 
import os
41
42
import collections
42
43
import json
43
 
import unittest
44
 
import logging
45
44
import io
46
45
import tempfile
47
46
import contextlib
48
47
 
 
48
if sys.version_info.major == 2:
 
49
    __metaclass__ = type
 
50
    str = unicode
 
51
    input = raw_input
 
52
 
 
53
 
 
54
class gi:
 
55
    """Dummy gi module, for the tests"""
 
56
    class repository:
 
57
        class GLib:
 
58
            class Error(Exception):
 
59
                pass
 
60
 
 
61
 
 
62
dbussy = None
 
63
ravel = None
 
64
dbus_python = None
 
65
pydbus = None
 
66
 
49
67
try:
50
 
    import pydbus
51
 
    import gi
52
 
    dbus_python = None
 
68
    import dbussy
 
69
    import ravel
53
70
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
 
71
    try:
 
72
        import pydbus
 
73
        import gi
 
74
    except ImportError:
 
75
        import dbus as dbus_python
 
76
 
62
77
 
63
78
# Show warnings by default
64
79
if not sys.warnoptions:
65
80
    import warnings
66
81
    warnings.simplefilter("default")
67
82
 
68
 
log = logging.getLogger(sys.argv[0])
69
 
logging.basicConfig(level="INFO", # Show info level messages
 
83
log = logging.getLogger(os.path.basename(sys.argv[0]))
 
84
logging.basicConfig(level="INFO",         # Show info level messages
70
85
                    format="%(message)s") # Show basic log messages
71
86
 
72
87
logging.captureWarnings(True)   # Show warnings via the logging system
73
88
 
74
89
if sys.version_info.major == 2:
75
 
    str = unicode
76
90
    import StringIO
77
91
    io.StringIO = StringIO.StringIO
78
92
 
79
93
locale.setlocale(locale.LC_ALL, "")
80
94
 
81
 
version = "1.8.5"
 
95
version = "1.8.17"
82
96
 
83
97
 
84
98
def main():
91
105
    clientnames = options.client
92
106
 
93
107
    if options.debug:
94
 
        log.setLevel(logging.DEBUG)
 
108
        logging.getLogger("").setLevel(logging.DEBUG)
95
109
 
96
 
    if pydbus is not None:
 
110
    if dbussy is not None and ravel is not None:
 
111
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
 
112
    elif pydbus is not None:
97
113
        bus = pydbus_adapter.CachingBus(pydbus)
98
114
    else:
99
115
        bus = dbus_python_adapter.CachingBus(dbus_python)
243
259
        return rfc3339_duration_to_delta(interval)
244
260
    except ValueError as e:
245
261
        log.warning("%s - Parsing as pre-1.6.1 interval instead",
246
 
                    ' '.join(e.args))
 
262
                    " ".join(e.args))
247
263
    return parse_pre_1_6_1_interval(interval)
248
264
 
249
265
 
379
395
 
380
396
 
381
397
def parse_pre_1_6_1_interval(interval):
382
 
    """Parse an interval string as documented by Mandos before 1.6.1,
 
398
    r"""Parse an interval string as documented by Mandos before 1.6.1,
383
399
    and return a datetime.timedelta
384
400
 
385
 
    >>> parse_pre_1_6_1_interval('7d') == datetime.timedelta(days=7)
386
 
    True
387
 
    >>> parse_pre_1_6_1_interval('60s') == datetime.timedelta(0, 60)
388
 
    True
389
 
    >>> parse_pre_1_6_1_interval('60m') == datetime.timedelta(hours=1)
390
 
    True
391
 
    >>> parse_pre_1_6_1_interval('24h') == datetime.timedelta(days=1)
392
 
    True
393
 
    >>> parse_pre_1_6_1_interval('1w') == datetime.timedelta(days=7)
394
 
    True
395
 
    >>> parse_pre_1_6_1_interval('5m 30s') == datetime.timedelta(0, 330)
396
 
    True
397
 
    >>> parse_pre_1_6_1_interval('') == datetime.timedelta(0)
 
401
    >>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)
 
402
    True
 
403
    >>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)
 
404
    True
 
405
    >>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)
 
406
    True
 
407
    >>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)
 
408
    True
 
409
    >>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)
 
410
    True
 
411
    >>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)
 
412
    True
 
413
    >>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)
398
414
    True
399
415
    >>> # Ignore unknown characters, allow any order and repetitions
400
 
    >>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m') == datetime.timedelta(2, 480, 18000)
 
416
    >>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") \
 
417
    ... == datetime.timedelta(2, 480, 18000)
401
418
    True
402
419
 
403
420
    """
467
484
        parser.error("--remove can only be combined with --deny")
468
485
 
469
486
 
470
 
class dbus(object):
 
487
class dbus:
471
488
 
472
 
    class SystemBus(object):
 
489
    class SystemBus:
473
490
 
474
491
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
 
492
 
475
493
        def get_managed_objects(self, busname, objectpath):
476
494
            return self.call_method("GetManagedObjects", busname,
477
495
                                    objectpath,
478
496
                                    self.object_manager_iface)
479
497
 
480
498
        properties_iface = "org.freedesktop.DBus.Properties"
 
499
 
481
500
        def set_property(self, busname, objectpath, interface, key,
482
501
                         value):
483
502
            self.call_method("Set", busname, objectpath,
484
503
                             self.properties_iface, interface, key,
485
504
                             value)
486
505
 
 
506
        def call_method(self, methodname, busname, objectpath,
 
507
                        interface, *args):
 
508
            raise NotImplementedError()
487
509
 
488
510
    class MandosBus(SystemBus):
489
511
        busname_domain = "se.recompile"
521
543
        pass
522
544
 
523
545
 
524
 
class dbus_python_adapter(object):
 
546
class dbus_python_adapter:
525
547
 
526
548
    class SystemBus(dbus.MandosBus):
527
549
        """Use dbus-python"""
581
603
                                     self.client_interface, key,
582
604
                                     value)
583
605
 
584
 
    class SilenceLogger(object):
 
606
    class SilenceLogger:
585
607
        "Simple context manager to silence a particular logger"
 
608
 
586
609
        def __init__(self, loggername):
587
610
            self.logger = logging.getLogger(loggername)
588
611
 
598
621
        def __exit__(self, exc_type, exc_val, exc_tb):
599
622
            self.logger.removeFilter(self.nullfilter)
600
623
 
601
 
 
602
624
    class CachingBus(SystemBus):
603
625
        """A caching layer for dbus_python_adapter.SystemBus"""
 
626
 
604
627
        def __init__(self, *args, **kwargs):
605
628
            self.object_cache = {}
606
629
            super(dbus_python_adapter.CachingBus,
607
630
                  self).__init__(*args, **kwargs)
 
631
 
608
632
        def get_object(self, busname, objectpath):
609
633
            try:
610
634
                return self.object_cache[(busname, objectpath)]
612
636
                new_object = super(
613
637
                    dbus_python_adapter.CachingBus,
614
638
                    self).get_object(busname, objectpath)
615
 
                self.object_cache[(busname, objectpath)]  = new_object
 
639
                self.object_cache[(busname, objectpath)] = new_object
616
640
                return new_object
617
641
 
618
642
 
619
 
class pydbus_adapter(object):
 
643
class pydbus_adapter:
620
644
    class SystemBus(dbus.MandosBus):
621
645
        def __init__(self, module=pydbus):
622
646
            self.pydbus = module
665
689
 
666
690
    class CachingBus(SystemBus):
667
691
        """A caching layer for pydbus_adapter.SystemBus"""
 
692
 
668
693
        def __init__(self, *args, **kwargs):
669
694
            self.object_cache = {}
670
695
            super(pydbus_adapter.CachingBus,
671
696
                  self).__init__(*args, **kwargs)
 
697
 
672
698
        def get(self, busname, objectpath):
673
699
            try:
674
700
                return self.object_cache[(busname, objectpath)]
675
701
            except KeyError:
676
702
                new_object = (super(pydbus_adapter.CachingBus, self)
677
703
                              .get(busname, objectpath))
678
 
                self.object_cache[(busname, objectpath)]  = new_object
 
704
                self.object_cache[(busname, objectpath)] = new_object
 
705
                return new_object
 
706
 
 
707
 
 
708
class dbussy_adapter:
 
709
    class SystemBus(dbus.SystemBus):
 
710
        """Use DBussy"""
 
711
 
 
712
        def __init__(self, dbussy, ravel):
 
713
            self.dbussy = dbussy
 
714
            self.ravel = ravel
 
715
            self.bus = ravel.system_bus()
 
716
 
 
717
        @contextlib.contextmanager
 
718
        def convert_exception(self, exception_class=dbus.Error):
 
719
            try:
 
720
                yield
 
721
            except self.dbussy.DBusError as e:
 
722
                # This does what "raise from" would do
 
723
                exc = exception_class(*e.args)
 
724
                exc.__cause__ = e
 
725
                raise exc
 
726
 
 
727
        def call_method(self, methodname, busname, objectpath,
 
728
                        interface, *args):
 
729
            proxy_object = self.get_object(busname, objectpath)
 
730
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
731
                      interface, methodname,
 
732
                      ", ".join(repr(a) for a in args))
 
733
            iface = proxy_object.get_interface(interface)
 
734
            method = getattr(iface, methodname)
 
735
            with self.convert_exception(dbus.Error):
 
736
                value = method(*args)
 
737
            # DBussy returns values either as an empty list or as a
 
738
            # list of one element with the return value
 
739
            if value:
 
740
                return self.type_filter(value[0])
 
741
 
 
742
        def get_object(self, busname, objectpath):
 
743
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
744
                      busname, objectpath)
 
745
            with self.convert_exception(dbus.ConnectFailed):
 
746
                return self.bus[busname][objectpath]
 
747
 
 
748
        def type_filter(self, value):
 
749
            """Convert the most bothersome types to Python types"""
 
750
            # A D-Bus Variant value is represented as the Python type
 
751
            # Tuple[dbussy.DBUS.Signature, Any]
 
752
            if isinstance(value, tuple):
 
753
                if (len(value) == 2
 
754
                    and isinstance(value[0],
 
755
                                   self.dbussy.DBUS.Signature)):
 
756
                    return self.type_filter(value[1])
 
757
            elif isinstance(value, self.dbussy.DBUS.ObjectPath):
 
758
                return str(value)
 
759
            # Also recurse into dictionaries
 
760
            elif isinstance(value, dict):
 
761
                return {self.type_filter(key):
 
762
                        self.type_filter(subval)
 
763
                        for key, subval in value.items()}
 
764
            return value
 
765
 
 
766
        def set_property(self, busname, objectpath, interface, key,
 
767
                         value):
 
768
            proxy_object = self.get_object(busname, objectpath)
 
769
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
 
770
                      objectpath, self.properties_iface, interface,
 
771
                      key, value)
 
772
            if key == "Secret":
 
773
                # DBussy wants a Byte Array to be a sequence of
 
774
                # values, not a byte string
 
775
                value = tuple(value)
 
776
            setattr(proxy_object.get_interface(interface), key, value)
 
777
 
 
778
    class MandosBus(SystemBus, dbus.MandosBus):
 
779
        pass
 
780
 
 
781
    class CachingBus(MandosBus):
 
782
        """A caching layer for dbussy_adapter.MandosBus"""
 
783
 
 
784
        def __init__(self, *args, **kwargs):
 
785
            self.object_cache = {}
 
786
            super(dbussy_adapter.CachingBus, self).__init__(*args,
 
787
                                                            **kwargs)
 
788
 
 
789
        def get_object(self, busname, objectpath):
 
790
            try:
 
791
                return self.object_cache[(busname, objectpath)]
 
792
            except KeyError:
 
793
                new_object = super(
 
794
                    dbussy_adapter.CachingBus,
 
795
                    self).get_object(busname, objectpath)
 
796
                self.object_cache[(busname, objectpath)] = new_object
679
797
                return new_object
680
798
 
681
799
 
712
830
    return commands
713
831
 
714
832
 
715
 
class command(object):
 
833
class command:
716
834
    """A namespace for command classes"""
717
835
 
718
 
    class Base(object):
 
836
    class Base:
719
837
        """Abstract base class for commands"""
 
838
 
720
839
        def run(self, clients, bus=None):
721
840
            """Normal commands should implement run_on_one_client(),
722
841
but commands which want to operate on all clients at the same time can
726
845
            for client, properties in clients.items():
727
846
                self.run_on_one_client(client, properties)
728
847
 
729
 
 
730
848
    class IsEnabled(Base):
731
849
        def run(self, clients, bus=None):
732
850
            properties = next(iter(clients.values()))
734
852
                sys.exit(0)
735
853
            sys.exit(1)
736
854
 
737
 
 
738
855
    class Approve(Base):
739
856
        def run_on_one_client(self, client, properties):
740
857
            self.bus.call_client_method(client, "Approve", True)
741
858
 
742
 
 
743
859
    class Deny(Base):
744
860
        def run_on_one_client(self, client, properties):
745
861
            self.bus.call_client_method(client, "Approve", False)
746
862
 
747
 
 
748
863
    class Remove(Base):
749
864
        def run(self, clients, bus):
750
865
            for clientpath in frozenset(clients.keys()):
751
866
                bus.call_server_method("RemoveClient", clientpath)
752
867
 
753
 
 
754
868
    class Output(Base):
755
869
        """Abstract class for commands outputting client details"""
756
870
        all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
762
876
                        "Checker", "ExtendedTimeout", "Expires",
763
877
                        "LastCheckerStatus")
764
878
 
765
 
 
766
879
    class DumpJSON(Output):
767
880
        def run(self, clients, bus=None):
768
881
            data = {properties["Name"]:
769
882
                    {key: properties[key]
770
883
                     for key in self.all_keywords}
771
884
                    for properties in clients.values()}
772
 
            print(json.dumps(data, indent=4, separators=(',', ': ')))
773
 
 
 
885
            print(json.dumps(data, indent=4, separators=(",", ": ")))
774
886
 
775
887
    class PrintTable(Output):
776
888
        def __init__(self, verbose=False):
784
896
                keywords = self.all_keywords
785
897
            print(self.TableOfClients(clients.values(), keywords))
786
898
 
787
 
        class TableOfClients(object):
 
899
        class TableOfClients:
788
900
            tableheaders = {
789
901
                "Name": "Name",
790
902
                "Enabled": "Enabled",
817
929
 
818
930
            if sys.version_info.major == 2:
819
931
                __unicode__ = __str__
 
932
 
820
933
                def __str__(self):
821
934
                    return str(self).encode(
822
935
                        locale.getpreferredencoding())
868
981
                                minutes=(td.seconds % 3600) // 60,
869
982
                                seconds=td.seconds % 60))
870
983
 
871
 
 
872
984
    class PropertySetter(Base):
873
985
        "Abstract class for Actions for setting one client property"
874
986
 
881
993
        def propname(self):
882
994
            raise NotImplementedError()
883
995
 
884
 
 
885
996
    class Enable(PropertySetter):
886
997
        propname = "Enabled"
887
998
        value_to_set = True
888
999
 
889
 
 
890
1000
    class Disable(PropertySetter):
891
1001
        propname = "Enabled"
892
1002
        value_to_set = False
893
1003
 
894
 
 
895
1004
    class BumpTimeout(PropertySetter):
896
1005
        propname = "LastCheckedOK"
897
1006
        value_to_set = ""
898
1007
 
899
 
 
900
1008
    class StartChecker(PropertySetter):
901
1009
        propname = "CheckerRunning"
902
1010
        value_to_set = True
903
1011
 
904
 
 
905
1012
    class StopChecker(PropertySetter):
906
1013
        propname = "CheckerRunning"
907
1014
        value_to_set = False
908
1015
 
909
 
 
910
1016
    class ApproveByDefault(PropertySetter):
911
1017
        propname = "ApprovedByDefault"
912
1018
        value_to_set = True
913
1019
 
914
 
 
915
1020
    class DenyByDefault(PropertySetter):
916
1021
        propname = "ApprovedByDefault"
917
1022
        value_to_set = False
918
1023
 
919
 
 
920
1024
    class PropertySetterValue(PropertySetter):
921
1025
        """Abstract class for PropertySetter recieving a value as
922
1026
constructor argument instead of a class attribute."""
 
1027
 
923
1028
        def __init__(self, value):
924
1029
            self.value_to_set = value
925
1030
 
932
1037
    class SetChecker(PropertySetterValue):
933
1038
        propname = "Checker"
934
1039
 
935
 
 
936
1040
    class SetHost(PropertySetterValue):
937
1041
        propname = "Host"
938
1042
 
939
 
 
940
1043
    class SetSecret(PropertySetterValue):
941
1044
        propname = "Secret"
942
1045
 
950
1053
            self._vts = value.read()
951
1054
            value.close()
952
1055
 
953
 
 
954
1056
    class PropertySetterValueMilliseconds(PropertySetterValue):
955
1057
        """Abstract class for PropertySetterValue taking a value
956
1058
argument as a datetime.timedelta() but should store it as
965
1067
            "When setting, convert value from a datetime.timedelta"
966
1068
            self._vts = int(round(value.total_seconds() * 1000))
967
1069
 
968
 
 
969
1070
    class SetTimeout(PropertySetterValueMilliseconds):
970
1071
        propname = "Timeout"
971
1072
 
972
 
 
973
1073
    class SetExtendedTimeout(PropertySetterValueMilliseconds):
974
1074
        propname = "ExtendedTimeout"
975
1075
 
976
 
 
977
1076
    class SetInterval(PropertySetterValueMilliseconds):
978
1077
        propname = "Interval"
979
1078
 
980
 
 
981
1079
    class SetApprovalDelay(PropertySetterValueMilliseconds):
982
1080
        propname = "ApprovalDelay"
983
1081
 
984
 
 
985
1082
    class SetApprovalDuration(PropertySetterValueMilliseconds):
986
1083
        propname = "ApprovalDuration"
987
1084
 
1021
1118
                                                     "output"))
1022
1119
 
1023
1120
 
1024
 
class Unique(object):
 
1121
class Unique:
1025
1122
    """Class for objects which exist only to be unique objects, since
1026
1123
unittest.mock.sentinel only exists in Python 3.3"""
1027
1124
 
1311
1408
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1312
1409
 
1313
1410
    def MockDBusPython_func(self, func):
1314
 
        class mock_dbus_python(object):
 
1411
        class mock_dbus_python:
1315
1412
            """mock dbus-python module"""
1316
 
            class exceptions(object):
 
1413
            class exceptions:
1317
1414
                """Pseudo-namespace"""
1318
1415
                class DBusException(Exception):
1319
1416
                    pass
1320
 
            class SystemBus(object):
 
1417
            class SystemBus:
1321
1418
                @staticmethod
1322
1419
                def get_object(busname, objectpath):
1323
1420
                    DBusObject = collections.namedtuple(
1337
1434
                                    dbus_interface=dbus_interface)
1338
1435
                    return DBusObject(methodname=method,
1339
1436
                                      Set=set_property)
1340
 
            class Boolean(object):
 
1437
            class Boolean:
1341
1438
                def __init__(self, value):
1342
1439
                    self.value = bool(value)
1343
1440
                def __bool__(self):
1527
1624
        finally:
1528
1625
            dbus_logger.removeFilter(counting_handler)
1529
1626
 
1530
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1627
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1531
1628
 
1532
1629
        # Make sure the dbus logger was suppressed
1533
1630
        self.assertEqual(0, counting_handler.count)
1555
1652
            self.call_method(bus, "methodname", "busname",
1556
1653
                             "objectpath", "interface")
1557
1654
 
1558
 
    class fake_dbus_python_raises_exception_on_connect(object):
 
1655
    class fake_dbus_python_raises_exception_on_connect:
1559
1656
        """fake dbus-python module"""
1560
 
        class exceptions(object):
 
1657
        class exceptions:
1561
1658
            """Pseudo-namespace"""
1562
1659
            class DBusException(Exception):
1563
1660
                pass
1571
1668
 
1572
1669
 
1573
1670
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1574
 
    class mock_dbus_python(object):
 
1671
    class mock_dbus_python:
1575
1672
        """mock dbus-python modules"""
1576
 
        class SystemBus(object):
 
1673
        class SystemBus:
1577
1674
            @staticmethod
1578
1675
            def get_object(busname, objectpath):
1579
1676
                return Unique()
1625
1722
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1626
1723
 
1627
1724
    def Stub_pydbus_func(self, func):
1628
 
        class stub_pydbus(object):
 
1725
        class stub_pydbus:
1629
1726
            """stub pydbus module"""
1630
 
            class SystemBus(object):
 
1727
            class SystemBus:
1631
1728
                @staticmethod
1632
1729
                def get(busname, objectpath):
1633
1730
                    DBusObject = collections.namedtuple(
1658
1755
        self.assertIs(ret, expected_method_return)
1659
1756
 
1660
1757
    def test_call_method_handles_exception(self):
1661
 
        dbus_logger = logging.getLogger("dbus.proxies")
1662
 
 
1663
1758
        def func():
1664
1759
            raise gi.repository.GLib.Error()
1665
1760
 
1670
1765
            self.call_method(bus, "methodname", "busname",
1671
1766
                             "objectpath", "interface")
1672
1767
 
1673
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1768
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1674
1769
 
1675
1770
    def test_get_converts_to_correct_exception(self):
1676
1771
        bus = pydbus_adapter.SystemBus(
1679
1774
            self.call_method(bus, "methodname", "busname",
1680
1775
                             "objectpath", "interface")
1681
1776
 
1682
 
    class fake_pydbus_raises_exception_on_connect(object):
 
1777
    class fake_pydbus_raises_exception_on_connect:
1683
1778
        """fake dbus-python module"""
1684
1779
        @classmethod
1685
1780
        def SystemBus(cls):
1689
1784
            return Bus(get=get)
1690
1785
 
1691
1786
    def test_set_property_uses_setattr(self):
1692
 
        class Object(object):
 
1787
        class Object:
1693
1788
            pass
1694
1789
        obj = Object()
1695
 
        class pydbus_spy(object):
1696
 
            class SystemBus(object):
 
1790
        class pydbus_spy:
 
1791
            class SystemBus:
1697
1792
                @staticmethod
1698
1793
                def get(busname, objectpath):
1699
1794
                    return {"interface": obj}
1706
1801
    def test_get_suppresses_xml_deprecation_warning(self):
1707
1802
        if sys.version_info.major >= 3:
1708
1803
            return
1709
 
        class stub_pydbus_get(object):
1710
 
            class SystemBus(object):
 
1804
        class stub_pydbus_get:
 
1805
            class SystemBus:
1711
1806
                @staticmethod
1712
1807
                def get(busname, objectpath):
1713
1808
                    warnings.warn_explicit(
1721
1816
 
1722
1817
 
1723
1818
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1724
 
    class stub_pydbus(object):
 
1819
    class stub_pydbus:
1725
1820
        """stub pydbus module"""
1726
 
        class SystemBus(object):
 
1821
        class SystemBus:
1727
1822
            @staticmethod
1728
1823
            def get(busname, objectpath):
1729
1824
                return Unique()
1771
1866
        self.assertIs(obj1, obj1b)
1772
1867
 
1773
1868
 
 
1869
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
 
1870
 
 
1871
    class dummy_dbussy:
 
1872
        class DBUS:
 
1873
            class ObjectPath(str):
 
1874
                pass
 
1875
        class DBusError(Exception):
 
1876
            pass
 
1877
 
 
1878
    def fake_ravel_func(self, func):
 
1879
        class fake_ravel:
 
1880
            @staticmethod
 
1881
            def system_bus():
 
1882
                class DBusInterfaceProxy:
 
1883
                    @staticmethod
 
1884
                    def methodname(*args):
 
1885
                        return [func(*args)]
 
1886
                class DBusObject:
 
1887
                    @staticmethod
 
1888
                    def get_interface(interface):
 
1889
                        if interface == "interface":
 
1890
                            return DBusInterfaceProxy()
 
1891
                return {"busname": {"objectpath": DBusObject()}}
 
1892
        return fake_ravel
 
1893
 
 
1894
    def call_method(self, bus, methodname, busname, objectpath,
 
1895
                    interface, *args):
 
1896
        with self.assertLogs(log, logging.DEBUG):
 
1897
            return bus.call_method(methodname, busname, objectpath,
 
1898
                                   interface, *args)
 
1899
 
 
1900
    def test_call_method_returns(self):
 
1901
        expected_method_return = Unique()
 
1902
        method_args = (Unique(), Unique())
 
1903
        def func(*args):
 
1904
            self.assertEqual(len(method_args), len(args))
 
1905
            for marg, arg in zip(method_args, args):
 
1906
                self.assertIs(marg, arg)
 
1907
            return expected_method_return
 
1908
        fake_ravel = self.fake_ravel_func(func)
 
1909
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1910
        ret = self.call_method(bus, "methodname", "busname",
 
1911
                               "objectpath", "interface",
 
1912
                               *method_args)
 
1913
        self.assertIs(ret, expected_method_return)
 
1914
 
 
1915
    def test_call_method_filters_objectpath(self):
 
1916
        def func():
 
1917
            return method_return
 
1918
        fake_ravel = self.fake_ravel_func(func)
 
1919
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1920
        method_return = (self.dummy_dbussy.DBUS
 
1921
                         .ObjectPath("objectpath"))
 
1922
        ret = self.call_method(bus, "methodname", "busname",
 
1923
                               "objectpath", "interface")
 
1924
        self.assertEqual("objectpath", ret)
 
1925
        self.assertNotIsInstance(ret,
 
1926
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
1927
 
 
1928
    def test_call_method_filters_objectpaths_in_dict(self):
 
1929
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1930
        def func():
 
1931
            return method_return
 
1932
        fake_ravel = self.fake_ravel_func(func)
 
1933
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1934
        method_return = {
 
1935
            ObjectPath("objectpath_key_1"):
 
1936
            ObjectPath("objectpath_value_1"),
 
1937
            ObjectPath("objectpath_key_2"):
 
1938
            ObjectPath("objectpath_value_2"),
 
1939
        }
 
1940
        ret = self.call_method(bus, "methodname", "busname",
 
1941
                               "objectpath", "interface")
 
1942
        expected_method_return = {str(key): str(value)
 
1943
                                  for key, value in
 
1944
                                  method_return.items()}
 
1945
        for key, value in ret.items():
 
1946
            self.assertNotIsInstance(key, ObjectPath)
 
1947
            self.assertNotIsInstance(value, ObjectPath)
 
1948
        self.assertEqual(expected_method_return, ret)
 
1949
        self.assertIsInstance(ret, dict)
 
1950
 
 
1951
    def test_call_method_filters_objectpaths_in_dict_in_dict(self):
 
1952
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1953
        def func():
 
1954
            return method_return
 
1955
        fake_ravel = self.fake_ravel_func(func)
 
1956
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1957
        method_return = {
 
1958
            ObjectPath("key1"): {
 
1959
                ObjectPath("key11"): ObjectPath("value11"),
 
1960
                ObjectPath("key12"): ObjectPath("value12"),
 
1961
            },
 
1962
            ObjectPath("key2"): {
 
1963
                ObjectPath("key21"): ObjectPath("value21"),
 
1964
                ObjectPath("key22"): ObjectPath("value22"),
 
1965
            },
 
1966
        }
 
1967
        ret = self.call_method(bus, "methodname", "busname",
 
1968
                               "objectpath", "interface")
 
1969
        expected_method_return = {
 
1970
            "key1": {"key11": "value11",
 
1971
                     "key12": "value12"},
 
1972
            "key2": {"key21": "value21",
 
1973
                     "key22": "value22"},
 
1974
        }
 
1975
        self.assertEqual(expected_method_return, ret)
 
1976
        for key, value in ret.items():
 
1977
            self.assertIsInstance(value, dict)
 
1978
            self.assertEqual(expected_method_return[key], value)
 
1979
            self.assertNotIsInstance(key, ObjectPath)
 
1980
            for inner_key, inner_value in value.items():
 
1981
                self.assertIsInstance(value, dict)
 
1982
                self.assertEqual(
 
1983
                    expected_method_return[key][inner_key],
 
1984
                    inner_value)
 
1985
                self.assertNotIsInstance(key, ObjectPath)
 
1986
 
 
1987
    def test_call_method_filters_objectpaths_in_dict_three_deep(self):
 
1988
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1989
        def func():
 
1990
            return method_return
 
1991
        fake_ravel = self.fake_ravel_func(func)
 
1992
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1993
        method_return = {
 
1994
            ObjectPath("key1"): {
 
1995
                ObjectPath("key2"): {
 
1996
                    ObjectPath("key3"): ObjectPath("value"),
 
1997
                },
 
1998
            },
 
1999
        }
 
2000
        ret = self.call_method(bus, "methodname", "busname",
 
2001
                               "objectpath", "interface")
 
2002
        expected_method_return = {"key1": {"key2": {"key3": "value"}}}
 
2003
        self.assertEqual(expected_method_return, ret)
 
2004
        self.assertIsInstance(ret, dict)
 
2005
        self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
 
2006
        self.assertIsInstance(ret["key1"], dict)
 
2007
        self.assertNotIsInstance(next(iter(ret["key1"].keys())),
 
2008
                                 ObjectPath)
 
2009
        self.assertIsInstance(ret["key1"]["key2"], dict)
 
2010
        self.assertNotIsInstance(
 
2011
            next(iter(ret["key1"]["key2"].keys())),
 
2012
            ObjectPath)
 
2013
        self.assertEqual("value", ret["key1"]["key2"]["key3"])
 
2014
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
 
2015
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
2016
 
 
2017
    def test_call_method_handles_exception(self):
 
2018
        def func():
 
2019
            raise self.dummy_dbussy.DBusError()
 
2020
 
 
2021
        fake_ravel = self.fake_ravel_func(func)
 
2022
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
2023
 
 
2024
        with self.assertRaises(dbus.Error) as e:
 
2025
            self.call_method(bus, "methodname", "busname",
 
2026
                             "objectpath", "interface")
 
2027
 
 
2028
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
2029
 
 
2030
    def test_get_object_converts_to_correct_exception(self):
 
2031
        class fake_ravel_raises_exception_on_connect:
 
2032
            @staticmethod
 
2033
            def system_bus():
 
2034
                class Bus:
 
2035
                    @staticmethod
 
2036
                    def __getitem__(key):
 
2037
                        if key == "objectpath":
 
2038
                            raise self.dummy_dbussy.DBusError()
 
2039
                        raise Exception(key)
 
2040
                return {"busname": Bus()}
 
2041
        def func():
 
2042
            raise self.dummy_dbussy.DBusError()
 
2043
        bus = dbussy_adapter.SystemBus(
 
2044
            self.dummy_dbussy,
 
2045
            fake_ravel_raises_exception_on_connect)
 
2046
        with self.assertRaises(dbus.ConnectFailed):
 
2047
            self.call_method(bus, "methodname", "busname",
 
2048
                             "objectpath", "interface")
 
2049
 
 
2050
 
1774
2051
class Test_commands_from_options(unittest.TestCase):
1775
2052
 
1776
2053
    def setUp(self):
2150
2427
        busname = "se.recompile.Mandos"
2151
2428
        client_interface = "se.recompile.Mandos.Client"
2152
2429
        command.Approve().run(self.bus.clients, self.bus)
 
2430
        self.assertTrue(self.bus.clients)
2153
2431
        for clientpath in self.bus.clients:
2154
2432
            self.assertIn(("Approve", busname, clientpath,
2155
2433
                           client_interface, (True,)), self.bus.calls)
2158
2436
        busname = "se.recompile.Mandos"
2159
2437
        client_interface = "se.recompile.Mandos.Client"
2160
2438
        command.Deny().run(self.bus.clients, self.bus)
 
2439
        self.assertTrue(self.bus.clients)
2161
2440
        for clientpath in self.bus.clients:
2162
2441
            self.assertIn(("Approve", busname, clientpath,
2163
2442
                           client_interface, (False,)),
2164
2443
                          self.bus.calls)
2165
2444
 
2166
2445
    def test_Remove(self):
 
2446
        busname = "se.recompile.Mandos"
 
2447
        server_path = "/"
 
2448
        server_interface = "se.recompile.Mandos"
 
2449
        orig_clients = self.bus.clients.copy()
2167
2450
        command.Remove().run(self.bus.clients, self.bus)
2168
 
        for clientpath in self.bus.clients:
2169
 
            self.assertIn(("RemoveClient", dbus_busname,
2170
 
                           dbus_server_path, dbus_server_interface,
 
2451
        self.assertFalse(self.bus.clients)
 
2452
        for clientpath in orig_clients:
 
2453
            self.assertIn(("RemoveClient", busname,
 
2454
                           server_path, server_interface,
2171
2455
                           (clientpath,)), self.bus.calls)
2172
2456
 
2173
2457
    expected_json = {
2375
2659
        else:
2376
2660
            cmd_args = [() for x in range(len(self.values_to_get))]
2377
2661
            values_to_get = self.values_to_get
 
2662
        self.assertTrue(values_to_get)
2378
2663
        for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2379
2664
            for clientpath in self.bus.clients:
2380
2665
                self.bus.clients[clientpath][self.propname] = (
2381
2666
                    Unique())
2382
2667
            self.command(*cmd_arg).run(self.bus.clients, self.bus)
 
2668
            self.assertTrue(self.bus.clients)
2383
2669
            for clientpath in self.bus.clients:
2384
2670
                value = (self.bus.clients[clientpath]
2385
2671
                         [self.propname])
2444
2730
class TestSetSecretCmd(TestPropertySetterCmd):
2445
2731
    command = command.SetSecret
2446
2732
    propname = "Secret"
2447
 
    values_to_set = [io.BytesIO(b""),
2448
 
                     io.BytesIO(b"secret\0xyzzy\nbar")]
2449
 
    values_to_get = [f.getvalue() for f in values_to_set]
 
2733
    def __init__(self, *args, **kwargs):
 
2734
        self.values_to_set = [io.BytesIO(b""),
 
2735
                              io.BytesIO(b"secret\0xyzzy\nbar")]
 
2736
        self.values_to_get = [f.getvalue() for f in
 
2737
                              self.values_to_set]
 
2738
        super(TestSetSecretCmd, self).__init__(*args, **kwargs)
2450
2739
 
2451
2740
 
2452
2741
class TestSetTimeoutCmd(TestPropertySetterCmd):
2505
2794
 
2506
2795
 
2507
2796
 
2508
 
def should_only_run_tests():
 
2797
def parse_test_args():
 
2798
    # type: () -> argparse.Namespace
2509
2799
    parser = argparse.ArgumentParser(add_help=False)
2510
 
    parser.add_argument("--check", action='store_true')
 
2800
    parser.add_argument("--check", action="store_true")
 
2801
    parser.add_argument("--prefix", )
2511
2802
    args, unknown_args = parser.parse_known_args()
2512
 
    run_tests = args.check
2513
 
    if run_tests:
2514
 
        # Remove --check argument from sys.argv
 
2803
    if args.check:
 
2804
        # Remove test options from sys.argv
2515
2805
        sys.argv[1:] = unknown_args
2516
 
    return run_tests
 
2806
    return args
2517
2807
 
2518
2808
# Add all tests from doctest strings
2519
2809
def load_tests(loader, tests, none):
2522
2812
    return tests
2523
2813
 
2524
2814
if __name__ == "__main__":
 
2815
    options = parse_test_args()
2525
2816
    try:
2526
 
        if should_only_run_tests():
2527
 
            # Call using ./tdd-python-script --check [--verbose]
2528
 
            unittest.main()
 
2817
        if options.check:
 
2818
            extra_test_prefix = options.prefix
 
2819
            if extra_test_prefix is not None:
 
2820
                if not (unittest.main(argv=[""], exit=False)
 
2821
                        .result.wasSuccessful()):
 
2822
                    sys.exit(1)
 
2823
                class ExtraTestLoader(unittest.TestLoader):
 
2824
                    testMethodPrefix = extra_test_prefix
 
2825
                # Call using ./scriptname --check [--verbose]
 
2826
                unittest.main(argv=[""], testLoader=ExtraTestLoader())
 
2827
            else:
 
2828
                unittest.main(argv=[""])
2529
2829
        else:
2530
2830
            main()
2531
2831
    finally:
2532
2832
        logging.shutdown()
 
2833
 
 
2834
# Local Variables:
 
2835
# run-tests:
 
2836
# (lambda (&optional extra)
 
2837
#   (if (not (funcall run-tests-in-test-buffer default-directory
 
2838
#             extra))
 
2839
#       (funcall show-test-buffer-in-test-window)
 
2840
#     (funcall remove-test-window)
 
2841
#     (if extra (message "Extra tests run successfully!"))))
 
2842
# run-tests-in-test-buffer:
 
2843
# (lambda (dir &optional extra)
 
2844
#   (with-current-buffer (get-buffer-create "*Test*")
 
2845
#     (setq buffer-read-only nil
 
2846
#           default-directory dir)
 
2847
#     (erase-buffer)
 
2848
#     (compilation-mode))
 
2849
#   (let ((process-result
 
2850
#          (let ((inhibit-read-only t))
 
2851
#            (process-file-shell-command
 
2852
#             (funcall get-command-line extra) nil "*Test*"))))
 
2853
#     (and (numberp process-result)
 
2854
#          (= process-result 0))))
 
2855
# get-command-line:
 
2856
# (lambda (&optional extra)
 
2857
#   (let ((quoted-script
 
2858
#          (shell-quote-argument (funcall get-script-name))))
 
2859
#     (format
 
2860
#      (concat "%s --check" (if extra " --prefix=atest" ""))
 
2861
#      quoted-script)))
 
2862
# get-script-name:
 
2863
# (lambda ()
 
2864
#   (if (fboundp 'file-local-name)
 
2865
#       (file-local-name (buffer-file-name))
 
2866
#     (or (file-remote-p (buffer-file-name) 'localname)
 
2867
#         (buffer-file-name))))
 
2868
# remove-test-window:
 
2869
# (lambda ()
 
2870
#   (let ((test-window (get-buffer-window "*Test*")))
 
2871
#     (if test-window (delete-window test-window))))
 
2872
# show-test-buffer-in-test-window:
 
2873
# (lambda ()
 
2874
#   (when (not (get-buffer-window-list "*Test*"))
 
2875
#     (setq next-error-last-buffer (get-buffer "*Test*"))
 
2876
#     (let* ((side (if (>= (window-width) 146) 'right 'bottom))
 
2877
#            (display-buffer-overriding-action
 
2878
#             `((display-buffer-in-side-window) (side . ,side)
 
2879
#               (window-height . fit-window-to-buffer)
 
2880
#               (window-width . fit-window-to-buffer))))
 
2881
#       (display-buffer "*Test*"))))
 
2882
# eval:
 
2883
# (progn
 
2884
#   (let* ((run-extra-tests (lambda () (interactive)
 
2885
#                             (funcall run-tests t)))
 
2886
#          (inner-keymap `(keymap (116 . ,run-extra-tests))) ; t
 
2887
#          (outer-keymap `(keymap (3 . ,inner-keymap))))     ; C-c
 
2888
#     (setq minor-mode-overriding-map-alist
 
2889
#           (cons `(run-tests . ,outer-keymap)
 
2890
#                 minor-mode-overriding-map-alist)))
 
2891
#   (add-hook 'after-save-hook run-tests 90 t))
 
2892
# End: