/mandos/trunk

To get this branch, use:
bzr branch http://bzr.recompile.se/loggerhead/mandos/trunk

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: teddy at recompile
  • Date: 2020-01-12 01:53:04 UTC
  • Revision ID: teddy@recompile.se-20200112015304-gqif9w4pbyix8w6b
mandos-ctl: Fix minor bug in tests

Fix two tests, the last assert of which would always erroneously
succeed.  (No change in non-test code needed.)

* mandos-ctl (Test_dbus_python_adapter_SystemBus): In the
  test_call_method_handles_exception() method, fix check for
  dbus.ConnectFailed exception.
  (Test_pydbus_adapter_SystemBus): - '' -

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
1
 
#!/usr/bin/python
 
1
#!/usr/bin/python3 -bbI
2
2
# -*- after-save-hook: (lambda () (let ((command (if (fboundp 'file-local-name) (file-local-name (buffer-file-name)) (or (file-remote-p (buffer-file-name) 'localname) (buffer-file-name))))) (if (= (progn (if (get-buffer "*Test*") (kill-buffer "*Test*")) (process-file-shell-command (format "%s --check" (shell-quote-argument command)) nil "*Test*")) 0) (let ((w (get-buffer-window "*Test*"))) (if w (delete-window w))) (progn (with-current-buffer "*Test*" (compilation-mode)) (display-buffer "*Test*" '(display-buffer-in-side-window)))))); coding: utf-8 -*-
3
3
#
4
4
# Mandos Monitor - Control and monitor the Mandos server
48
48
 
49
49
if sys.version_info.major == 2:
50
50
    __metaclass__ = type
 
51
    str = unicode
 
52
 
 
53
class gi:
 
54
    """Dummy gi module, for the tests"""
 
55
    class repository:
 
56
        class GLib:
 
57
            class Error(Exception):
 
58
                pass
 
59
dbussy = None
 
60
ravel = None
 
61
dbus_python = None
 
62
pydbus = None
51
63
 
52
64
try:
53
 
    import pydbus
54
 
    import gi
55
 
    dbus_python = None
 
65
    import dbussy
 
66
    import ravel
56
67
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
 
68
    try:
 
69
        import pydbus
 
70
        import gi
 
71
    except ImportError:
 
72
        import dbus as dbus_python
 
73
 
65
74
 
66
75
# Show warnings by default
67
76
if not sys.warnoptions:
75
84
logging.captureWarnings(True)   # Show warnings via the logging system
76
85
 
77
86
if sys.version_info.major == 2:
78
 
    str = unicode
79
87
    import StringIO
80
88
    io.StringIO = StringIO.StringIO
81
89
 
82
90
locale.setlocale(locale.LC_ALL, "")
83
91
 
84
 
version = "1.8.7"
 
92
version = "1.8.9"
85
93
 
86
94
 
87
95
def main():
96
104
    if options.debug:
97
105
        log.setLevel(logging.DEBUG)
98
106
 
99
 
    if pydbus is not None:
 
107
    if dbussy is not None and ravel is not None:
 
108
        bus = dbussy_adapter.CachingBus(dbussy, ravel)
 
109
    elif pydbus is not None:
100
110
        bus = pydbus_adapter.CachingBus(pydbus)
101
111
    else:
102
112
        bus = dbus_python_adapter.CachingBus(dbus_python)
487
497
                             self.properties_iface, interface, key,
488
498
                             value)
489
499
 
 
500
        def call_method(self, methodname, busname, objectpath,
 
501
                        interface, *args):
 
502
            raise NotImplementedError()
 
503
 
490
504
 
491
505
    class MandosBus(SystemBus):
492
506
        busname_domain = "se.recompile"
682
696
                return new_object
683
697
 
684
698
 
 
699
class dbussy_adapter:
 
700
    class SystemBus(dbus.SystemBus):
 
701
        """Use DBussy"""
 
702
 
 
703
        def __init__(self, dbussy, ravel):
 
704
            self.dbussy = dbussy
 
705
            self.ravel = ravel
 
706
            self.bus = ravel.system_bus()
 
707
 
 
708
        @contextlib.contextmanager
 
709
        def convert_exception(self, exception_class=dbus.Error):
 
710
            try:
 
711
                yield
 
