/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
1
#!/usr/bin/python3 -bbI
2
 
# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-
3
 
#
4
 
# Mandos Monitor - Control and monitor the Mandos server
5
 
#
6
 
# Copyright © 2008-2019 Teddy Hogeborn
7
 
# Copyright © 2008-2019 Björn Påhlsson
 
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
 
49
48
if sys.version_info.major == 2:
50
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
51
66
 
52
67
try:
53
 
    import pydbus
54
 
    import gi
55
 
    dbus_python = None
 
68
    import dbussy
 
69
    import ravel
56
70
except ImportError:
57
 
    import dbus as dbus_python
58
 
    pydbus = None
59
 
    class gi:
60
 
        """Dummy gi module, for the tests"""
61
 
        class repository:
62
 
            class GLib:
63
 
                class Error(Exception):
64
 
                    pass
 
71
    try:
 
72
        import pydbus
 
73
        import gi
 
74
    except ImportError:
 
75
        import dbus as dbus_python
 
76
 
65
77
 
66
78
# Show warnings by default
67
79
if not sys.warnoptions:
68
80
    import warnings
69
81
    warnings.simplefilter("default")
70
82
 
71
 
log = logging.getLogger(sys.argv[0])
72
 
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
73
85
                    format="%(message)s") # Show basic log messages
74
86
 
75
87
logging.captureWarnings(True)   # Show warnings via the logging system
76
88
 
77
89
if sys.version_info.major == 2:
78
 
    str = unicode
79
90
    import StringIO
80
91
    io.StringIO = StringIO.StringIO
81
92
 
82
93
locale.setlocale(locale.LC_ALL, "")
83
94
 
84
 
version = "1.8.9"
 
95
version = "1.8.17"
85
96
 
86
97
 
87
98
def main():
94
105
    clientnames = options.client
95
106
 
96
107
    if options.debug:
97
 
        log.setLevel(logging.DEBUG)
 
108
        logging.getLogger("").setLevel(logging.DEBUG)
98
109
 
99
 
    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:
100
113
        bus = pydbus_adapter.CachingBus(pydbus)
101
114
    else:
102
115
        bus = dbus_python_adapter.CachingBus(dbus_python)
246
259
        return rfc3339_duration_to_delta(interval)
247
260
    except ValueError as e:
248
261
        log.warning("%s - Parsing as pre-1.6.1 interval instead",
249
 
                    ' '.join(e.args))
 
262
                    " ".join(e.args))
250
263
    return parse_pre_1_6_1_interval(interval)
251
264
 
252
265
 
382
395
 
383
396
 
384
397
def parse_pre_1_6_1_interval(interval):
385
 
    """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,
386
399
    and return a datetime.timedelta
387
400
 
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
    >>> 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)
401
414
    True
402
415
    >>> # Ignore unknown characters, allow any order and repetitions
403
 
    >>> 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)
404
418
    True
405
419
 
406
420
    """
475
489
    class SystemBus:
476
490
 
477
491
        object_manager_iface = "org.freedesktop.DBus.ObjectManager"
 
492
 
478
493
        def get_managed_objects(self, busname, objectpath):
479
494
            return self.call_method("GetManagedObjects", busname,
480
495
                                    objectpath,
481
496
                                    self.object_manager_iface)
482
497
 
483
498
        properties_iface = "org.freedesktop.DBus.Properties"
 
499
 
484
500
        def set_property(self, busname, objectpath, interface, key,
485
501
                         value):
486
502
            self.call_method("Set", busname, objectpath,
487
503
                             self.properties_iface, interface, key,
488
504
                             value)
489
505
 
 
506
        def call_method(self, methodname, busname, objectpath,
 
507
                        interface, *args):
 
508
            raise NotImplementedError()
490
509
 
491
510
    class MandosBus(SystemBus):
492
511
        busname_domain = "se.recompile"
586
605
 
587
606
    class SilenceLogger:
588
607
        "Simple context manager to silence a particular logger"
 
608
 
589
609
        def __init__(self, loggername):
590
610
            self.logger = logging.getLogger(loggername)
591
611
 
