/mandos/release

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

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-03-15 21:59:44 UTC
  • mto: This revision was merged to the branch mainline in revision 382.
  • Revision ID: teddy@recompile.se-20190315215944-lomtn36y7o3jphp8
mandos-ctl: Refactor

* mandos-ctl (main): Call get_mandos_dbus_object() to get
                     "mandos_dbus_objc", and rename it to
                     "mandos_dbus_object".
  (get_mandos_dbus_object): New.
  (Test_get_mandos_dbus_object): - '' -

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
93
93
    if options.debug:
94
94
        log.setLevel(logging.DEBUG)
95
95
 
96
 
    try:
97
 
        bus = dbus.SystemBus()
98
 
        log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
99
 
                  dbus_busname, server_dbus_path)
100
 
        mandos_dbus_objc = bus.get_object(dbus_busname,
101
 
                                          server_dbus_path)
102
 
    except dbus.exceptions.DBusException:
103
 
        log.critical("Could not connect to Mandos server")
104
 
        sys.exit(1)
105
 
 
106
 
    mandos_serv = dbus.Interface(mandos_dbus_objc,
107
 
                                 dbus_interface=server_dbus_interface)
 
96
    bus = dbus.SystemBus()
 
97
 
 
98
    mandos_dbus_object = get_mandos_dbus_object(bus)
 
99
 
 
100
    mandos_serv = dbus.Interface(
 
101
        mandos_dbus_object, dbus_interface=server_dbus_interface)
108
102
    mandos_serv_object_manager = dbus.Interface(
109
 
        mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
 
103
        mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
110
104
 
111
105
    try:
112
106
        log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
436
430
        options.remove = True
437
431
 
438
432
 
 
433
def get_mandos_dbus_object(bus):
 
434
    try:
 
435
        log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
 
436
                  dbus_busname, server_dbus_path)
 
437
        mandos_dbus_object = bus.get_object(dbus_busname,
 
438
                                            server_dbus_path)
 
439
    except dbus.exceptions.DBusException:
 
440
        log.critical("Could not connect to Mandos server")
 
441
        sys.exit(1)
 
442
 
 
443
    return mandos_dbus_object
 
444
 
 
445
 
439
446
class SilenceLogger(object):
440
447
    "Simple context manager to silence a particular logger"
441
448
    def __init__(self, loggername):
988
995
                self.check_option_syntax(options)
989
996
 
990
997
 
 
998
class Test_get_mandos_dbus_object(unittest.TestCase):
 
999
    def test_calls_and_returns_get_object_on_bus(self):
 
1000
        class MockBus(object):
 
1001
            called = False
 
1002
            def get_object(mockbus_self, busname, dbus_path):
 
1003
                # Note that "self" is still the testcase instance,
 
1004
                # this MockBus instance is in "mockbus_self".
 
1005
                self.assertEqual(busname, dbus_busname)
 
1006
                self.assertEqual(dbus_path, server_dbus_path)
 
1007
                mockbus_self.called = True
 
1008
                return mockbus_self
 
1009
 
 
1010
        mockbus = get_mandos_dbus_object(bus=MockBus())
 
1011
        self.assertIsInstance(mockbus, MockBus)
 
1012
        self.assertTrue(mockbus.called)
 
1013
 
 
1014
    def test_logs_and_exits_on_dbus_error(self):
 
1015
        class MockBusFailing(object):
 
1016
            def get_object(self, busname, dbus_path):
 
1017
                raise dbus.exceptions.DBusException("Test")
 
1018
 
 
1019
        # assertLogs only exists in Python 3.4
 
1020
        if hasattr(self, "assertLogs"):
 
1021
            with self.assertLogs(log, logging.CRITICAL):
 
1022
                with self.assertRaises(SystemExit) as e:
 
1023
                    bus = get_mandos_dbus_object(bus=MockBus())
 
1024
        else:
 
1025
            critical_filter = self.CriticalFilter()
 
1026
            log.addFilter(critical_filter)
 
1027
            try:
 
1028
                with self.assertRaises(SystemExit) as e:
 
1029
                    get_mandos_dbus_object(bus=MockBusFailing())
 
1030
            finally:
 
1031
                log.removeFilter(critical_filter)
 
1032
            self.assertTrue(critical_filter.found)
 
1033
        if isinstance(e.exception.code, int):
 
1034
            self.assertNotEqual(e.exception.code, 0)
 
1035
        else:
 
1036
            self.assertIsNotNone(e.exception.code)
 
1037
 
 
1038
    class CriticalFilter(logging.Filter):
 
1039
        """Don't show, but register, critical messages"""
 
1040
        found = False
 
1041
        def filter(self, record):
 
1042
            is_critical = record.levelno >= logging.CRITICAL
 
1043
            self.found = is_critical or self.found
 
1044
            return not is_critical
 
1045
 
 
1046
 
991
1047
class Test_SilenceLogger(unittest.TestCase):
992
1048
    loggername = "mandos-ctl.Test_SilenceLogger"
993
1049
    log = logging.getLogger(loggername)