712
            except self.dbussy.DBusError as e:
 
713
                # This does what "raise from" would do
 
714
                exc = exception_class(*e.args)
 
715
                exc.__cause__ = e
 
716
                raise exc
 
717
 
 
718
        def call_method(self, methodname, busname, objectpath,
 
719
                        interface, *args):
 
720
            proxy_object = self.get_object(busname, objectpath)
 
721
            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
 
722
                      interface, methodname,
 
723
                      ", ".join(repr(a) for a in args))
 
724
            iface = proxy_object.get_interface(interface)
 
725
            method = getattr(iface, methodname)
 
726
            with self.convert_exception(dbus.Error):
 
727
                value =  method(*args)
 
728
            # DBussy returns values either as an empty list or as a
 
729
            # tuple: (signature, value)
 
730
            if value:
 
731
                return self.type_filter(value[0])
 
732
 
 
733
        def get_object(self, busname, objectpath):
 
734
            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
735
                      busname, objectpath)
 
736
            with self.convert_exception(dbus.ConnectFailed):
 
737
                return self.bus[busname][objectpath]
 
738
 
 
739
        def type_filter(self, value):
 
740
            """Convert the most bothersome types to Python types"""
 
741
            if isinstance(value, tuple):
 
742
                if (len(value) == 2
 
743
                    and isinstance(value[0],
 
744
                                   self.dbussy.DBUS.Signature)):
 
745
                    return self.type_filter(value[1])
 
746
            elif isinstance(value, self.dbussy.DBUS.ObjectPath):
 
747
                return str(value)
 
748
            # Also recurse into dictionaries
 
749
            elif isinstance(value, dict):
 
750
                return {self.type_filter(key):
 
751
                        self.type_filter(subval)
 
752
                        for key, subval in value.items()}
 
753
            return value
 
754
 
 
755
        def set_property(self, busname, objectpath, interface, key,
 
756
                         value):
 
757
            proxy_object = self.get_object(busname, objectpath)
 
758
            log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
 
759
                      objectpath, self.properties_iface, interface,
 
760
                      key, value)
 
761
            if key == "Secret":
 
762
                # DBussy wants a Byte Array to be a sequence of
 
763
                # values, not a byte string
 
764
                value = tuple(value)
 
765
            setattr(proxy_object.get_interface(interface), key, value)
 
766
 
 
767
    class MandosBus(SystemBus, dbus.MandosBus):
 
768
        pass
 
769
 
 
770
    class CachingBus(MandosBus):
 
771
        """A caching layer for dbussy_adapter.MandosBus"""
 
772
        def __init__(self, *args, **kwargs):
 
773
            self.object_cache = {}
 
774
            super(dbussy_adapter.CachingBus, self).__init__(*args,
 
775
                                                            **kwargs)
 
776
        def get_object(self, busname, objectpath):
 
777
            try:
 
778
                return self.object_cache[(busname, objectpath)]
 
779
            except KeyError:
 
780
                new_object = super(
 
781
                    dbussy_adapter.CachingBus,
 
782
                    self).get_object(busname, objectpath)
 
783
                self.object_cache[(busname, objectpath)]  = new_object
 
784
                return new_object
 
785
 
 
786
 
685
787
def commands_from_options(options):
686
788
 
687
789
    commands = list(options.commands)
1530
1632
        finally:
1531
1633
            dbus_logger.removeFilter(counting_handler)
1532
1634
 
1533
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1635
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1534
1636
 
1535
1637
        # Make sure the dbus logger was suppressed
1536
1638
        self.assertEqual(0, counting_handler.count)
1673
1775
            self.call_method(bus, "methodname", "busname",
1674
1776
                             "objectpath", "interface")
1675
1777
 
1676
 
        self.assertNotIsInstance(e, dbus.ConnectFailed)
 
1778
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
1677
1779
 
1678
1780
    def test_get_converts_to_correct_exception(self):