601
621
        def __exit__(self, exc_type, exc_val, exc_tb):
602
622
            self.logger.removeFilter(self.nullfilter)
603
623
 
604
 
 
605
624
    class CachingBus(SystemBus):
606
625
        """A caching layer for dbus_python_adapter.SystemBus"""
 
626
 
607
627
        def __init__(self, *args, **kwargs):
608
628
            self.object_cache = {}
609
629
            super(dbus_python_adapter.CachingBus,
610
630
                  self).__init__(*args, **kwargs)
 
631
 
611
632
        def get_object(self, busname, objectpath):
612
633
            try:
613
634
                return self.object_cache[(busname, objectpath)]
615
636
                new_object = super(
616
637
                    dbus_python_adapter.CachingBus,
617
638
                    self).get_object(busname, objectpath)
618
 
                self.object_cache[(busname, objectpath)]  = new_object
 
639
                self.object_cache[(busname, objectpath)] = new_object
619
640
                return new_object
620
641
 
621
642
 
668
689
 
669
690
    class CachingBus(SystemBus):
670
691
        """A caching layer for pydbus_adapter.SystemBus"""
 
692
 
671
693
        def __init__(self, *args, **kwargs):
672
694
            self.object_cache = {}
673
695
            super(pydbus_adapter.CachingBus,
674
696
                  self).__init__(*args, **kwargs)
 
697
 
675
698
        def get(self, busname, objectpath):
676
699
            try:
677
700
                return self.object_cache[(busname, objectpath)]
678
701
            except KeyError:
679
702
                new_object = (super(pydbus_adapter.CachingBus, self)
680
703
                              .get(busname, objectpath))
681
 
                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
682
797
                return new_object
683
798
 
684
799
 
720
835
 
721
836
    class Base:
722
837
        """Abstract base class for commands"""
 
838
 
723
839
        def run(self, clients, bus=None):
724
840
            """Normal commands should implement run_on_one_client(),
725
841
but commands which want to operate on all clients at the same time can
729
845
            for client, properties in clients.items():
730
846
                self.run_on_one_client(client, properties)
731
847
 
732
 
 
733
848
    class IsEnabled(Base):
734
849
        def run(self, clients, bus=None):
735
850
            properties = next(iter(clients.values()))
737
852
                sys.exit(0)
738
853
            sys.exit(1)
739
854
 
740
 
 
741
855
    class Approve(Base):
742
856
        def run_on_one_client(self, client, properties):
743
857
            self.bus.call_client_method(client, "Approve", True)
744
858
 
745
 
 
746
859
    class Deny(Base):
747
860
        def run_on_one_client(self, client, properties):
748
861
            self.bus.call_client_method(client, "Approve", False)
749
862
 
750
 
 
751
863
    class Remove(Base):
752
864
        def run(self, clients, bus):
753
865
            for clientpath in frozenset(clients.keys()):
754
866
                bus.call_server_method("RemoveClient", clientpath)
755
867
 
756
 
 
757
868
    class Output(Base):
758
869
        """Abstract class for commands outputting client details"""
759
870
        all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
765
876
                        "Checker", "ExtendedTimeout", "Expires",
766
877
                        "LastCheckerStatus")
767
878
 
768
 
 
769
879
    class DumpJSON(Output):
770
880
        def run(self, clients, bus=None):
771
881
            data = {properties["Name"]:
772
882
                    {key: properties[key]
773
883
                     for key in self.all_keywords}
774
884
                    for properties in clients.values()}
775
 
            print(json.dumps(data, indent=4, separators=(',', ': ')))
776
 
 
 
885
            print(json.dumps(data, indent=4, separators=(",", ": ")))
777
886
 
778
887
    class PrintTable(Output):
779
888
        def __init__(self, verbose=False):
820
929
 
821
930
            if sys.version_info.major == 2:
822
931
                __unicode__ = __str__
 
932
 
823
933
                def __str__(self):
824
934
                    return str(self).encode(
825
935
                        locale.getpreferredencoding())
871
981
                                minutes=(td.seconds % 3600) // 60,
872
982
                                seconds=td.seconds % 60))
873
983
 
874
 
 
875
984
    class PropertySetter(Base):
876
985
        "Abstract class for Actions for setting one client property"
877
986
 
884
993
        def propname(self):
885
994
            raise NotImplementedError()
886
995
 
887
 
 
888
996
    class Enable(PropertySetter):
889
997
        propname = "Enabled"
890
998
        value_to_set = True
891
999
 
892
 
 
893
1000
    class Disable(PropertySetter):
894
1001
        propname = "Enabled"
895
1002
        value_to_set = False
896
1003
 
897
 
 
898
1004
    class BumpTimeout(PropertySetter):
899
1005
        propname = "LastCheckedOK"
900
1006
        value_to_set = ""
901
1007
 
902
 
 
903
1008
    class StartChecker(PropertySetter):
904
1009
        propname = "CheckerRunning"
905
1010
        value_to_set = True
906
1011
 
907
 
 
908
1012
    class StopChecker(PropertySetter):
909
1013
        propname = "CheckerRunning"
910
1014
        value_to_set = False
911
1015
 
912
 
 
913
1016
    class ApproveByDefault(PropertySetter):
914
1017
        propname = "ApprovedByDefault"
915
1018
        value_to_set = True
916
1019
 
917
 
 
918
1020
    class DenyByDefault(PropertySetter):
919
1021
        propname = "ApprovedByDefault"
920
1022
        value_to_set = False
921
1023
 
922
 
 
923
1024
    class PropertySetterValue(PropertySetter):
924
1025
        """Abstract class for PropertySetter recieving a value as
