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
- browse files at revision 1080
- compare with another revision
- download diff
- download tarball
- view history from revision 1080
- Committer: Teddy Hogeborn
- Date: 2019-03-17 16:48:56 UTC
- Revision ID: teddy@recompile.se-20190317164856-de11glvsfwhnj34w
mandos-ctl: Refactor tests
* mandos-ctl (Test_string_to_delta.test_handles_basic_rfc3339): Split
into multiple tests.
(Test_string_to_delta.test_rfc3339_zero_seconds): New.
(Test_string_to_delta.test_rfc3339_zero_days): - '' -
(Test_string_to_delta.test_rfc3339_one_second): - '' -
(Test_string_to_delta.test_rfc3339_two_hours): - '' -
(TestCommand.setUp.MockClient.Get): Remove; unused.
(TestBaseCommands.test_Remove): Remove unnecessary assertion.
(TestPropertyCmd.runTest): - '' -
* mandos-ctl (Test_string_to_delta.test_handles_basic_rfc3339): Split
into multiple tests.
(Test_string_to_delta.test_rfc3339_zero_seconds): New.
(Test_string_to_delta.test_rfc3339_zero_days): - '' -
(Test_string_to_delta.test_rfc3339_one_second): - '' -
(Test_string_to_delta.test_rfc3339_two_hours): - '' -
(TestCommand.setUp.MockClient.Get): Remove; unused.
(TestBaseCommands.test_Remove): Remove unnecessary assertion.
(TestPropertyCmd.runTest): - '' -
- files modified:
- intro.xml
added
removed
mandos-ctl
46
46
import tempfile
47
47
import contextlib
48
48
49
try:
50
import pydbus
51
import gi
52
dbus_python = None
53
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
49
import dbus
62
50
63
51
# Show warnings by default
64
52
if not sys.warnoptions:
73
61
74
62
if sys.version_info.major == 2:
75
63
str = unicode
76
import StringIO
77
io.StringIO = StringIO.StringIO
78
64
79
65
locale.setlocale(locale.LC_ALL, "")
80
66
67
dbus_busname_domain = "se.recompile"
68
dbus_busname = dbus_busname_domain + ".Mandos"
69
server_dbus_path = "/"
70
server_dbus_interface = dbus_busname_domain + ".Mandos"
71
client_dbus_interface = dbus_busname_domain + ".Mandos.Client"
72
del dbus_busname_domain
81
73
version = "1.8.3"
82
74
83
75
76
try:
77
dbus.OBJECT_MANAGER_IFACE
78
except AttributeError:
79
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
80
81
84
82
def main():
85
83
parser = argparse.ArgumentParser()
84
86
85
add_command_line_options(parser)
87
86
88
87
options = parser.parse_args()
88
89
89
check_option_syntax(parser, options)
90
90
91
91
clientnames = options.client
93
93
if options.debug:
94
94
log.setLevel(logging.DEBUG)
95
95
96
if pydbus is not None:
97
bus = pydbus_adapter.CachingBus(pydbus)
98
else:
99
bus = dbus_python_adapter.CachingBus(dbus_python)
100
101
try:
102
all_clients = bus.get_clients_and_properties()
103
except dbus.ConnectFailed as e:
104
log.critical("Could not connect to Mandos server: %s", e)
105
sys.exit(1)
106
except dbus.Error as e:
107
log.critical(
108
"Failed to access Mandos server through D-Bus:\n%s", e)
109
sys.exit(1)
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)
102
mandos_serv_object_manager = dbus.Interface(
103
mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
104
105
managed_objects = get_managed_objects(mandos_serv_object_manager)
106
107
all_clients = {}
108
for path, ifs_and_props in managed_objects.items():
109
try:
110
all_clients[path] = ifs_and_props[client_dbus_interface]
111
except KeyError:
112
pass
110
113
111
114
# Compile dict of (clientpath: properties) to process
112
115
if not clientnames:
125
128
commands = commands_from_options(options)
126
129
127
130
for command in commands:
128
command.run(clients, bus)
131
command.run(clients, bus, mandos_serv)
129
132
130
133
131
134
def add_command_line_options(parser):
136
139
help="Select all clients")
137
140
parser.add_argument("-v", "--verbose", action="store_true",
138
141
help="Print all fields")
139
parser.add_argument("-j", "--dump-json", dest="commands",
140
action="append_const", default=[],
141
const=command.DumpJSON(),
142
parser.add_argument("-j", "--dump-json", action="store_true",
142
143
help="Dump client data in JSON format")
143
144
enable_disable = parser.add_mutually_exclusive_group()
144
enable_disable.add_argument("-e", "--enable", dest="commands",
145
action="append_const", default=[],
146
const=command.Enable(),
145
enable_disable.add_argument("-e", "--enable", action="store_true",
147
146
help="Enable client")
148
enable_disable.add_argument("-d", "--disable", dest="commands",
149
action="append_const", default=[],
150
const=command.Disable(),
147
enable_disable.add_argument("-d", "--disable",
148
action="store_true",
151
149
help="disable client")
152
parser.add_argument("-b", "--bump-timeout", dest="commands",
153
action="append_const", default=[],
154
const=command.BumpTimeout(),
150
parser.add_argument("-b", "--bump-timeout", action="store_true",
155
151
help="Bump timeout for client")
156
152
start_stop_checker = parser.add_mutually_exclusive_group()
157
153
start_stop_checker.add_argument("--start-checker",
158
dest="commands",
159
action="append_const", default=[],
160
const=command.StartChecker(),
154
action="store_true",
161
155
help="Start checker for client")
162
start_stop_checker.add_argument("--stop-checker", dest="commands",
163
action="append_const", default=[],
164
const=command.StopChecker(),
156
start_stop_checker.add_argument("--stop-checker",
157
action="store_true",
165
158
help="Stop checker for client")
166
parser.add_argument("-V", "--is-enabled", dest="commands",
167
action="append_const", default=[],
168
const=command.IsEnabled(),
159
parser.add_argument("-V", "--is-enabled", action="store_true",
169
160
help="Check if client is enabled")
170
parser.add_argument("-r", "--remove", dest="commands",
171
action="append_const", default=[],
172
const=command.Remove(),
161
parser.add_argument("-r", "--remove", action="store_true",
173
162
help="Remove client")
174
parser.add_argument("-c", "--checker", dest="commands",
175
action="append", default=[],
176
metavar="COMMAND", type=command.SetChecker,
163
parser.add_argument("-c", "--checker",
177
164
help="Set checker command for client")
178
parser.add_argument(
179
"-t", "--timeout", dest="commands", action="append",
180
default=[], metavar="TIME",
181
type=command.SetTimeout.argparse(string_to_delta),
182
help="Set timeout for client")
183
parser.add_argument(
184
"--extended-timeout", dest="commands", action="append",
185
default=[], metavar="TIME",
186
type=command.SetExtendedTimeout.argparse(string_to_delta),
187
help="Set extended timeout for client")
188
parser.add_argument(
189
"-i", "--interval", dest="commands", action="append",
190
default=[], metavar="TIME",
191
type=command.SetInterval.argparse(string_to_delta),
192
help="Set checker interval for client")
165
parser.add_argument("-t", "--timeout", type=string_to_delta,
166
help="Set timeout for client")
167
parser.add_argument("--extended-timeout", type=string_to_delta,
168
help="Set extended timeout for client")
169
parser.add_argument("-i", "--interval", type=string_to_delta,
170
help="Set checker interval for client")
193
171
approve_deny_default = parser.add_mutually_exclusive_group()
194
172
approve_deny_default.add_argument(
195
"--approve-by-default", dest="commands",
196
action="append_const", default=[],
197
const=command.ApproveByDefault(),
173
"--approve-by-default", action="store_true",
174
default=None, dest="approved_by_default",
198
175
help="Set client to be approved by default")
199
176
approve_deny_default.add_argument(
200
"--deny-by-default", dest="commands",
201
action="append_const", default=[],
202
const=command.DenyByDefault(),
177
"--deny-by-default", action="store_false",
178
dest="approved_by_default",
203
179
help="Set client to be denied by default")
204
parser.add_argument(
205
"--approval-delay", dest="commands", action="append",
206
default=[], metavar="TIME",
207
type=command.SetApprovalDelay.argparse(string_to_delta),
208
help="Set delay before client approve/deny")
209
parser.add_argument(
210
"--approval-duration", dest="commands", action="append",
211
default=[], metavar="TIME",
212
type=command.SetApprovalDuration.argparse(string_to_delta),
213
help="Set duration of one client approval")
214
parser.add_argument("-H", "--host", dest="commands",
215
action="append", default=[], metavar="STRING",
216
type=command.SetHost,
217
help="Set host for client")
218
parser.add_argument(
219
"-s", "--secret", dest="commands", action="append",
220
default=[], metavar="FILENAME",
221
type=command.SetSecret.argparse(argparse.FileType(mode="rb")),
222
help="Set password blob (file) for client")
180
parser.add_argument("--approval-delay", type=string_to_delta,
181
help="Set delay before client approve/deny")
182
parser.add_argument("--approval-duration", type=string_to_delta,
183
help="Set duration of one client approval")
184
parser.add_argument("-H", "--host", help="Set host for client")
185
parser.add_argument("-s", "--secret",
186
type=argparse.FileType(mode="rb"),
187
help="Set password blob (file) for client")
223
188
approve_deny = parser.add_mutually_exclusive_group()
224
189
approve_deny.add_argument(
225
"-A", "--approve", dest="commands", action="append_const",
226
default=[], const=command.Approve(),
190
"-A", "--approve", action="store_true",
227
191
help="Approve any current client request")
228
approve_deny.add_argument("-D", "--deny", dest="commands",
229
action="append_const", default=[],
230
const=command.Deny(),
192
approve_deny.add_argument("-D", "--deny", action="store_true",
231
193
help="Deny any current client request")
232
194
parser.add_argument("--debug", action="store_true",
233
195
help="Debug mode (show D-Bus commands)")
424
386
"""Apply additional restrictions on options, not expressible in
425
387
argparse"""
426
388
427
def has_commands(options, commands=None):
428
if commands is None:
429
commands = (command.Enable,
430
command.Disable,
431
command.BumpTimeout,
432
command.StartChecker,
433
command.StopChecker,
434
command.IsEnabled,
435
command.Remove,
436
command.SetChecker,
437
command.SetTimeout,
438
command.SetExtendedTimeout,
439
command.SetInterval,
440
command.ApproveByDefault,
441
command.DenyByDefault,
442
command.SetApprovalDelay,
443
command.SetApprovalDuration,
444
command.SetHost,
445
command.SetSecret,
446
command.Approve,
447
command.Deny)
448
return any(isinstance(cmd, commands)
449
for cmd in options.commands)
389
def has_actions(options):
390
return any((options.enable,
391
options.disable,
392
options.bump_timeout,
393
options.start_checker,
394
options.stop_checker,
395
options.is_enabled,
396
options.remove,
397
options.checker is not None,
398
options.timeout is not None,
399
options.extended_timeout is not None,
400
options.interval is not None,
401
options.approved_by_default is not None,
402
options.approval_delay is not None,
403
options.approval_duration is not None,
404
options.host is not None,
405
options.secret is not None,
406
options.approve,
407
options.deny))
450
408
451
if has_commands(options) and not (options.client or options.all):
409
if has_actions(options) and not (options.client or options.all):
452
410
parser.error("Options require clients names or --all.")
453
if options.verbose and has_commands(options):
411
if options.verbose and has_actions(options):
454
412
parser.error("--verbose can only be used alone.")
455
if (has_commands(options, (command.DumpJSON,))
456
and (options.verbose or len(options.commands) > 1)):
413
if options.dump_json and (options.verbose
414
or has_actions(options)):
457
415
parser.error("--dump-json can only be used alone.")
458
if options.all and not has_commands(options):
416
if options.all and not has_actions(options):
459
417
parser.error("--all requires an action.")
460
if (has_commands(options, (command.IsEnabled,))
461
and len(options.client) > 1):
418
if options.is_enabled and len(options.client) > 1:
462
419
parser.error("--is-enabled requires exactly one client")
463
if (len(options.commands) > 1
464
and has_commands(options, (command.Remove,))
465
and not has_commands(options, (command.Deny,))):
466
parser.error("--remove can only be combined with --deny")
467
468
469
class dbus(object):
470
471
class SystemBus(object):
472
473
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
474
def get_managed_objects(self, busname, objectpath):
475
return self.call_method("GetManagedObjects", busname,
476
objectpath,
477
self.object_manager_iface)
478
479
properties_iface = "org.freedesktop.DBus.Properties"
480
def set_property(self, busname, objectpath, interface, key,
481
value):
482
self.call_method("Set", busname, objectpath,
483
self.properties_iface, interface, key,
484
value)
485
486
487
class MandosBus(SystemBus):
488
busname_domain = "se.recompile"
489
busname = busname_domain + ".Mandos"
490
server_path = "/"
491
server_interface = busname_domain + ".Mandos"
492
client_interface = busname_domain + ".Mandos.Client"
493
del busname_domain
494
495
def get_clients_and_properties(self):
496
managed_objects = self.get_managed_objects(
497
self.busname, self.server_path)
498
return {objpath: properties[self.client_interface]
499
for objpath, properties in managed_objects.items()
500
if self.client_interface in properties}
501
502
def set_client_property(self, objectpath, key, value):
503
return self.set_property(self.busname, objectpath,
504
self.client_interface, key,
505
value)
506
507
def call_client_method(self, objectpath, method, *args):
508
return self.call_method(method, self.busname, objectpath,
509
self.client_interface, *args)
510
511
def call_server_method(self, method, *args):
512
return self.call_method(method, self.busname,
513
self.server_path,
514
self.server_interface, *args)
515
516
class Error(Exception):
517
pass
518
519
class ConnectFailed(Error):
520
pass
521
522
523
class dbus_python_adapter(object):
524
525
class SystemBus(dbus.MandosBus):
526
"""Use dbus-python"""
527
528
def __init__(self, module=dbus_python):
529
self.dbus_python = module
530
self.bus = self.dbus_python.SystemBus()
531
532
@contextlib.contextmanager
533
def convert_exception(self, exception_class=dbus.Error):
534
try:
535
yield
536
except self.dbus_python.exceptions.DBusException as e:
537
# This does what "raise from" would do
538
exc = exception_class(*e.args)
539
exc.__cause__ = e
540
raise exc
541
542
def call_method(self, methodname, busname, objectpath,
543
interface, *args):
544
proxy_object = self.get_object(busname, objectpath)
545
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
546
interface, methodname,
547
", ".join(repr(a) for a in args))
548
method = getattr(proxy_object, methodname)
549
with self.convert_exception():
550
with dbus_python_adapter.SilenceLogger(
551
"dbus.proxies"):
552
value = method(*args, dbus_interface=interface)
553
return self.type_filter(value)
554
555
def get_object(self, busname, objectpath):
556
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
557
busname, objectpath)
558
with self.convert_exception(dbus.ConnectFailed):
559
return self.bus.get_object(busname, objectpath)
560
561
def type_filter(self, value):
562
"""Convert the most bothersome types to Python types"""
563
if isinstance(value, self.dbus_python.Boolean):
564
return bool(value)
565
if isinstance(value, self.dbus_python.ObjectPath):
566
return str(value)
567
# Also recurse into dictionaries
568
if isinstance(value, self.dbus_python.Dictionary):
569
return {self.type_filter(key):
570
self.type_filter(subval)
571
for key, subval in value.items()}
572
return value
573
574
575
class SilenceLogger(object):
576
"Simple context manager to silence a particular logger"
577
def __init__(self, loggername):
578
self.logger = logging.getLogger(loggername)
579
580
def __enter__(self):
581
self.logger.addFilter(self.nullfilter)
582
583
class NullFilter(logging.Filter):
584
def filter(self, record):
585
return False
586
587
nullfilter = NullFilter()
588
589
def __exit__(self, exc_type, exc_val, exc_tb):
590
self.logger.removeFilter(self.nullfilter)
591
592
593
class CachingBus(SystemBus):
594
"""A caching layer for dbus_python_adapter.SystemBus"""
595
def __init__(self, *args, **kwargs):
596
self.object_cache = {}
597
super(dbus_python_adapter.CachingBus,
598
self).__init__(*args, **kwargs)
599
def get_object(self, busname, objectpath):
600
try:
601
return self.object_cache[(busname, objectpath)]
602
except KeyError:
603
new_object = super(
604
dbus_python_adapter.CachingBus,
605
self).get_object(busname, objectpath)
606
self.object_cache[(busname, objectpath)] = new_object
607
return new_object
608
609
610
class pydbus_adapter(object):
611
class SystemBus(dbus.MandosBus):
612
def __init__(self, module=pydbus):
613
self.pydbus = module
614
self.bus = self.pydbus.SystemBus()
615
616
@contextlib.contextmanager
617
def convert_exception(self, exception_class=dbus.Error):
618
try:
619
yield
620
except gi.repository.GLib.Error as e:
621
# This does what "raise from" would do
622
exc = exception_class(*e.args)
623
exc.__cause__ = e
624
raise exc
625
626
def call_method(self, methodname, busname, objectpath,
627
interface, *args):
628
proxy_object = self.get(busname, objectpath)
629
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
630
interface, methodname,
631
", ".join(repr(a) for a in args))
632
method = getattr(proxy_object[interface], methodname)
633
with self.convert_exception():
634
return method(*args)
635
636
def get(self, busname, objectpath):
637
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
638
busname, objectpath)
639
with self.convert_exception(dbus.ConnectFailed):
640
if sys.version_info.major <= 2:
641
with warnings.catch_warnings():
642
warnings.filterwarnings(
643
"ignore", "", DeprecationWarning,
644
r"^xml\.etree\.ElementTree$")
645
return self.bus.get(busname, objectpath)
646
else:
647
return self.bus.get(busname, objectpath)
648
649
def set_property(self, busname, objectpath, interface, key,
650
value):
651
proxy_object = self.get(busname, objectpath)
652
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
653
objectpath, self.properties_iface, interface,
654
key, value)
655
setattr(proxy_object[interface], key, value)
656
657
class CachingBus(SystemBus):
658
"""A caching layer for pydbus_adapter.SystemBus"""
659
def __init__(self, *args, **kwargs):
660
self.object_cache = {}
661
super(pydbus_adapter.CachingBus,
662
self).__init__(*args, **kwargs)
663
def get(self, busname, objectpath):
664
try:
665
return self.object_cache[(busname, objectpath)]
666
except KeyError:
667
new_object = (super(pydbus_adapter.CachingBus, self)
668
.get(busname, objectpath))
669
self.object_cache[(busname, objectpath)] = new_object
670
return new_object
420
if options.remove:
421
options.remove = False
422
if has_actions(options) and not options.deny:
423
parser.error("--remove can only be combined with --deny")
424
options.remove = True
425
426
427
def get_mandos_dbus_object(bus):
428
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
429
dbus_busname, server_dbus_path)
430
with if_dbus_exception_log_with_exception_and_exit(
431
"Could not connect to Mandos server: %s"):
432
mandos_dbus_object = bus.get_object(dbus_busname,
433
server_dbus_path)
434
return mandos_dbus_object
435
436
437
@contextlib.contextmanager
438
def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):
439
try:
440
yield
441
except dbus.exceptions.DBusException as e:
442
log.critical(*(args + (e,)), **kwargs)
443
sys.exit(1)
444
445
446
def get_managed_objects(object_manager):
447
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
448
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
449
with if_dbus_exception_log_with_exception_and_exit(
450
"Failed to access Mandos server through D-Bus:\n%s"):
451
with SilenceLogger("dbus.proxies"):
452
managed_objects = object_manager.GetManagedObjects()
453
return managed_objects
454
455
456
class SilenceLogger(object):
457
"Simple context manager to silence a particular logger"
458
def __init__(self, loggername):
459
self.logger = logging.getLogger(loggername)
460
461
def __enter__(self):
462
self.logger.addFilter(self.nullfilter)
463
return self
464
465
class NullFilter(logging.Filter):
466
def filter(self, record):
467
return False
468
469
nullfilter = NullFilter()
470
471
def __exit__(self, exc_type, exc_val, exc_tb):
472
self.logger.removeFilter(self.nullfilter)
671
473
672
474
673
475
def commands_from_options(options):
674
476
675
commands = list(options.commands)
676
677
def find_cmd(cmd, commands):
678
i = 0
679
for i, c in enumerate(commands):
680
if isinstance(c, cmd):
681
return i
682
return i+1
683
684
# If command.Remove is present, move any instances of command.Deny
685
# to occur ahead of command.Remove.
686
index_of_remove = find_cmd(command.Remove, commands)
687
before_remove = commands[:index_of_remove]
688
after_remove = commands[index_of_remove:]
689
cleaned_after = []
690
for cmd in after_remove:
691
if isinstance(cmd, command.Deny):
692
before_remove.append(cmd)
477
commands = []
478
479
if options.is_enabled:
480
commands.append(command.IsEnabled())
481
482
if options.approve:
483
commands.append(command.Approve())
484
485
if options.deny:
486
commands.append(command.Deny())
487
488
if options.remove:
489
commands.append(command.Remove())
490
491
if options.dump_json:
492
commands.append(command.DumpJSON())
493
494
if options.enable:
495
commands.append(command.Enable())
496
497
if options.disable:
498
commands.append(command.Disable())
499
500
if options.bump_timeout:
501
commands.append(command.BumpTimeout())
502
503
if options.start_checker:
504
commands.append(command.StartChecker())
505
506
if options.stop_checker:
507
commands.append(command.StopChecker())
508
509
if options.approved_by_default is not None:
510
if options.approved_by_default:
511
commands.append(command.ApproveByDefault())
693
512
else:
694
cleaned_after.append(cmd)
695
if cleaned_after != after_remove:
696
commands = before_remove + cleaned_after
513
commands.append(command.DenyByDefault())
514
515
if options.checker is not None:
516
commands.append(command.SetChecker(options.checker))
517
518
if options.host is not None:
519
commands.append(command.SetHost(options.host))
520
521
if options.secret is not None:
522
commands.append(command.SetSecret(options.secret))
523
524
if options.timeout is not None:
525
commands.append(command.SetTimeout(options.timeout))
526
527
if options.extended_timeout:
528
commands.append(
529
command.SetExtendedTimeout(options.extended_timeout))
530
531
if options.interval is not None:
532
commands.append(command.SetInterval(options.interval))
533
534
if options.approval_delay is not None:
535
commands.append(
536
command.SetApprovalDelay(options.approval_delay))
537
538
if options.approval_duration is not None:
539
commands.append(
540
command.SetApprovalDuration(options.approval_duration))
697
541
698
542
# If no command option has been given, show table of clients,
699
543
# optionally verbosely
708
552
709
553
class Base(object):
710
554
"""Abstract base class for commands"""
711
def run(self, clients, bus=None):
555
def run(self, clients, bus=None, mandos=None):
712
556
"""Normal commands should implement run_on_one_client(),
713
557
but commands which want to operate on all clients at the same time can
714
558
override this run() method instead.
715
559
"""
716
self.bus = bus
717
for client, properties in clients.items():
560
self.mandos = mandos
561
for clientpath, properties in clients.items():
562
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
563
dbus_busname, str(clientpath))
564
client = bus.get_object(dbus_busname, clientpath)
718
565
self.run_on_one_client(client, properties)
719
566
720
567
721
568
class IsEnabled(Base):
722
def run(self, clients, bus=None):
723
properties = next(iter(clients.values()))
724
if properties["Enabled"]:
569
def run(self, clients, bus=None, mandos=None):
570
client, properties = next(iter(clients.items()))
571
if self.is_enabled(client, properties):
725
572
sys.exit(0)
726
573
sys.exit(1)
574
def is_enabled(self, client, properties):
575
return properties["Enabled"]
727
576
728
577
729
578
class Approve(Base):
730
579
def run_on_one_client(self, client, properties):
731
self.bus.call_client_method(client, "Approve", True)
580
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
581
client.__dbus_object_path__,
582
client_dbus_interface)
583
client.Approve(dbus.Boolean(True),
584
dbus_interface=client_dbus_interface)
732
585
733
586
734
587
class Deny(Base):
735
588
def run_on_one_client(self, client, properties):
736
self.bus.call_client_method(client, "Approve", False)
589
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
590
client.__dbus_object_path__,
591
client_dbus_interface)
592
client.Approve(dbus.Boolean(False),
593
dbus_interface=client_dbus_interface)
737
594
738
595
739
596
class Remove(Base):
740
def run(self, clients, bus):
741
for clientpath in frozenset(clients.keys()):
742
bus.call_server_method("RemoveClient", clientpath)
597
def run_on_one_client(self, client, properties):
598
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)",
599
dbus_busname, server_dbus_path,
600
server_dbus_interface,
601
str(client.__dbus_object_path__))
602
self.mandos.RemoveClient(client.__dbus_object_path__)
743
603
744
604
745
605
class Output(Base):
753
613
"Checker", "ExtendedTimeout", "Expires",
754
614
"LastCheckerStatus")
755
615
616
def run(self, clients, bus=None, mandos=None):
617
print(self.output(clients.values()))
618
619
def output(self, clients):
620
raise NotImplementedError()
621
756
622
757
623
class DumpJSON(Output):
758
def run(self, clients, bus=None):
759
data = {properties["Name"]:
760
{key: properties[key]
624
def output(self, clients):
625
data = {client["Name"]:
626
{key: self.dbus_boolean_to_bool(client[key])
761
627
for key in self.all_keywords}
762
for properties in clients.values()}
763
print(json.dumps(data, indent=4, separators=(',', ': ')))
628
for client in clients}
629
return json.dumps(data, indent=4, separators=(',', ': '))
630
631
@staticmethod
632
def dbus_boolean_to_bool(value):
633
if isinstance(value, dbus.Boolean):
634
value = bool(value)
635
return value
764
636
765
637
766
638
class PrintTable(Output):
767
639
def __init__(self, verbose=False):
768
640
self.verbose = verbose
769
641
770
def run(self, clients, bus=None):
642
def output(self, clients):
771
643
default_keywords = ("Name", "Enabled", "Timeout",
772
644
"LastCheckedOK")
773
645
keywords = default_keywords
774
646
if self.verbose:
775
647
keywords = self.all_keywords
776
print(self.TableOfClients(clients.values(), keywords))
648
return str(self.TableOfClients(clients, keywords))
777
649
778
650
class TableOfClients(object):
779
651
tableheaders = {
834
706
835
707
@classmethod
836
708
def valuetostring(cls, value, keyword):
837
if isinstance(value, bool):
709
if isinstance(value, dbus.Boolean):
838
710
return "Yes" if value else "No"
839
711
if keyword in ("Timeout", "Interval", "ApprovalDelay",
840
712
"ApprovalDuration", "ExtendedTimeout"):
860
732
seconds=td.seconds % 60))
861
733
862
734
863
class PropertySetter(Base):
735
class Property(Base):
864
736
"Abstract class for Actions for setting one client property"
865
737
866
def run_on_one_client(self, client, properties=None):
738
def run_on_one_client(self, client, properties):
867
739
"""Set the Client's D-Bus property"""
868
self.bus.set_client_property(client, self.propname,
869
self.value_to_set)
740
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
741
client.__dbus_object_path__,
742
dbus.PROPERTIES_IFACE, client_dbus_interface,
743
self.propname, self.value_to_set
744
if not isinstance(self.value_to_set,
745
dbus.Boolean)
746
else bool(self.value_to_set))
747
client.Set(client_dbus_interface, self.propname,
748
self.value_to_set,
749
dbus_interface=dbus.PROPERTIES_IFACE)
870
750
871
751
@property
872
752
def propname(self):
873
753
raise NotImplementedError()
874
754
875
755
876
class Enable(PropertySetter):
877
propname = "Enabled"
878
value_to_set = True
879
880
881
class Disable(PropertySetter):
882
propname = "Enabled"
883
value_to_set = False
884
885
886
class BumpTimeout(PropertySetter):
756
class Enable(Property):
757
propname = "Enabled"
758
value_to_set = dbus.Boolean(True)
759
760
761
class Disable(Property):
762
propname = "Enabled"
763
value_to_set = dbus.Boolean(False)
764
765
766
class BumpTimeout(Property):
887
767
propname = "LastCheckedOK"
888
768
value_to_set = ""
889
769
890
770
891
class StartChecker(PropertySetter):
892
propname = "CheckerRunning"
893
value_to_set = True
894
895
896
class StopChecker(PropertySetter):
897
propname = "CheckerRunning"
898
value_to_set = False
899
900
901
class ApproveByDefault(PropertySetter):
902
propname = "ApprovedByDefault"
903
value_to_set = True
904
905
906
class DenyByDefault(PropertySetter):
907
propname = "ApprovedByDefault"
908
value_to_set = False
909
910
911
class PropertySetterValue(PropertySetter):
912
"""Abstract class for PropertySetter recieving a value as
913
constructor argument instead of a class attribute."""
771
class StartChecker(Property):
772
propname = "CheckerRunning"
773
value_to_set = dbus.Boolean(True)
774
775
776
class StopChecker(Property):
777
propname = "CheckerRunning"
778
value_to_set = dbus.Boolean(False)
779
780
781
class ApproveByDefault(Property):
782
propname = "ApprovedByDefault"
783
value_to_set = dbus.Boolean(True)
784
785
786
class DenyByDefault(Property):
787
propname = "ApprovedByDefault"
788
value_to_set = dbus.Boolean(False)
789
790
791
class PropertyValue(Property):
792
"Abstract class for Property recieving a value as argument"
914
793
def __init__(self, value):
915
794
self.value_to_set = value
916
795
917
@classmethod
918
def argparse(cls, argtype):
919
def cmdtype(arg):
920
return cls(argtype(arg))
921
return cmdtype
922
796
923
class SetChecker(PropertySetterValue):
797
class SetChecker(PropertyValue):
924
798
propname = "Checker"
925
799
926
800
927
class SetHost(PropertySetterValue):
801
class SetHost(PropertyValue):
928
802
propname = "Host"
929
803
930
804
931
class SetSecret(PropertySetterValue):
805
class SetSecret(PropertyValue):
932
806
propname = "Secret"
933
807
934
808
@property
942
816
value.close()
943
817
944
818
945
class PropertySetterValueMilliseconds(PropertySetterValue):
946
"""Abstract class for PropertySetterValue taking a value
947
argument as a datetime.timedelta() but should store it as
948
milliseconds."""
819
class MillisecondsPropertyValueArgument(PropertyValue):
820
"""Abstract class for PropertyValue taking a value argument as
821
a datetime.timedelta() but should store it as milliseconds."""
949
822
950
823
@property
951
824
def value_to_set(self):
957
830
self._vts = int(round(value.total_seconds() * 1000))
958
831
959
832
960
class SetTimeout(PropertySetterValueMilliseconds):
833
class SetTimeout(MillisecondsPropertyValueArgument):
961
834
propname = "Timeout"
962
835
963
836
964
class SetExtendedTimeout(PropertySetterValueMilliseconds):
837
class SetExtendedTimeout(MillisecondsPropertyValueArgument):
965
838
propname = "ExtendedTimeout"
966
839
967
840
968
class SetInterval(PropertySetterValueMilliseconds):
841
class SetInterval(MillisecondsPropertyValueArgument):
969
842
propname = "Interval"
970
843
971
844
972
class SetApprovalDelay(PropertySetterValueMilliseconds):
845
class SetApprovalDelay(MillisecondsPropertyValueArgument):
973
846
propname = "ApprovalDelay"
974
847
975
848
976
class SetApprovalDuration(PropertySetterValueMilliseconds):
849
class SetApprovalDuration(MillisecondsPropertyValueArgument):
977
850
propname = "ApprovalDuration"
978
851
979
852
1012
885
"output"))
1013
886
1014
887
1015
class Unique(object):
1016
"""Class for objects which exist only to be unique objects, since
1017
unittest.mock.sentinel only exists in Python 3.3"""
1018
1019
1020
888
class Test_string_to_delta(TestCaseWithAssertLogs):
1021
889
# Just test basic RFC 3339 functionality here, the doc string for
1022
890
# rfc3339_duration_to_delta() already has more comprehensive
1023
# tests, which are run by doctest.
891
# tests, which is run by doctest.
1024
892
1025
893
def test_rfc3339_zero_seconds(self):
1026
self.assertEqual(datetime.timedelta(),
1027
string_to_delta("PT0S"))
894
self.assertEqual(string_to_delta("PT0S"),
895
datetime.timedelta())
1028
896
1029
897
def test_rfc3339_zero_days(self):
1030
self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
898
self.assertEqual(string_to_delta("P0D"),
899
datetime.timedelta())
1031
900
1032
901
def test_rfc3339_one_second(self):
1033
self.assertEqual(datetime.timedelta(0, 1),
1034
string_to_delta("PT1S"))
902
self.assertEqual(string_to_delta("PT1S"),
903
datetime.timedelta(0, 1))
1035
904
1036
905
def test_rfc3339_two_hours(self):
1037
self.assertEqual(datetime.timedelta(0, 7200),
1038
string_to_delta("PT2H"))
906
self.assertEqual(string_to_delta("PT2H"),
907
datetime.timedelta(0, 7200))
1039
908
1040
909
def test_falls_back_to_pre_1_6_1_with_warning(self):
1041
910
with self.assertLogs(log, logging.WARNING):
1042
911
value = string_to_delta("2h")
1043
self.assertEqual(datetime.timedelta(0, 7200), value)
912
self.assertEqual(value, datetime.timedelta(0, 7200))
1044
913
1045
914
1046
915
class Test_check_option_syntax(unittest.TestCase):
1050
919
1051
920
def test_actions_requires_client_or_all(self):
1052
921
for action, value in self.actions.items():
1053
args = self.actionargs(action, value)
922
options = self.parser.parse_args()
923
setattr(options, action, value)
1054
924
with self.assertParseError():
1055
self.parse_args(args)
925
self.check_option_syntax(options)
1056
926
1057
# This mostly corresponds to the definition from has_commands() in
927
# This mostly corresponds to the definition from has_actions() in
1058
928
# check_option_syntax()
1059
929
actions = {
1060
"--enable": None,
1061
"--disable": None,
1062
"--bump-timeout": None,
1063
"--start-checker": None,
1064
"--stop-checker": None,
1065
"--is-enabled": None,
1066
"--remove": None,
1067
"--checker": "x",
1068
"--timeout": "PT0S",
1069
"--extended-timeout": "PT0S",
1070
"--interval": "PT0S",
1071
"--approve-by-default": None,
1072
"--deny-by-default": None,
1073
"--approval-delay": "PT0S",
1074
"--approval-duration": "PT0S",
1075
"--host": "hostname",
1076
"--secret": "/dev/null",
1077
"--approve": None,
1078
"--deny": None,
930
# The actual values set here are not that important, but we do
931
# at least stick to the correct types, even though they are
932
# never used
933
"enable": True,
934
"disable": True,
935
"bump_timeout": True,
936
"start_checker": True,
937
"stop_checker": True,
938
"is_enabled": True,
939
"remove": True,
940
"checker": "x",
941
"timeout": datetime.timedelta(),
942
"extended_timeout": datetime.timedelta(),
943
"interval": datetime.timedelta(),
944
"approved_by_default": True,
945
"approval_delay": datetime.timedelta(),
946
"approval_duration": datetime.timedelta(),
947
"host": "x",
948
"secret": io.BytesIO(b"x"),
949
"approve": True,
950
"deny": True,
1079
951
}
1080
952
1081
@staticmethod
1082
def actionargs(action, value, *args):
1083
if value is not None:
1084
return [action, value] + list(args)
1085
else:
1086
return [action] + list(args)
1087
1088
953
@contextlib.contextmanager
1089
954
def assertParseError(self):
1090
955
with self.assertRaises(SystemExit) as e:
1093
958
# Exit code from argparse is guaranteed to be "2". Reference:
1094
959
# https://docs.python.org/3/library
1095
960
# /argparse.html#exiting-methods
1096
self.assertEqual(2, e.exception.code)
1097
1098
def parse_args(self, args):
1099
options = self.parser.parse_args(args)
1100
check_option_syntax(self.parser, options)
961
self.assertEqual(e.exception.code, 2)
1101
962
1102
963
@staticmethod
1103
964
@contextlib.contextmanager
1104
965
def redirect_stderr_to_devnull():
1105
old_stderr = sys.stderr
1106
with contextlib.closing(open(os.devnull, "w")) as null:
1107
sys.stderr = null
1108
try:
1109
yield
1110
finally:
1111
sys.stderr = old_stderr
966
null = os.open(os.path.devnull, os.O_RDWR)
967
stderrcopy = os.dup(sys.stderr.fileno())
968
os.dup2(null, sys.stderr.fileno())
969
os.close(null)
970
try:
971
yield
972
finally:
973
# restore stderr
974
os.dup2(stderrcopy, sys.stderr.fileno())
975
os.close(stderrcopy)
1112
976
1113
977
def check_option_syntax(self, options):
1114
978
check_option_syntax(self.parser, options)
1115
979
1116
980
def test_actions_all_conflicts_with_verbose(self):
1117
981
for action, value in self.actions.items():
1118
args = self.actionargs(action, value, "--all",
1119
"--verbose")
982
options = self.parser.parse_args()
983
setattr(options, action, value)
984
options.all = True
985
options.verbose = True
1120
986
with self.assertParseError():
1121
self.parse_args(args)
987
self.check_option_syntax(options)
1122
988
1123
989
def test_actions_with_client_conflicts_with_verbose(self):
1124
990
for action, value in self.actions.items():
1125
args = self.actionargs(action, value, "--verbose",
1126
"client")
991
options = self.parser.parse_args()
992
setattr(options, action, value)
993
options.verbose = True
994
options.client = ["foo"]
1127
995
with self.assertParseError():
1128
self.parse_args(args)
996
self.check_option_syntax(options)
1129
997
1130
998
def test_dump_json_conflicts_with_verbose(self):
1131
args = ["--dump-json", "--verbose"]
999
options = self.parser.parse_args()
1000
options.dump_json = True
1001
options.verbose = True
1132
1002
with self.assertParseError():
1133
self.parse_args(args)
1003
self.check_option_syntax(options)
1134
1004
1135
1005
def test_dump_json_conflicts_with_action(self):
1136
1006
for action, value in self.actions.items():
1137
args = self.actionargs(action, value, "--dump-json")
1007
options = self.parser.parse_args()
1008
setattr(options, action, value)
1009
options.dump_json = True
1138
1010
with self.assertParseError():
1139
self.parse_args(args)
1011
self.check_option_syntax(options)
1140
1012
1141
1013
def test_all_can_not_be_alone(self):
1142
args = ["--all"]
1014
options = self.parser.parse_args()
1015
options.all = True
1143
1016
with self.assertParseError():
1144
self.parse_args(args)
1017
self.check_option_syntax(options)
1145
1018
1146
1019
def test_all_is_ok_with_any_action(self):
1147
1020
for action, value in self.actions.items():
1148
args = self.actionargs(action, value, "--all")
1149
self.parse_args(args)
1021
options = self.parser.parse_args()
1022
setattr(options, action, value)
1023
options.all = True
1024
self.check_option_syntax(options)
1150
1025
1151
1026
def test_any_action_is_ok_with_one_client(self):
1152
1027
for action, value in self.actions.items():
1153
args = self.actionargs(action, value, "client")
1154
self.parse_args(args)
1028
options = self.parser.parse_args()
1029
setattr(options, action, value)
1030
options.client = ["foo"]
1031
self.check_option_syntax(options)
1155
1032
1156
1033
def test_one_client_with_all_actions_except_is_enabled(self):
1034
options = self.parser.parse_args()
1157
1035
for action, value in self.actions.items():
1158
if action == "--is-enabled":
1036
if action == "is_enabled":
1159
1037
continue
1160
args = self.actionargs(action, value, "client")
1161
self.parse_args(args)
1038
setattr(options, action, value)
1039
options.client = ["foo"]
1040
self.check_option_syntax(options)
1162
1041
1163
1042
def test_two_clients_with_all_actions_except_is_enabled(self):
1043
options = self.parser.parse_args()
1164
1044
for action, value in self.actions.items():
1165
if action == "--is-enabled":
1045
if action == "is_enabled":
1166
1046
continue
1167
args = self.actionargs(action, value, "client1",
1168
"client2")
1169
self.parse_args(args)
1047
setattr(options, action, value)
1048
options.client = ["foo", "barbar"]
1049
self.check_option_syntax(options)
1170
1050
1171
1051
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1172
1052
for action, value in self.actions.items():
1173
if action == "--is-enabled":
1053
if action == "is_enabled":
1174
1054
continue
1175
args = self.actionargs(action, value, "client1",
1176
"client2")
1177
self.parse_args(args)
1055
options = self.parser.parse_args()
1056
setattr(options, action, value)
1057
options.client = ["foo", "barbar"]
1058
self.check_option_syntax(options)
1178
1059
1179
1060
def test_is_enabled_fails_without_client(self):
1180
args = ["--is-enabled"]
1061
options = self.parser.parse_args()
1062
options.is_enabled = True
1181
1063
with self.assertParseError():
1182
self.parse_args(args)
1064
self.check_option_syntax(options)
1183
1065
1184
1066
def test_is_enabled_fails_with_two_clients(self):
1185
args = ["--is-enabled", "client1", "client2"]
1067
options = self.parser.parse_args()
1068
options.is_enabled = True
1069
options.client = ["foo", "barbar"]
1186
1070
with self.assertParseError():
1187
self.parse_args(args)
1071
self.check_option_syntax(options)
1188
1072
1189
1073
def test_remove_can_only_be_combined_with_action_deny(self):
1190
1074
for action, value in self.actions.items():
1191
if action in {"--remove", "--deny"}:
1075
if action in {"remove", "deny"}:
1192
1076
continue
1193
args = self.actionargs(action, value, "--all",
1194
"--remove")
1077
options = self.parser.parse_args()
1078
setattr(options, action, value)
1079
options.all = True
1080
options.remove = True
1195
1081
with self.assertParseError():
1196
self.parse_args(args)
1197
1198
1199
class Test_dbus_exceptions(unittest.TestCase):
1200
1201
def test_dbus_ConnectFailed_is_Error(self):
1202
with self.assertRaises(dbus.Error):
1203
raise dbus.ConnectFailed()
1204
1205
1206
class Test_dbus_MandosBus(unittest.TestCase):
1207
1208
class MockMandosBus(dbus.MandosBus):
1209
def __init__(self):
1210
self._name = "se.recompile.Mandos"
1211
self._server_path = "/"
1212
self._server_interface = "se.recompile.Mandos"
1213
self._client_interface = "se.recompile.Mandos.Client"
1214
self.calls = []
1215
self.call_method_return = Unique()
1216
1217
def call_method(self, methodname, busname, objectpath,
1218
interface, *args):
1219
self.calls.append((methodname, busname, objectpath,
1220
interface, args))
1221
return self.call_method_return
1222
1223
def setUp(self):
1224
self.bus = self.MockMandosBus()
1225
1226
def test_set_client_property(self):
1227
self.bus.set_client_property("objectpath", "key", "value")
1228
expected_call = ("Set", self.bus._name, "objectpath",
1229
"org.freedesktop.DBus.Properties",
1230
(self.bus._client_interface, "key", "value"))
1231
self.assertIn(expected_call, self.bus.calls)
1232
1233
def test_call_client_method(self):
1234
ret = self.bus.call_client_method("objectpath", "methodname")
1235
self.assertIs(self.bus.call_method_return, ret)
1236
expected_call = ("methodname", self.bus._name, "objectpath",
1237
self.bus._client_interface, ())
1238
self.assertIn(expected_call, self.bus.calls)
1239
1240
def test_call_client_method_with_args(self):
1241
args = (Unique(), Unique())
1242
ret = self.bus.call_client_method("objectpath", "methodname",
1243
*args)
1244
self.assertIs(self.bus.call_method_return, ret)
1245
expected_call = ("methodname", self.bus._name, "objectpath",
1246
self.bus._client_interface,
1247
(args[0], args[1]))
1248
self.assertIn(expected_call, self.bus.calls)
1249
1250
def test_get_clients_and_properties(self):
1251
managed_objects = {
1252
"objectpath": {
1253
self.bus._client_interface: {
1254
"key": "value",
1255
"bool": True,
1256
},
1257
"irrelevant_interface": {
1258
"key": "othervalue",
1259
"bool": False,
1260
},
1261
},
1262
"other_objectpath": {
1263
"other_irrelevant_interface": {
1264
"key": "value 3",
1265
"bool": None,
1266
},
1267
},
1268
}
1269
expected_clients_and_properties = {
1270
"objectpath": {
1271
"key": "value",
1272
"bool": True,
1273
}
1274
}
1275
self.bus.call_method_return = managed_objects
1276
ret = self.bus.get_clients_and_properties()
1277
self.assertDictEqual(expected_clients_and_properties, ret)
1278
expected_call = ("GetManagedObjects", self.bus._name,
1279
self.bus._server_path,
1280
"org.freedesktop.DBus.ObjectManager", ())
1281
self.assertIn(expected_call, self.bus.calls)
1282
1283
def test_call_server_method(self):
1284
ret = self.bus.call_server_method("methodname")
1285
self.assertIs(self.bus.call_method_return, ret)
1286
expected_call = ("methodname", self.bus._name,
1287
self.bus._server_path,
1288
self.bus._server_interface, ())
1289
self.assertIn(expected_call, self.bus.calls)
1290
1291
def test_call_server_method_with_args(self):
1292
args = (Unique(), Unique())
1293
ret = self.bus.call_server_method("methodname", *args)
1294
self.assertIs(self.bus.call_method_return, ret)
1295
expected_call = ("methodname", self.bus._name,
1296
self.bus._server_path,
1297
self.bus._server_interface,
1298
(args[0], args[1]))
1299
self.assertIn(expected_call, self.bus.calls)
1300
1301
1302
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1303
1304
def MockDBusPython_func(self, func):
1305
class mock_dbus_python(object):
1306
"""mock dbus-python module"""
1307
class exceptions(object):
1308
"""Pseudo-namespace"""
1309
class DBusException(Exception):
1310
pass
1311
class SystemBus(object):
1312
@staticmethod
1313
def get_object(busname, objectpath):
1314
DBusObject = collections.namedtuple(
1315
"DBusObject", ("methodname",))
1316
def method(*args, **kwargs):
1317
self.assertEqual({"dbus_interface":
1318
"interface"},
1319
kwargs)
1320
return func(*args)
1321
return DBusObject(methodname=method)
1322
class Boolean(object):
1323
def __init__(self, value):
1324
self.value = bool(value)
1325
def __bool__(self):
1326
return self.value
1327
if sys.version_info.major == 2:
1328
__nonzero__ = __bool__
1329
class ObjectPath(str):
1330
pass
1331
class Dictionary(dict):
1332
pass
1333
return mock_dbus_python
1334
1335
def call_method(self, bus, methodname, busname, objectpath,
1336
interface, *args):
1337
with self.assertLogs(log, logging.DEBUG):
1338
return bus.call_method(methodname, busname, objectpath,
1339
interface, *args)
1340
1341
def test_call_method_returns(self):
1342
expected_method_return = Unique()
1343
method_args = (Unique(), Unique())
1344
def func(*args):
1345
self.assertEqual(len(method_args), len(args))
1346
for marg, arg in zip(method_args, args):
1347
self.assertIs(marg, arg)
1348
return expected_method_return
1349
mock_dbus_python = self.MockDBusPython_func(func)
1350
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1351
ret = self.call_method(bus, "methodname", "busname",
1352
"objectpath", "interface",
1353
*method_args)
1354
self.assertIs(ret, expected_method_return)
1355
1356
def test_call_method_filters_bool_true(self):
1357
def func():
1358
return method_return
1359
mock_dbus_python = self.MockDBusPython_func(func)
1360
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1361
method_return = mock_dbus_python.Boolean(True)
1362
ret = self.call_method(bus, "methodname", "busname",
1363
"objectpath", "interface")
1364
self.assertTrue(ret)
1365
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1366
1367
def test_call_method_filters_bool_false(self):
1368
def func():
1369
return method_return
1370
mock_dbus_python = self.MockDBusPython_func(func)
1371
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1372
method_return = mock_dbus_python.Boolean(False)
1373
ret = self.call_method(bus, "methodname", "busname",
1374
"objectpath", "interface")
1375
self.assertFalse(ret)
1376
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1377
1378
def test_call_method_filters_objectpath(self):
1379
def func():
1380
return method_return
1381
mock_dbus_python = self.MockDBusPython_func(func)
1382
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1383
method_return = mock_dbus_python.ObjectPath("objectpath")
1384
ret = self.call_method(bus, "methodname", "busname",
1385
"objectpath", "interface")
1386
self.assertEqual("objectpath", ret)
1387
self.assertIsNot("objectpath", ret)
1388
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1389
1390
def test_call_method_filters_booleans_in_dict(self):
1391
def func():
1392
return method_return
1393
mock_dbus_python = self.MockDBusPython_func(func)
1394
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1395
method_return = mock_dbus_python.Dictionary(
1396
{mock_dbus_python.Boolean(True):
1397
mock_dbus_python.Boolean(False),
1398
mock_dbus_python.Boolean(False):
1399
mock_dbus_python.Boolean(True)})
1400
ret = self.call_method(bus, "methodname", "busname",
1401
"objectpath", "interface")
1402
expected_method_return = {True: False,
1403
False: True}
1404
self.assertEqual(expected_method_return, ret)
1405
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1406
1407
def test_call_method_filters_objectpaths_in_dict(self):
1408
def func():
1409
return method_return
1410
mock_dbus_python = self.MockDBusPython_func(func)
1411
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1412
method_return = mock_dbus_python.Dictionary(
1413
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1414
mock_dbus_python.ObjectPath("objectpath_value_1"),
1415
mock_dbus_python.ObjectPath("objectpath_key_2"):
1416
mock_dbus_python.ObjectPath("objectpath_value_2")})
1417
ret = self.call_method(bus, "methodname", "busname",
1418
"objectpath", "interface")
1419
expected_method_return = {str(key): str(value)
1420
for key, value in
1421
method_return.items()}
1422
self.assertEqual(expected_method_return, ret)
1423
self.assertIsInstance(ret, dict)
1424
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1425
1426
def test_call_method_filters_dict_in_dict(self):
1427
def func():
1428
return method_return
1429
mock_dbus_python = self.MockDBusPython_func(func)
1430
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1431
method_return = mock_dbus_python.Dictionary(
1432
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1433
"key12": "value12"}),
1434
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1435
"key22": "value22"})})
1436
ret = self.call_method(bus, "methodname", "busname",
1437
"objectpath", "interface")
1438
expected_method_return = {
1439
"key1": {"key11": "value11",
1440
"key12": "value12"},
1441
"key2": {"key21": "value21",
1442
"key22": "value22"},
1443
}
1444
self.assertEqual(expected_method_return, ret)
1445
self.assertIsInstance(ret, dict)
1446
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1447
for key, value in ret.items():
1448
self.assertIsInstance(value, dict)
1449
self.assertEqual(expected_method_return[key], value)
1450
self.assertNotIsInstance(value,
1451
mock_dbus_python.Dictionary)
1452
1453
def test_call_method_filters_dict_three_deep(self):
1454
def func():
1455
return method_return
1456
mock_dbus_python = self.MockDBusPython_func(func)
1457
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1458
method_return = mock_dbus_python.Dictionary(
1459
{"key1":
1460
mock_dbus_python.Dictionary(
1461
{"key2":
1462
mock_dbus_python.Dictionary(
1463
{"key3":
1464
mock_dbus_python.Boolean(True),
1465
}),
1466
}),
1467
})
1468
ret = self.call_method(bus, "methodname", "busname",
1469
"objectpath", "interface")
1470
expected_method_return = {"key1": {"key2": {"key3": True}}}
1471
self.assertEqual(expected_method_return, ret)
1472
self.assertIsInstance(ret, dict)
1473
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1474
self.assertIsInstance(ret["key1"], dict)
1475
self.assertNotIsInstance(ret["key1"],
1476
mock_dbus_python.Dictionary)
1477
self.assertIsInstance(ret["key1"]["key2"], dict)
1478
self.assertNotIsInstance(ret["key1"]["key2"],
1479
mock_dbus_python.Dictionary)
1480
self.assertTrue(ret["key1"]["key2"]["key3"])
1481
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1482
mock_dbus_python.Boolean)
1483
1484
def test_call_method_handles_exception(self):
1082
self.check_option_syntax(options)
1083
1084
1085
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1086
def test_calls_and_returns_get_object_on_bus(self):
1087
class MockBus(object):
1088
called = False
1089
def get_object(mockbus_self, busname, dbus_path):
1090
# Note that "self" is still the testcase instance,
1091
# this MockBus instance is in "mockbus_self".
1092
self.assertEqual(busname, dbus_busname)
1093
self.assertEqual(dbus_path, server_dbus_path)
1094
mockbus_self.called = True
1095
return mockbus_self
1096
1097
mockbus = get_mandos_dbus_object(bus=MockBus())
1098
self.assertIsInstance(mockbus, MockBus)
1099
self.assertTrue(mockbus.called)
1100
1101
def test_logs_and_exits_on_dbus_error(self):
1102
class MockBusFailing(object):
1103
def get_object(self, busname, dbus_path):
1104
raise dbus.exceptions.DBusException("Test")
1105
1106
with self.assertLogs(log, logging.CRITICAL):
1107
with self.assertRaises(SystemExit) as e:
1108
bus = get_mandos_dbus_object(bus=MockBusFailing())
1109
1110
if isinstance(e.exception.code, int):
1111
self.assertNotEqual(e.exception.code, 0)
1112
else:
1113
self.assertIsNotNone(e.exception.code)
1114
1115
1116
class Test_get_managed_objects(TestCaseWithAssertLogs):
1117
def test_calls_and_returns_GetManagedObjects(self):
1118
managed_objects = {"/clients/foo": { "Name": "foo"}}
1119
class MockObjectManager(object):
1120
def GetManagedObjects(self):
1121
return managed_objects
1122
retval = get_managed_objects(MockObjectManager())
1123
self.assertDictEqual(managed_objects, retval)
1124
1125
def test_logs_and_exits_on_dbus_error(self):
1485
1126
dbus_logger = logging.getLogger("dbus.proxies")
1486
1127
1487
def func():
1488
dbus_logger.error("Test")
1489
raise mock_dbus_python.exceptions.DBusException()
1490
1491
mock_dbus_python = self.MockDBusPython_func(func)
1492
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1128
class MockObjectManagerFailing(object):
1129
def GetManagedObjects(self):
1130
dbus_logger.error("Test")
1131
raise dbus.exceptions.DBusException("Test")
1493
1132
1494
1133
class CountingHandler(logging.Handler):
1495
1134
count = 0
1501
1140
dbus_logger.addHandler(counting_handler)
1502
1141
1503
1142
try:
1504
with self.assertRaises(dbus.Error) as e:
1505
self.call_method(bus, "methodname", "busname",
1506
"objectpath", "interface")
1143
with self.assertLogs(log, logging.CRITICAL) as watcher:
1144
with self.assertRaises(SystemExit) as e:
1145
get_managed_objects(MockObjectManagerFailing())
1507
1146
finally:
1508
1147
dbus_logger.removeFilter(counting_handler)
1509
1148
1510
self.assertNotIsInstance(e, dbus.ConnectFailed)
1511
1512
1149
# Make sure the dbus logger was suppressed
1513
self.assertEqual(0, counting_handler.count)
1514
1515
def test_get_object_converts_to_correct_exception(self):
1516
bus = dbus_python_adapter.SystemBus(
1517
self.fake_dbus_python_raises_exception_on_connect)
1518
with self.assertRaises(dbus.ConnectFailed):
1519
self.call_method(bus, "methodname", "busname",
1520
"objectpath", "interface")
1521
1522
class fake_dbus_python_raises_exception_on_connect(object):
1523
"""fake dbus-python module"""
1524
class exceptions(object):
1525
"""Pseudo-namespace"""
1526
class DBusException(Exception):
1527
pass
1528
1529
@classmethod
1530
def SystemBus(cls):
1531
def get_object(busname, objectpath):
1532
raise cls.exceptions.DBusException()
1533
Bus = collections.namedtuple("Bus", ["get_object"])
1534
return Bus(get_object=get_object)
1535
1536
1537
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1538
class mock_dbus_python(object):
1539
"""mock dbus-python modules"""
1540
class SystemBus(object):
1541
@staticmethod
1542
def get_object(busname, objectpath):
1543
return Unique()
1544
1545
def setUp(self):
1546
self.bus = dbus_python_adapter.CachingBus(
1547
self.mock_dbus_python)
1548
1549
def test_returns_distinct_objectpaths(self):
1550
obj1 = self.bus.get_object("busname", "objectpath1")
1551
self.assertIsInstance(obj1, Unique)
1552
obj2 = self.bus.get_object("busname", "objectpath2")
1553
self.assertIsInstance(obj2, Unique)
1554
self.assertIsNot(obj1, obj2)
1555
1556
def test_returns_distinct_busnames(self):
1557
obj1 = self.bus.get_object("busname1", "objectpath")
1558
self.assertIsInstance(obj1, Unique)
1559
obj2 = self.bus.get_object("busname2", "objectpath")
1560
self.assertIsInstance(obj2, Unique)
1561
self.assertIsNot(obj1, obj2)
1562
1563
def test_returns_distinct_both(self):
1564
obj1 = self.bus.get_object("busname1", "objectpath")
1565
self.assertIsInstance(obj1, Unique)
1566
obj2 = self.bus.get_object("busname2", "objectpath")
1567
self.assertIsInstance(obj2, Unique)
1568
self.assertIsNot(obj1, obj2)
1569
1570
def test_returns_same(self):
1571
obj1 = self.bus.get_object("busname", "objectpath")
1572
self.assertIsInstance(obj1, Unique)
1573
obj2 = self.bus.get_object("busname", "objectpath")
1574
self.assertIsInstance(obj2, Unique)
1575
self.assertIs(obj1, obj2)
1576
1577
def test_returns_same_old(self):
1578
obj1 = self.bus.get_object("busname1", "objectpath1")
1579
self.assertIsInstance(obj1, Unique)
1580
obj2 = self.bus.get_object("busname2", "objectpath2")
1581
self.assertIsInstance(obj2, Unique)
1582
obj1b = self.bus.get_object("busname1", "objectpath1")
1583
self.assertIsInstance(obj1b, Unique)
1584
self.assertIsNot(obj1, obj2)
1585
self.assertIsNot(obj2, obj1b)
1586
self.assertIs(obj1, obj1b)
1587
1588
1589
class Test_pydbus_adapter_SystemBus(TestCaseWithAssertLogs):
1590
1591
def Stub_pydbus_func(self, func):
1592
class stub_pydbus(object):
1593
"""stub pydbus module"""
1594
class SystemBus(object):
1595
@staticmethod
1596
def get(busname, objectpath):
1597
DBusObject = collections.namedtuple(
1598
"DBusObject", ("methodname",))
1599
return {"interface":
1600
DBusObject(methodname=func)}
1601
return stub_pydbus
1602
1603
def call_method(self, bus, methodname, busname, objectpath,
1604
interface, *args):
1605
with self.assertLogs(log, logging.DEBUG):
1606
return bus.call_method(methodname, busname, objectpath,
1607
interface, *args)
1608
1609
def test_call_method_returns(self):
1610
expected_method_return = Unique()
1611
method_args = (Unique(), Unique())
1612
def func(*args):
1613
self.assertEqual(len(method_args), len(args))
1614
for marg, arg in zip(method_args, args):
1615
self.assertIs(marg, arg)
1616
return expected_method_return
1617
stub_pydbus = self.Stub_pydbus_func(func)
1618
bus = pydbus_adapter.SystemBus(stub_pydbus)
1619
ret = self.call_method(bus, "methodname", "busname",
1620
"objectpath", "interface",
1621
*method_args)
1622
self.assertIs(ret, expected_method_return)
1623
1624
def test_call_method_handles_exception(self):
1625
dbus_logger = logging.getLogger("dbus.proxies")
1626
1627
def func():
1628
raise gi.repository.GLib.Error()
1629
1630
stub_pydbus = self.Stub_pydbus_func(func)
1631
bus = pydbus_adapter.SystemBus(stub_pydbus)
1632
1633
with self.assertRaises(dbus.Error) as e:
1634
self.call_method(bus, "methodname", "busname",
1635
"objectpath", "interface")
1636
1637
self.assertNotIsInstance(e, dbus.ConnectFailed)
1638
1639
def test_get_converts_to_correct_exception(self):
1640
bus = pydbus_adapter.SystemBus(
1641
self.fake_pydbus_raises_exception_on_connect)
1642
with self.assertRaises(dbus.ConnectFailed):
1643
self.call_method(bus, "methodname", "busname",
1644
"objectpath", "interface")
1645
1646
class fake_pydbus_raises_exception_on_connect(object):
1647
"""fake dbus-python module"""
1648
@classmethod
1649
def SystemBus(cls):
1650
def get(busname, objectpath):
1651
raise gi.repository.GLib.Error()
1652
Bus = collections.namedtuple("Bus", ["get"])
1653
return Bus(get=get)
1654
1655
def test_set_property_uses_setattr(self):
1656
class Object(object):
1657
pass
1658
obj = Object()
1659
class pydbus_spy(object):
1660
class SystemBus(object):
1661
@staticmethod
1662
def get(busname, objectpath):
1663
return {"interface": obj}
1664
bus = pydbus_adapter.SystemBus(pydbus_spy)
1665
value = Unique()
1666
bus.set_property("busname", "objectpath", "interface", "key",
1667
value)
1668
self.assertIs(value, obj.key)
1669
1670
def test_get_suppresses_xml_deprecation_warning(self):
1671
if sys.version_info.major >= 3:
1672
return
1673
class stub_pydbus_get(object):
1674
class SystemBus(object):
1675
@staticmethod
1676
def get(busname, objectpath):
1677
warnings.warn_explicit(
1678
"deprecated", DeprecationWarning,
1679
"xml.etree.ElementTree", 0)
1680
bus = pydbus_adapter.SystemBus(stub_pydbus_get)
1681
with warnings.catch_warnings(record=True) as w:
1682
warnings.simplefilter("always")
1683
bus.get("busname", "objectpath")
1684
self.assertEqual(0, len(w))
1685
1686
1687
class Test_pydbus_adapter_CachingBus(unittest.TestCase):
1688
class stub_pydbus(object):
1689
"""stub pydbus module"""
1690
class SystemBus(object):
1691
@staticmethod
1692
def get(busname, objectpath):
1693
return Unique()
1694
1695
def setUp(self):
1696
self.bus = pydbus_adapter.CachingBus(self.stub_pydbus)
1697
1698
def test_returns_distinct_objectpaths(self):
1699
obj1 = self.bus.get("busname", "objectpath1")
1700
self.assertIsInstance(obj1, Unique)
1701
obj2 = self.bus.get("busname", "objectpath2")
1702
self.assertIsInstance(obj2, Unique)
1703
self.assertIsNot(obj1, obj2)
1704
1705
def test_returns_distinct_busnames(self):
1706
obj1 = self.bus.get("busname1", "objectpath")
1707
self.assertIsInstance(obj1, Unique)
1708
obj2 = self.bus.get("busname2", "objectpath")
1709
self.assertIsInstance(obj2, Unique)
1710
self.assertIsNot(obj1, obj2)
1711
1712
def test_returns_distinct_both(self):
1713
obj1 = self.bus.get("busname1", "objectpath")
1714
self.assertIsInstance(obj1, Unique)
1715
obj2 = self.bus.get("busname2", "objectpath")
1716
self.assertIsInstance(obj2, Unique)
1717
self.assertIsNot(obj1, obj2)
1718
1719
def test_returns_same(self):
1720
obj1 = self.bus.get("busname", "objectpath")
1721
self.assertIsInstance(obj1, Unique)
1722
obj2 = self.bus.get("busname", "objectpath")
1723
self.assertIsInstance(obj2, Unique)
1724
self.assertIs(obj1, obj2)
1725
1726
def test_returns_same_old(self):
1727
obj1 = self.bus.get("busname1", "objectpath1")
1728
self.assertIsInstance(obj1, Unique)
1729
obj2 = self.bus.get("busname2", "objectpath2")
1730
self.assertIsInstance(obj2, Unique)
1731
obj1b = self.bus.get("busname1", "objectpath1")
1732
self.assertIsInstance(obj1b, Unique)
1733
self.assertIsNot(obj1, obj2)
1734
self.assertIsNot(obj2, obj1b)
1735
self.assertIs(obj1, obj1b)
1150
self.assertEqual(counting_handler.count, 0)
1151
1152
# Test that the dbus_logger still works
1153
with self.assertLogs(dbus_logger, logging.ERROR):
1154
dbus_logger.error("Test")
1155
1156
if isinstance(e.exception.code, int):
1157
self.assertNotEqual(e.exception.code, 0)
1158
else:
1159
self.assertIsNotNone(e.exception.code)
1736
1160
1737
1161
1738
1162
class Test_commands_from_options(unittest.TestCase):
1739
1740
1163
def setUp(self):
1741
1164
self.parser = argparse.ArgumentParser()
1742
1165
add_command_line_options(self.parser)
1743
1166
1744
1167
def test_is_enabled(self):
1745
self.assert_command_from_args(["--is-enabled", "client"],
1168
self.assert_command_from_args(["--is-enabled", "foo"],
1746
1169
command.IsEnabled)
1747
1170
1748
1171
def assert_command_from_args(self, args, command_cls,
1752
1175
options = self.parser.parse_args(args)
1753
1176
check_option_syntax(self.parser, options)
1754
1177
commands = commands_from_options(options)
1755
self.assertEqual(1, len(commands))
1178
self.assertEqual(len(commands), 1)
1756
1179
command = commands[0]
1757
1180
self.assertIsInstance(command, command_cls)
1758
1181
for key, value in cmd_attrs.items():
1759
self.assertEqual(value, getattr(command, key))
1182
self.assertEqual(getattr(command, key), value)
1760
1183
1761
1184
def test_is_enabled_short(self):
1762
self.assert_command_from_args(["-V", "client"],
1185
self.assert_command_from_args(["-V", "foo"],
1763
1186
command.IsEnabled)
1764
1187
1765
1188
def test_approve(self):
1766
self.assert_command_from_args(["--approve", "client"],
1189
self.assert_command_from_args(["--approve", "foo"],
1767
1190
command.Approve)
1768
1191
1769
1192
def test_approve_short(self):
1770
self.assert_command_from_args(["-A", "client"],
1771
command.Approve)
1193
self.assert_command_from_args(["-A", "foo"], command.Approve)
1772
1194
1773
1195
def test_deny(self):
1774
self.assert_command_from_args(["--deny", "client"],
1775
command.Deny)
1196
self.assert_command_from_args(["--deny", "foo"], command.Deny)
1776
1197
1777
1198
def test_deny_short(self):
1778
self.assert_command_from_args(["-D", "client"], command.Deny)
1199
self.assert_command_from_args(["-D", "foo"], command.Deny)
1779
1200
1780
1201
def test_remove(self):
1781
self.assert_command_from_args(["--remove", "client"],
1202
self.assert_command_from_args(["--remove", "foo"],
1782
1203
command.Remove)
1783
1204
1784
1205
def test_deny_before_remove(self):
1785
1206
options = self.parser.parse_args(["--deny", "--remove",
1786
"client"])
1207
"foo"])
1787
1208
check_option_syntax(self.parser, options)
1788
1209
commands = commands_from_options(options)
1789
self.assertEqual(2, len(commands))
1210
self.assertEqual(len(commands), 2)
1790
1211
self.assertIsInstance(commands[0], command.Deny)
1791
1212
self.assertIsInstance(commands[1], command.Remove)
1792
1213
1795
1216
"--all"])
1796
1217
check_option_syntax(self.parser, options)
1797
1218
commands = commands_from_options(options)
1798
self.assertEqual(2, len(commands))
1219
self.assertEqual(len(commands), 2)
1799
1220
self.assertIsInstance(commands[0], command.Deny)
1800
1221
self.assertIsInstance(commands[1], command.Remove)
1801
1222
1802
1223
def test_remove_short(self):
1803
self.assert_command_from_args(["-r", "client"],
1804
command.Remove)
1224
self.assert_command_from_args(["-r", "foo"], command.Remove)
1805
1225
1806
1226
def test_dump_json(self):
1807
1227
self.assert_command_from_args(["--dump-json"],
1808
1228
command.DumpJSON)
1809
1229
1810
1230
def test_enable(self):
1811
self.assert_command_from_args(["--enable", "client"],
1231
self.assert_command_from_args(["--enable", "foo"],
1812
1232
command.Enable)
1813
1233
1814
1234
def test_enable_short(self):
1815
self.assert_command_from_args(["-e", "client"],
1816
command.Enable)
1235
self.assert_command_from_args(["-e", "foo"], command.Enable)
1817
1236
1818
1237
def test_disable(self):
1819
self.assert_command_from_args(["--disable", "client"],
1238
self.assert_command_from_args(["--disable", "foo"],
1820
1239
command.Disable)
1821
1240
1822
1241
def test_disable_short(self):
1823
self.assert_command_from_args(["-d", "client"],
1824
command.Disable)
1242
self.assert_command_from_args(["-d", "foo"], command.Disable)
1825
1243
1826
1244
def test_bump_timeout(self):
1827
self.assert_command_from_args(["--bump-timeout", "client"],
1245
self.assert_command_from_args(["--bump-timeout", "foo"],
1828
1246
command.BumpTimeout)
1829
1247
1830
1248
def test_bump_timeout_short(self):
1831
self.assert_command_from_args(["-b", "client"],
1249
self.assert_command_from_args(["-b", "foo"],
1832
1250
command.BumpTimeout)
1833
1251
1834
1252
def test_start_checker(self):
1835
self.assert_command_from_args(["--start-checker", "client"],
1253
self.assert_command_from_args(["--start-checker", "foo"],
1836
1254
command.StartChecker)
1837
1255
1838
1256
def test_stop_checker(self):
1839
self.assert_command_from_args(["--stop-checker", "client"],
1257
self.assert_command_from_args(["--stop-checker", "foo"],
1840
1258
command.StopChecker)
1841
1259
1842
1260
def test_approve_by_default(self):
1843
self.assert_command_from_args(["--approve-by-default",
1844
"client"],
1261
self.assert_command_from_args(["--approve-by-default", "foo"],
1845
1262
command.ApproveByDefault)
1846
1263
1847
1264
def test_deny_by_default(self):
1848
self.assert_command_from_args(["--deny-by-default", "client"],
1265
self.assert_command_from_args(["--deny-by-default", "foo"],
1849
1266
command.DenyByDefault)
1850
1267
1851
1268
def test_checker(self):
1852
self.assert_command_from_args(["--checker", ":", "client"],
1269
self.assert_command_from_args(["--checker", ":", "foo"],
1853
1270
command.SetChecker,
1854
1271
value_to_set=":")
1855
1272
1856
1273
def test_checker_empty(self):
1857
self.assert_command_from_args(["--checker", "", "client"],
1274
self.assert_command_from_args(["--checker", "", "foo"],
1858
1275
command.SetChecker,
1859
1276
value_to_set="")
1860
1277
1861
1278
def test_checker_short(self):
1862
self.assert_command_from_args(["-c", ":", "client"],
1279
self.assert_command_from_args(["-c", ":", "foo"],
1863
1280
command.SetChecker,
1864
1281
value_to_set=":")
1865
1282
1866
1283
def test_host(self):
1867
self.assert_command_from_args(
1868
["--host", "client.example.org", "client"],
1869
command.SetHost, value_to_set="client.example.org")
1284
self.assert_command_from_args(["--host", "foo.example.org",
1285
"foo"], command.SetHost,
1286
value_to_set="foo.example.org")
1870
1287
1871
1288
def test_host_short(self):
1872
self.assert_command_from_args(
1873
["-H", "client.example.org", "client"], command.SetHost,
1874
value_to_set="client.example.org")
1289
self.assert_command_from_args(["-H", "foo.example.org",
1290
"foo"], command.SetHost,
1291
value_to_set="foo.example.org")
1875
1292
1876
1293
def test_secret_devnull(self):
1877
1294
self.assert_command_from_args(["--secret", os.path.devnull,
1878
"client"], command.SetSecret,
1295
"foo"], command.SetSecret,
1879
1296
value_to_set=b"")
1880
1297
1881
1298
def test_secret_tempfile(self):
1884
1301
f.write(value)
1885
1302
f.seek(0)
1886
1303
self.assert_command_from_args(["--secret", f.name,
1887
"client"],
1888
command.SetSecret,
1304
"foo"], command.SetSecret,
1889
1305
value_to_set=value)
1890
1306
1891
1307
def test_secret_devnull_short(self):
1892
self.assert_command_from_args(["-s", os.path.devnull,
1893
"client"], command.SetSecret,
1308
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1309
command.SetSecret,
1894
1310
value_to_set=b"")
1895
1311
1896
1312
def test_secret_tempfile_short(self):
1898
1314
value = b"secret\0xyzzy\nbar"
1899
1315
f.write(value)
1900
1316
f.seek(0)
1901
self.assert_command_from_args(["-s", f.name, "client"],
1317
self.assert_command_from_args(["-s", f.name, "foo"],
1902
1318
command.SetSecret,
1903
1319
value_to_set=value)
1904
1320
1905
1321
def test_timeout(self):
1906
self.assert_command_from_args(["--timeout", "PT5M", "client"],
1322
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1907
1323
command.SetTimeout,
1908
1324
value_to_set=300000)
1909
1325
1910
1326
def test_timeout_short(self):
1911
self.assert_command_from_args(["-t", "PT5M", "client"],
1327
self.assert_command_from_args(["-t", "PT5M", "foo"],
1912
1328
command.SetTimeout,
1913
1329
value_to_set=300000)
1914
1330
1915
1331
def test_extended_timeout(self):
1916
1332
self.assert_command_from_args(["--extended-timeout", "PT15M",
1917
"client"],
1333
"foo"],
1918
1334
command.SetExtendedTimeout,
1919
1335
value_to_set=900000)
1920
1336
1921
1337
def test_interval(self):
1922
self.assert_command_from_args(["--interval", "PT2M",
1923
"client"], command.SetInterval,
1338
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1339
command.SetInterval,
1924
1340
value_to_set=120000)
1925
1341
1926
1342
def test_interval_short(self):
1927
self.assert_command_from_args(["-i", "PT2M", "client"],
1343
self.assert_command_from_args(["-i", "PT2M", "foo"],
1928
1344
command.SetInterval,
1929
1345
value_to_set=120000)
1930
1346
1931
1347
def test_approval_delay(self):
1932
1348
self.assert_command_from_args(["--approval-delay", "PT30S",
1933
"client"],
1349
"foo"],
1934
1350
command.SetApprovalDelay,
1935
1351
value_to_set=30000)
1936
1352
1937
1353
def test_approval_duration(self):
1938
1354
self.assert_command_from_args(["--approval-duration", "PT1S",
1939
"client"],
1355
"foo"],
1940
1356
command.SetApprovalDuration,
1941
1357
value_to_set=1000)
1942
1358
1957
1373
class TestCommand(unittest.TestCase):
1958
1374
"""Abstract class for tests of command classes"""
1959
1375
1960
class FakeMandosBus(dbus.MandosBus):
1961
def __init__(self, testcase):
1962
self.client_properties = {
1963
"Name": "foo",
1964
"KeyID": ("92ed150794387c03ce684574b1139a65"
1965
"94a34f895daaaf09fd8ea90a27cddb12"),
1966
"Secret": b"secret",
1967
"Host": "foo.example.org",
1968
"Enabled": True,
1969
"Timeout": 300000,
1970
"LastCheckedOK": "2019-02-03T00:00:00",
1971
"Created": "2019-01-02T00:00:00",
1972
"Interval": 120000,
1973
"Fingerprint": ("778827225BA7DE539C5A"
1974
"7CFA59CFF7CDBD9A5920"),
1975
"CheckerRunning": False,
1976
"LastEnabled": "2019-01-03T00:00:00",
1977
"ApprovalPending": False,
1978
"ApprovedByDefault": True,
1979
"LastApprovalRequest": "",
1980
"ApprovalDelay": 0,
1981
"ApprovalDuration": 1000,
1982
"Checker": "fping -q -- %(host)s",
1983
"ExtendedTimeout": 900000,
1984
"Expires": "2019-02-04T00:00:00",
1985
"LastCheckerStatus": 0,
1986
}
1987
self.other_client_properties = {
1988
"Name": "barbar",
1989
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1990
"6ab612cff5ad227247e46c2b020f441c"),
1991
"Secret": b"secretbar",
1992
"Host": "192.0.2.3",
1993
"Enabled": True,
1994
"Timeout": 300000,
1995
"LastCheckedOK": "2019-02-04T00:00:00",
1996
"Created": "2019-01-03T00:00:00",
1997
"Interval": 120000,
1998
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1999
"F547B3A107558FCA3A27"),
2000
"CheckerRunning": True,
2001
"LastEnabled": "2019-01-04T00:00:00",
2002
"ApprovalPending": False,
2003
"ApprovedByDefault": False,
2004
"LastApprovalRequest": "2019-01-03T00:00:00",
2005
"ApprovalDelay": 30000,
2006
"ApprovalDuration": 93785000,
2007
"Checker": ":",
2008
"ExtendedTimeout": 900000,
2009
"Expires": "2019-02-05T00:00:00",
2010
"LastCheckerStatus": -2,
2011
}
2012
self.clients = collections.OrderedDict(
2013
[
2014
("client_objectpath", self.client_properties),
2015
("other_client_objectpath",
2016
self.other_client_properties),
2017
])
2018
self.one_client = {"client_objectpath":
2019
self.client_properties}
2020
self.testcase = testcase
2021
self.calls = []
2022
2023
def call_method(self, methodname, busname, objectpath,
2024
interface, *args):
2025
self.testcase.assertEqual("se.recompile.Mandos", busname)
2026
self.calls.append((methodname, busname, objectpath,
2027
interface, args))
2028
if interface == "org.freedesktop.DBus.Properties":
2029
if methodname == "Set":
2030
self.testcase.assertEqual(3, len(args))
2031
interface, key, value = args
2032
self.testcase.assertEqual(
2033
"se.recompile.Mandos.Client", interface)
2034
self.clients[objectpath][key] = value
2035
return
2036
elif interface == "se.recompile.Mandos":
2037
self.testcase.assertEqual("RemoveClient", methodname)
2038
self.testcase.assertEqual(1, len(args))
2039
clientpath = args[0]
2040
del self.clients[clientpath]
2041
return
2042
elif interface == "se.recompile.Mandos.Client":
2043
if methodname == "Approve":
2044
self.testcase.assertEqual(1, len(args))
2045
return
2046
raise ValueError()
2047
2048
1376
def setUp(self):
2049
self.bus = self.FakeMandosBus(self)
1377
testcase = self
1378
class MockClient(object):
1379
def __init__(self, name, **attributes):
1380
self.__dbus_object_path__ = "/clients/{}".format(name)
1381
self.attributes = attributes
1382
self.attributes["Name"] = name
1383
self.calls = []
1384
def Set(self, interface, propname, value, dbus_interface):
1385
testcase.assertEqual(interface, client_dbus_interface)
1386
testcase.assertEqual(dbus_interface,
1387
dbus.PROPERTIES_IFACE)
1388
self.attributes[propname] = value
1389
def Approve(self, approve, dbus_interface):
1390
testcase.assertEqual(dbus_interface,
1391
client_dbus_interface)
1392
self.calls.append(("Approve", (approve,
1393
dbus_interface)))
1394
self.client = MockClient(
1395
"foo",
1396
KeyID=("92ed150794387c03ce684574b1139a65"
1397
"94a34f895daaaf09fd8ea90a27cddb12"),
1398
Secret=b"secret",
1399
Host="foo.example.org",
1400
Enabled=dbus.Boolean(True),
1401
Timeout=300000,
1402
LastCheckedOK="2019-02-03T00:00:00",
1403
Created="2019-01-02T00:00:00",
1404
Interval=120000,
1405
Fingerprint=("778827225BA7DE539C5A"
1406
"7CFA59CFF7CDBD9A5920"),
1407
CheckerRunning=dbus.Boolean(False),
1408
LastEnabled="2019-01-03T00:00:00",
1409
ApprovalPending=dbus.Boolean(False),
1410
ApprovedByDefault=dbus.Boolean(True),
1411
LastApprovalRequest="",
1412
ApprovalDelay=0,
1413
ApprovalDuration=1000,
1414
Checker="fping -q -- %(host)s",
1415
ExtendedTimeout=900000,
1416
Expires="2019-02-04T00:00:00",
1417
LastCheckerStatus=0)
1418
self.other_client = MockClient(
1419
"barbar",
1420
KeyID=("0558568eedd67d622f5c83b35a115f79"
1421
"6ab612cff5ad227247e46c2b020f441c"),
1422
Secret=b"secretbar",
1423
Host="192.0.2.3",
1424
Enabled=dbus.Boolean(True),
1425
Timeout=300000,
1426
LastCheckedOK="2019-02-04T00:00:00",
1427
Created="2019-01-03T00:00:00",
1428
Interval=120000,
1429
Fingerprint=("3E393AEAEFB84C7E89E2"
1430
"F547B3A107558FCA3A27"),
1431
CheckerRunning=dbus.Boolean(True),
1432
LastEnabled="2019-01-04T00:00:00",
1433
ApprovalPending=dbus.Boolean(False),
1434
ApprovedByDefault=dbus.Boolean(False),
1435
LastApprovalRequest="2019-01-03T00:00:00",
1436
ApprovalDelay=30000,
1437
ApprovalDuration=93785000,
1438
Checker=":",
1439
ExtendedTimeout=900000,
1440
Expires="2019-02-05T00:00:00",
1441
LastCheckerStatus=-2)
1442
self.clients = collections.OrderedDict(
1443
[
1444
("/clients/foo", self.client.attributes),
1445
("/clients/barbar", self.other_client.attributes),
1446
])
1447
self.one_client = {"/clients/foo": self.client.attributes}
1448
1449
@property
1450
def bus(self):
1451
class Bus(object):
1452
@staticmethod
1453
def get_object(client_bus_name, path):
1454
self.assertEqual(client_bus_name, dbus_busname)
1455
return {
1456
# Note: "self" here is the TestCmd instance, not
1457
# the Bus instance, since this is a static method!
1458
"/clients/foo": self.client,
1459
"/clients/barbar": self.other_client,
1460
}[path]
1461
return Bus()
2050
1462
2051
1463
2052
1464
class TestBaseCommands(TestCommand):
2053
1465
2054
def test_IsEnabled_exits_successfully(self):
1466
def test_IsEnabled(self):
1467
self.assertTrue(all(command.IsEnabled().is_enabled(client,
1468
properties)
1469
for client, properties
1470
in self.clients.items()))
1471
1472
def test_IsEnabled_run_exits_successfully(self):
2055
1473
with self.assertRaises(SystemExit) as e:
2056
command.IsEnabled().run(self.bus.one_client)
1474
command.IsEnabled().run(self.one_client)
2057
1475
if e.exception.code is not None:
2058
self.assertEqual(0, e.exception.code)
1476
self.assertEqual(e.exception.code, 0)
2059
1477
else:
2060
1478
self.assertIsNone(e.exception.code)
2061
1479
2062
def test_IsEnabled_exits_with_failure(self):
2063
self.bus.client_properties["Enabled"] = False
1480
def test_IsEnabled_run_exits_with_failure(self):
1481
self.client.attributes["Enabled"] = dbus.Boolean(False)
2064
1482
with self.assertRaises(SystemExit) as e:
2065
command.IsEnabled().run(self.bus.one_client)
1483
command.IsEnabled().run(self.one_client)
2066
1484
if isinstance(e.exception.code, int):
2067
self.assertNotEqual(0, e.exception.code)
1485
self.assertNotEqual(e.exception.code, 0)
2068
1486
else:
2069
1487
self.assertIsNotNone(e.exception.code)
2070
1488
2071
1489
def test_Approve(self):
2072
busname = "se.recompile.Mandos"
2073
client_interface = "se.recompile.Mandos.Client"
2074
command.Approve().run(self.bus.clients, self.bus)
2075
for clientpath in self.bus.clients:
2076
self.assertIn(("Approve", busname, clientpath,
2077
client_interface, (True,)), self.bus.calls)
1490
command.Approve().run(self.clients, self.bus)
1491
for clientpath in self.clients:
1492
client = self.bus.get_object(dbus_busname, clientpath)
1493
self.assertIn(("Approve", (True, client_dbus_interface)),
1494
client.calls)
2078
1495
2079
1496
def test_Deny(self):
2080
busname = "se.recompile.Mandos"
2081
client_interface = "se.recompile.Mandos.Client"
2082
command.Deny().run(self.bus.clients, self.bus)
2083
for clientpath in self.bus.clients:
2084
self.assertIn(("Approve", busname, clientpath,
2085
client_interface, (False,)),
2086
self.bus.calls)
1497
command.Deny().run(self.clients, self.bus)
1498
for clientpath in self.clients:
1499
client = self.bus.get_object(dbus_busname, clientpath)
1500
self.assertIn(("Approve", (False, client_dbus_interface)),
1501
client.calls)
2087
1502
2088
1503
def test_Remove(self):
2089
command.Remove().run(self.bus.clients, self.bus)
2090
for clientpath in self.bus.clients:
2091
self.assertIn(("RemoveClient", dbus_busname,
2092
dbus_server_path, dbus_server_interface,
2093
(clientpath,)), self.bus.calls)
1504
class MockMandos(object):
1505
def __init__(self):
1506
self.calls = []
1507
def RemoveClient(self, dbus_path):
1508
self.calls.append(("RemoveClient", (dbus_path,)))
1509
mandos = MockMandos()
1510
command.Remove().run(self.clients, self.bus, mandos)
1511
for clientpath in self.clients:
1512
self.assertIn(("RemoveClient", (clientpath,)),
1513
mandos.calls)
2094
1514
2095
1515
expected_json = {
2096
1516
"foo": {
2144
1564
}
2145
1565
2146
1566
def test_DumpJSON_normal(self):
2147
with self.capture_stdout_to_buffer() as buffer:
2148
command.DumpJSON().run(self.bus.clients)
2149
json_data = json.loads(buffer.getvalue())
2150
self.assertDictEqual(self.expected_json, json_data)
2151
2152
@staticmethod
2153
@contextlib.contextmanager
2154
def capture_stdout_to_buffer():
2155
capture_buffer = io.StringIO()
2156
old_stdout = sys.stdout
2157
sys.stdout = capture_buffer
2158
try:
2159
yield capture_buffer
2160
finally:
2161
sys.stdout = old_stdout
1567
output = command.DumpJSON().output(self.clients.values())
1568
json_data = json.loads(output)
1569
self.assertDictEqual(json_data, self.expected_json)
2162
1570
2163
1571
def test_DumpJSON_one_client(self):
2164
with self.capture_stdout_to_buffer() as buffer:
2165
command.DumpJSON().run(self.bus.one_client)
2166
json_data = json.loads(buffer.getvalue())
1572
output = command.DumpJSON().output(self.one_client.values())
1573
json_data = json.loads(output)
2167
1574
expected_json = {"foo": self.expected_json["foo"]}
2168
self.assertDictEqual(expected_json, json_data)
1575
self.assertDictEqual(json_data, expected_json)
2169
1576
2170
1577
def test_PrintTable_normal(self):
2171
with self.capture_stdout_to_buffer() as buffer:
2172
command.PrintTable().run(self.bus.clients)
1578
output = command.PrintTable().output(self.clients.values())
2173
1579
expected_output = "\n".join((
2174
1580
"Name Enabled Timeout Last Successful Check",
2175
1581
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2176
1582
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
2177
)) + "\n"
2178
self.assertEqual(expected_output, buffer.getvalue())
1583
))
1584
self.assertEqual(output, expected_output)
2179
1585
2180
1586
def test_PrintTable_verbose(self):
2181
with self.capture_stdout_to_buffer() as buffer:
2182
command.PrintTable(verbose=True).run(self.bus.clients)
1587
output = command.PrintTable(verbose=True).output(
1588
self.clients.values())
2183
1589
columns = (
2184
1590
(
2185
1591
"Name ",
2267
1673
)
2268
1674
)
2269
1675
num_lines = max(len(rows) for rows in columns)
2270
expected_output = ("\n".join("".join(rows[line]
2271
for rows in columns)
2272
for line in range(num_lines))
2273
+ "\n")
2274
self.assertEqual(expected_output, buffer.getvalue())
1676
expected_output = "\n".join("".join(rows[line]
1677
for rows in columns)
1678
for line in range(num_lines))
1679
self.assertEqual(output, expected_output)
2275
1680
2276
1681
def test_PrintTable_one_client(self):
2277
with self.capture_stdout_to_buffer() as buffer:
2278
command.PrintTable().run(self.bus.one_client)
1682
output = command.PrintTable().output(self.one_client.values())
2279
1683
expected_output = "\n".join((
2280
1684
"Name Enabled Timeout Last Successful Check",
2281
1685
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2282
)) + "\n"
2283
self.assertEqual(expected_output, buffer.getvalue())
2284
2285
2286
class TestPropertySetterCmd(TestCommand):
2287
"""Abstract class for tests of command.PropertySetter classes"""
2288
1686
))
1687
self.assertEqual(output, expected_output)
1688
1689
1690
class TestPropertyCmd(TestCommand):
1691
"""Abstract class for tests of command.Property classes"""
2289
1692
def runTest(self):
2290
1693
if not hasattr(self, "command"):
2291
return # Abstract TestCase class
2292
2293
if hasattr(self, "values_to_set"):
2294
cmd_args = [(value,) for value in self.values_to_set]
2295
values_to_get = getattr(self, "values_to_get",
2296
self.values_to_set)
2297
else:
2298
cmd_args = [() for x in range(len(self.values_to_get))]
2299
values_to_get = self.values_to_get
2300
for value_to_get, cmd_arg in zip(values_to_get, cmd_args):
2301
for clientpath in self.bus.clients:
2302
self.bus.clients[clientpath][self.propname] = (
2303
Unique())
2304
self.command(*cmd_arg).run(self.bus.clients, self.bus)
2305
for clientpath in self.bus.clients:
2306
value = (self.bus.clients[clientpath]
2307
[self.propname])
2308
self.assertNotIsInstance(value, Unique)
2309
self.assertEqual(value_to_get, value)
2310
2311
2312
class TestEnableCmd(TestPropertySetterCmd):
1694
return
1695
values_to_get = getattr(self, "values_to_get",
1696
self.values_to_set)
1697
for value_to_set, value_to_get in zip(self.values_to_set,
1698
values_to_get):
1699
for clientpath in self.clients:
1700
client = self.bus.get_object(dbus_busname, clientpath)
1701
old_value = client.attributes[self.propname]
1702
client.attributes[self.propname] = self.Unique()
1703
self.run_command(value_to_set, self.clients)
1704
for clientpath in self.clients:
1705
client = self.bus.get_object(dbus_busname, clientpath)
1706
value = client.attributes[self.propname]
1707
self.assertNotIsInstance(value, self.Unique)
1708
self.assertEqual(value, value_to_get)
1709
1710
class Unique(object):
1711
"""Class for objects which exist only to be unique objects,
1712
since unittest.mock.sentinel only exists in Python 3.3"""
1713
1714
def run_command(self, value, clients):
1715
self.command().run(clients, self.bus)
1716
1717
1718
class TestEnableCmd(TestPropertyCmd):
2313
1719
command = command.Enable
2314
1720
propname = "Enabled"
2315
values_to_get = [True]
2316
2317
2318
class TestDisableCmd(TestPropertySetterCmd):
1721
values_to_set = [dbus.Boolean(True)]
1722
1723
1724
class TestDisableCmd(TestPropertyCmd):
2319
1725
command = command.Disable
2320
1726
propname = "Enabled"
2321
values_to_get = [False]
2322
2323
2324
class TestBumpTimeoutCmd(TestPropertySetterCmd):
1727
values_to_set = [dbus.Boolean(False)]
1728
1729
1730
class TestBumpTimeoutCmd(TestPropertyCmd):
2325
1731
command = command.BumpTimeout
2326
1732
propname = "LastCheckedOK"
2327
values_to_get = [""]
2328
2329
2330
class TestStartCheckerCmd(TestPropertySetterCmd):
1733
values_to_set = [""]
1734
1735
1736
class TestStartCheckerCmd(TestPropertyCmd):
2331
1737
command = command.StartChecker
2332
1738
propname = "CheckerRunning"
2333
values_to_get = [True]
2334
2335
2336
class TestStopCheckerCmd(TestPropertySetterCmd):
1739
values_to_set = [dbus.Boolean(True)]
1740
1741
1742
class TestStopCheckerCmd(TestPropertyCmd):
2337
1743
command = command.StopChecker
2338
1744
propname = "CheckerRunning"
2339
values_to_get = [False]
2340
2341
2342
class TestApproveByDefaultCmd(TestPropertySetterCmd):
1745
values_to_set = [dbus.Boolean(False)]
1746
1747
1748
class TestApproveByDefaultCmd(TestPropertyCmd):
2343
1749
command = command.ApproveByDefault
2344
1750
propname = "ApprovedByDefault"
2345
values_to_get = [True]
2346
2347
2348
class TestDenyByDefaultCmd(TestPropertySetterCmd):
1751
values_to_set = [dbus.Boolean(True)]
1752
1753
1754
class TestDenyByDefaultCmd(TestPropertyCmd):
2349
1755
command = command.DenyByDefault
2350
1756
propname = "ApprovedByDefault"
2351
values_to_get = [False]
2352
2353
2354
class TestSetCheckerCmd(TestPropertySetterCmd):
1757
values_to_set = [dbus.Boolean(False)]
1758
1759
1760
class TestPropertyValueCmd(TestPropertyCmd):
1761
"""Abstract class for tests of PropertyValueCmd classes"""
1762
1763
def runTest(self):
1764
if type(self) is TestPropertyValueCmd:
1765
return
1766
return super(TestPropertyValueCmd, self).runTest()
1767
1768
def run_command(self, value, clients):
1769
self.command(value).run(clients, self.bus)
1770
1771
1772
class TestSetCheckerCmd(TestPropertyValueCmd):
2355
1773
command = command.SetChecker
2356
1774
propname = "Checker"
2357
1775
values_to_set = ["", ":", "fping -q -- %s"]
2358
1776
2359
1777
2360
class TestSetHostCmd(TestPropertySetterCmd):
1778
class TestSetHostCmd(TestPropertyValueCmd):
2361
1779
command = command.SetHost
2362
1780
propname = "Host"
2363
values_to_set = ["192.0.2.3", "client.example.org"]
2364
2365
2366
class TestSetSecretCmd(TestPropertySetterCmd):
1781
values_to_set = ["192.0.2.3", "foo.example.org"]
1782
1783
1784
class TestSetSecretCmd(TestPropertyValueCmd):
2367
1785
command = command.SetSecret
2368
1786
propname = "Secret"
2369
1787
values_to_set = [io.BytesIO(b""),
2370
1788
io.BytesIO(b"secret\0xyzzy\nbar")]
2371
values_to_get = [f.getvalue() for f in values_to_set]
2372
2373
2374
class TestSetTimeoutCmd(TestPropertySetterCmd):
1789
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1790
1791
1792
class TestSetTimeoutCmd(TestPropertyValueCmd):
2375
1793
command = command.SetTimeout
2376
1794
propname = "Timeout"
2377
1795
values_to_set = [datetime.timedelta(),
2379
1797
datetime.timedelta(seconds=1),
2380
1798
datetime.timedelta(weeks=1),
2381
1799
datetime.timedelta(weeks=52)]
2382
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2383
2384
2385
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
1800
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1801
1802
1803
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
2386
1804
command = command.SetExtendedTimeout
2387
1805
propname = "ExtendedTimeout"
2388
1806
values_to_set = [datetime.timedelta(),
2390
1808
datetime.timedelta(seconds=1),
2391
1809
datetime.timedelta(weeks=1),
2392
1810
datetime.timedelta(weeks=52)]
2393
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2394
2395
2396
class TestSetIntervalCmd(TestPropertySetterCmd):
1811
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1812
1813
1814
class TestSetIntervalCmd(TestPropertyValueCmd):
2397
1815
command = command.SetInterval
2398
1816
propname = "Interval"
2399
1817
values_to_set = [datetime.timedelta(),
2401
1819
datetime.timedelta(seconds=1),
2402
1820
datetime.timedelta(weeks=1),
2403
1821
datetime.timedelta(weeks=52)]
2404
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2405
2406
2407
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
1822
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1823
1824
1825
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
2408
1826
command = command.SetApprovalDelay
2409
1827
propname = "ApprovalDelay"
2410
1828
values_to_set = [datetime.timedelta(),
2412
1830
datetime.timedelta(seconds=1),
2413
1831
datetime.timedelta(weeks=1),
2414
1832
datetime.timedelta(weeks=52)]
2415
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2416
2417
2418
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
1833
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1834
1835
1836
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
2419
1837
command = command.SetApprovalDuration
2420
1838
propname = "ApprovalDuration"
2421
1839
values_to_set = [datetime.timedelta(),
2423
1841
datetime.timedelta(seconds=1),
2424
1842
datetime.timedelta(weeks=1),
2425
1843
datetime.timedelta(weeks=52)]
2426
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1844
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
2427
1845
2428
1846
2429
1847
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0