1679
1781
        bus = pydbus_adapter.SystemBus(
1774
1876
        self.assertIs(obj1, obj1b)
1775
1877
 
1776
1878
 
 
1879
class Test_dbussy_adapter_SystemBus(TestCaseWithAssertLogs):
 
1880
 
 
1881
    class dummy_dbussy:
 
1882
        class DBUS:
 
1883
            class ObjectPath(str):
 
1884
                pass
 
1885
        class DBusError(Exception):
 
1886
            pass
 
1887
 
 
1888
    def fake_ravel_func(self, func):
 
1889
        class fake_ravel:
 
1890
            @staticmethod
 
1891
            def system_bus():
 
1892
                class DBusInterfaceProxy:
 
1893
                    @staticmethod
 
1894
                    def methodname(*args):
 
1895
                        return [func(*args)]
 
1896
                class DBusObject:
 
1897
                    @staticmethod
 
1898
                    def get_interface(interface):
 
1899
                        if interface == "interface":
 
1900
                            return DBusInterfaceProxy()
 
1901
                return {"busname": {"objectpath": DBusObject()}}
 
1902
        return fake_ravel
 
1903
 
 
1904
    def call_method(self, bus, methodname, busname, objectpath,
 
1905
                    interface, *args):
 
1906
        with self.assertLogs(log, logging.DEBUG):
 
1907
            return bus.call_method(methodname, busname, objectpath,
 
1908
                                   interface, *args)
 
1909
 
 
1910
    def test_call_method_returns(self):
 
1911
        expected_method_return = Unique()
 
1912
        method_args = (Unique(), Unique())
 
1913
        def func(*args):
 
1914
            self.assertEqual(len(method_args), len(args))
 
1915
            for marg, arg in zip(method_args, args):
 
1916
                self.assertIs(marg, arg)
 
1917
            return expected_method_return
 
1918
        fake_ravel = self.fake_ravel_func(func)
 
1919
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1920
        ret = self.call_method(bus, "methodname", "busname",
 
1921
                               "objectpath", "interface",
 
1922
                               *method_args)
 
1923
        self.assertIs(ret, expected_method_return)
 
1924
 
 
1925
    def test_call_method_filters_objectpath(self):
 
1926
        def func():
 
1927
            return method_return
 
1928
        fake_ravel = self.fake_ravel_func(func)
 
1929
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1930
        method_return = (self.dummy_dbussy.DBUS
 
1931
                         .ObjectPath("objectpath"))
 
1932
        ret = self.call_method(bus, "methodname", "busname",
 
1933
                               "objectpath", "interface")
 
1934
        self.assertEqual("objectpath", ret)
 
1935
        self.assertNotIsInstance(ret,
 
1936
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
1937
 
 
1938
    def test_call_method_filters_objectpaths_in_dict(self):
 
1939
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1940
        def func():
 
1941
            return method_return
 
1942
        fake_ravel = self.fake_ravel_func(func)
 
1943
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1944
        method_return = {
 
1945
            ObjectPath("objectpath_key_1"):
 
1946
            ObjectPath("objectpath_value_1"),
 
1947
            ObjectPath("objectpath_key_2"):
 
1948
            ObjectPath("objectpath_value_2"),
 
1949
        }
 
1950
        ret = self.call_method(bus, "methodname", "busname",
 
1951
                               "objectpath", "interface")
 
1952
        expected_method_return = {str(key): str(value)
 
1953
                                  for key, value in
 
1954
                                  method_return.items()}
 
1955
        for key, value in ret.items():
 
1956
            self.assertNotIsInstance(key, ObjectPath)
 
1957
            self.assertNotIsInstance(value, ObjectPath)
 
1958
        self.assertEqual(expected_method_return, ret)
 
1959
        self.assertIsInstance(ret, dict)
 
1960
 
 
1961
    def test_call_method_filters_objectpaths_in_dict_in_dict(self):
 
1962
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1963
        def func():
 
1964
            return method_return
 
1965
        fake_ravel = self.fake_ravel_func(func)
 
1966
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
1967
        method_return = {
 
1968
            ObjectPath("key1"): {
 
1969
                ObjectPath("key11"): ObjectPath("value11"),
 
1970
                ObjectPath("key12"): ObjectPath("value12"),
 
1971
            },
 
1972
            ObjectPath("key2"): {
 
1973
                ObjectPath("key21"): ObjectPath("value21"),
 
1974
                ObjectPath("key22"): ObjectPath("value22"),
 
1975
            },
 
1976
        }
 
1977
        ret = self.call_method(bus, "methodname", "busname",
 
1978
                               "objectpath", "interface")
 
1979
        expected_method_return = {
 
1980
            "key1": {"key11": "value11",
 
1981
                     "key12": "value12"},
 
1982
            "key2": {"key21": "value21",
 
1983
                     "key22": "value22"},
 
1984
        }
 
1985
        self.assertEqual(expected_method_return, ret)
 
1986
        for key, value in ret.items():
 
1987
            self.assertIsInstance(value, dict)
 
1988
            self.assertEqual(expected_method_return[key], value)
 
1989
            self.assertNotIsInstance(key, ObjectPath)
 
1990
            for inner_key, inner_value in value.items():
 
1991
                self.assertIsInstance(value, dict)
 
1992
                self.assertEqual(
 
1993
                    expected_method_return[key][inner_key],
 
1994
                    inner_value)
 
1995
                self.assertNotIsInstance(key, ObjectPath)
 
1996
 
 
1997
    def test_call_method_filters_objectpaths_in_dict_three_deep(self):
 
1998
        ObjectPath = self.dummy_dbussy.DBUS.ObjectPath
 
1999
        def func():
 
2000
            return method_return
 
2001
        fake_ravel = self.fake_ravel_func(func)
 
2002
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
2003
        method_return = {
 
2004
            ObjectPath("key1"): {
 
2005
                ObjectPath("key2"): {
 
2006
                    ObjectPath("key3"): ObjectPath("value"),
 
2007
                },
 
2008
            },
 
2009
        }
 
2010
        ret = self.call_method(bus, "methodname", "busname",
 
2011
                               "objectpath", "interface")
 
2012
        expected_method_return = {"key1": {"key2": {"key3": "value"}}}
 
2013
        self.assertEqual(expected_method_return, ret)
 
2014
        self.assertIsInstance(ret, dict)
 
2015
        self.assertNotIsInstance(next(iter(ret.keys())), ObjectPath)
 
2016
        self.assertIsInstance(ret["key1"], dict)
 
2017
        self.assertNotIsInstance(next(iter(ret["key1"].keys())),
 
2018
                                 ObjectPath)
 
2019
        self.assertIsInstance(ret["key1"]["key2"], dict)
 
2020
        self.assertNotIsInstance(
 
2021
            next(iter(ret["key1"]["key2"].keys())),
 
2022
            ObjectPath)
 
2023
        self.assertEqual("value", ret["key1"]["key2"]["key3"])
 
2024
        self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
 
2025
                                 self.dummy_dbussy.DBUS.ObjectPath)
 
2026
 
 
2027
    def test_call_method_handles_exception(self):
 
2028
        def func():
 
2029
            raise self.dummy_dbussy.DBusError()
 
2030
 
 
2031
        fake_ravel = self.fake_ravel_func(func)
 
2032
        bus = dbussy_adapter.SystemBus(self.dummy_dbussy, fake_ravel)
 
2033
 
 
2034
        with self.assertRaises(dbus.Error) as e:
 
2035
            self.call_method(bus, "methodname", "busname",
 
2036
                             "objectpath", "interface")
 
2037
 
 
2038
        self.assertNotIsInstance(e.exception, dbus.ConnectFailed)
 
2039
 
 
2040
    def test_get_object_converts_to_correct_exception(self):
 
2041
        class fake_ravel_raises_exception_on_connect:
 
2042
            @staticmethod
 
2043
            def system_bus():
 
2044
                class Bus:
 
2045
                    @staticmethod
 
2046
                    def __getitem__(key):
 
2047
                        if key == "objectpath":
 
2048
                            raise self.dummy_dbussy.DBusError()
 
2049
                        raise Exception(key)
 
2050
                return {"busname": Bus()}
 
2051
        def func():
 
2052
            raise self.dummy_dbussy.DBusError()
 
2053
        bus = dbussy_adapter.SystemBus(
 
2054
            self.dummy_dbussy,
 
2055
            fake_ravel_raises_exception_on_connect)
 
2056
        with self.assertRaises(dbus.ConnectFailed):
 
2057
            self.call_method(bus, "methodname", "busname",
 
2058
                             "objectpath", "interface")
 
2059
 
 
2060
 
1777
2061
class Test_commands_from_options(unittest.TestCase):
1778
2062
 
1779
2063
    def setUp(self):