925
1026
constructor argument instead of a class attribute."""
 
1027
 
926
1028
        def __init__(self, value):
927
1029
            self.value_to_set = value
928
1030
 
935
1037
    class SetChecker(PropertySetterValue):
936
1038
        propname = "Checker"
937
1039
 
938
 
 
939
1040
    class SetHost(PropertySetterValue):
940
1041
        propname = "Host"
941
1042
 
942
 
 
943
1043
    class SetSecret(PropertySetterValue):
944
1044
        propname = "Secret"
945
1045
 
953
1053
            self._vts = value.read()
954
1054
            value.close()
955
1055
 
956
 
 
957
1056
    class PropertySetterValueMilliseconds(PropertySetterValue):
958
1057
        """Abstract class for PropertySetterValue taking a value
959
1058
argument as a datetime.timedelta() but should store it as
968
1067
            "When setting, convert value from a datetime.timedelta"
969
1068
            self._vts = int(round(value.total_seconds() * 1000))
970
1069
 
971
 
 
972
1070
    class SetTimeout(PropertySetterValueMilliseconds):
973
1071
        propname = "Timeout"
974
1072
 
975
 
 
976
1073
    class SetExtendedTimeout(PropertySetterValueMilliseconds):
977
1074
        propname = "ExtendedTimeout"
978
1075
 
979
 
 
980
1076
    class SetInterval(PropertySetterValueMilliseconds):
981
1077
        propname = "Interval"
982
1078
 
983
 
 
984
1079
    class SetApprovalDelay(PropertySetterValueMilliseconds):
985
1080
        propname = "ApprovalDelay"
986
1081
 
987
 
 
988
1082
    class SetApprovalDuration(PropertySetterValueMilliseconds):
989
1083
        propname = "ApprovalDuration"
990
1084
 
1530
1624
        finally:
1531
1625
            dbus_logger.removeFilter(counting_handler)
1532
1626
 
1533
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1627
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1534
1628
 
1535
1629
        # Make sure the dbus logger was suppressed
1536
1630
        self.assertEqual(0, counting_handler.count)
1661
1755
        self.assertIs(ret, expected_method_return)
1662
1756
 
1663
1757
    def test_call_method_handles_exception(self):
1664
 
        dbus_logger = logging.getLogger("dbus.proxies")
1665
 
 
1666
1758
        def func():
1667
1759
            raise gi.repository.GLib.Error()
1668
1760
 
1673
1765
            self.call_method(bus, "methodname", "busname",
1674
1766
                             "objectpath", "interface")
1675
1767
 
1676
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1768
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1677
1769
 
1678
1770
    def test_get_converts_to_correct_exception(self):
1679
1771
        bus = pydbus_adapter.SystemBus(
1774
1866
        self.assertIs(obj1, obj1b)
1775
1867
 
1776
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
 
1777
2051
class Test_commands_from_options(unittest.TestCase):
1778
2052
 
1779
2053
    def setUp(self):
2153
2427
        busname = "se.recompile.Mandos"
2154
2428
        client_interface = "se.recompile.Mandos.Client"
2155
2429
        command.Approve().run(self.bus.clients, self.bus)
 
2430
        self.assertTrue(self.bus.clients)
2156
2431
        for clientpath in self.bus.clients:
2157
2432
            self.assertIn(("Approve", busname, clientpath,
2158
2433
                           client_interface, (True,)), self.bus.calls)
2161
2436
        busname = "se.recompile.Mandos"
2162
2437
        client_interface = "se.recompile.Mandos.Client"
2163
2438
        command.Deny().run(self.bus.clients, self.bus)
 
2439
        self.assertTrue(self.bus.clients)
2164
2440
        for clientpath in self.bus.clients:
2165
2441
            self.assertIn(("Approve", busname, clientpath,
2166
2442
                           client_interface, (False,)),
2167
2443
                          self.bus.calls)
2168
2444
 
2169
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()
2170
2450
        command.Remove().run(self.bus.clients, self.bus)
2171
 
        for clientpath in self.bus.clients:
2172
 
            self.assertIn(("RemoveClient", dbus_busname,
2173
 
                           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,
2174
2455
                           (clientpath,)), self.bus.calls)
2175
2456
 
2176
2457
    expected_json = {
2378
2659
        else:
2379
2660
            cmd_args = [() for x in range(len(self.values_to_get))]
2380
2661
            values_to_get = self.values_to_get
 
2662
        self.assertTrue(values_to_get)
2381
2663
        for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2382
2664
            for clientpath in self.bus.clients:
2383
2665
                self.bus.clients[clientpath][self.propname] = (
2384
2666
                    Unique())
2385
2667
            self.command(*cmd_arg).run(self.bus.clients, self.bus)
 
2668
            self.assertTrue(self.bus.clients)
2386
2669
            for clientpath in self.bus.clients:
2387
2670
                value = (self.bus.clients[clientpath]
2388
2671
                         [self.propname])
2447
2730
class TestSetSecretCmd(TestPropertySetterCmd):
2448
2731
    command = command.SetSecret
2449
2732
    propname = "Secret"
2450
 
    values_to_set = [io.BytesIO(b""),
2451
 
                     io.BytesIO(b"secret\0xyzzy\nbar")]
2452
 
    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)
2453
2739
 
2454
2740
 
2455
2741
class TestSetTimeoutCmd(TestPropertySetterCmd):
2508
2794
 
2509
2795
 
2510
2796
 
2511
 
def should_only_run_tests():
 
2797
def parse_test_args():
 
2798
    # type: () -> argparse.Namespace
2512
2799
    parser = argparse.ArgumentParser(add_help=False)
2513
 
    parser.add_argument("--check", action='store_true')
 
2800
    parser.add_argument("--check", action="store_true")
 
2801
    parser.add_argument("--prefix", )
2514
2802
    args, unknown_args = parser.parse_known_args()
2515
 
    run_tests = args.check
2516
 
    if run_tests:
2517
 
        # Remove --check argument from sys.argv
 
2803
    if args.check:
 
2804
        # Remove test options from sys.argv
2518
2805
        sys.argv[1:] = unknown_args
2519
 
    return run_tests
 
2806
    return args
2520
2807
 
2521
2808
# Add all tests from doctest strings
2522
2809
def load_tests(loader, tests, none):
2525
2812
    return tests
2526
2813
 
2527
2814
if __name__ == "__main__":
 
2815
    options = parse_test_args()
2528
2816
    try:
2529
 
        if should_only_run_tests():
2530
 
            # Call using ./tdd-python-script --check [--verbose]
2531
 
            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=[""])
2532
2829
        else:
2533
2830
            main()
2534
2831
    finally:
2535
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: