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 1092
- compare with another revision
- download diff
- download tarball
- view history from revision 1092
- Committer: Teddy Hogeborn
- Date: 2019-03-30 17:02:33 UTC
- Revision ID: teddy@recompile.se-20190330170233-nxu3cgb98q2g3o5j
Fix typo in intro(8mandos).
* intro.xml (INTRODUCTION): Add missing period at end of last sentence
of penultimate paragraph.
* intro.xml (INTRODUCTION): Add missing period at end of last sentence
of penultimate paragraph.
- files modified:
- mandos-ctl
added
removed
mandos-ctl
45
45
import io
46
46
import tempfile
47
47
import contextlib
48
import abc
48
49
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
50
import dbus as dbus_python
62
51
63
52
# Show warnings by default
64
53
if not sys.warnoptions:
93
82
if options.debug:
94
83
log.setLevel(logging.DEBUG)
95
84
96
if pydbus is not None:
97
bus = pydbus_adapter.CachingBus(pydbus)
98
else:
99
bus = dbus_python_adapter.CachingBus(dbus_python)
85
bus = dbus_python_adapter.CachingBus(dbus_python)
100
86
101
87
try:
102
88
all_clients = bus.get_clients_and_properties()
136
122
help="Select all clients")
137
123
parser.add_argument("-v", "--verbose", action="store_true",
138
124
help="Print all fields")
139
parser.add_argument("-j", "--dump-json", dest="commands",
140
action="append_const", default=[],
141
const=command.DumpJSON(),
125
parser.add_argument("-j", "--dump-json", action="store_true",
142
126
help="Dump client data in JSON format")
143
127
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(),
128
enable_disable.add_argument("-e", "--enable", action="store_true",
147
129
help="Enable client")
148
enable_disable.add_argument("-d", "--disable", dest="commands",
149
action="append_const", default=[],
150
const=command.Disable(),
130
enable_disable.add_argument("-d", "--disable",
131
action="store_true",
151
132
help="disable client")
152
parser.add_argument("-b", "--bump-timeout", dest="commands",
153
action="append_const", default=[],
154
const=command.BumpTimeout(),
133
parser.add_argument("-b", "--bump-timeout", action="store_true",
155
134
help="Bump timeout for client")
156
135
start_stop_checker = parser.add_mutually_exclusive_group()
157
136
start_stop_checker.add_argument("--start-checker",
158
dest="commands",
159
action="append_const", default=[],
160
const=command.StartChecker(),
137
action="store_true",
161
138
help="Start checker for client")
162
start_stop_checker.add_argument("--stop-checker", dest="commands",
163
action="append_const", default=[],
164
const=command.StopChecker(),
139
start_stop_checker.add_argument("--stop-checker",
140
action="store_true",
165
141
help="Stop checker for client")
166
parser.add_argument("-V", "--is-enabled", dest="commands",
167
action="append_const", default=[],
168
const=command.IsEnabled(),
142
parser.add_argument("-V", "--is-enabled", action="store_true",
169
143
help="Check if client is enabled")
170
parser.add_argument("-r", "--remove", dest="commands",
171
action="append_const", default=[],
172
const=command.Remove(),
144
parser.add_argument("-r", "--remove", action="store_true",
173
145
help="Remove client")
174
parser.add_argument("-c", "--checker", dest="commands",
175
action="append", default=[],
176
metavar="COMMAND", type=command.SetChecker,
146
parser.add_argument("-c", "--checker",
177
147
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")
148
parser.add_argument("-t", "--timeout", type=string_to_delta,
149
help="Set timeout for client")
150
parser.add_argument("--extended-timeout", type=string_to_delta,
151
help="Set extended timeout for client")
152
parser.add_argument("-i", "--interval", type=string_to_delta,
153
help="Set checker interval for client")
193
154
approve_deny_default = parser.add_mutually_exclusive_group()
194
155
approve_deny_default.add_argument(
195
"--approve-by-default", dest="commands",
196
action="append_const", default=[],
197
const=command.ApproveByDefault(),
156
"--approve-by-default", action="store_true",
157
default=None, dest="approved_by_default",
198
158
help="Set client to be approved by default")
199
159
approve_deny_default.add_argument(
200
"--deny-by-default", dest="commands",
201
action="append_const", default=[],
202
const=command.DenyByDefault(),
160
"--deny-by-default", action="store_false",
161
dest="approved_by_default",
203
162
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")
163
parser.add_argument("--approval-delay", type=string_to_delta,
164
help="Set delay before client approve/deny")
165
parser.add_argument("--approval-duration", type=string_to_delta,
166
help="Set duration of one client approval")
167
parser.add_argument("-H", "--host", help="Set host for client")
168
parser.add_argument("-s", "--secret",
169
type=argparse.FileType(mode="rb"),
170
help="Set password blob (file) for client")
223
171
approve_deny = parser.add_mutually_exclusive_group()
224
172
approve_deny.add_argument(
225
"-A", "--approve", dest="commands", action="append_const",
226
default=[], const=command.Approve(),
173
"-A", "--approve", action="store_true",
227
174
help="Approve any current client request")
228
approve_deny.add_argument("-D", "--deny", dest="commands",
229
action="append_const", default=[],
230
const=command.Deny(),
175
approve_deny.add_argument("-D", "--deny", action="store_true",
231
176
help="Deny any current client request")
232
177
parser.add_argument("--debug", action="store_true",
233
178
help="Debug mode (show D-Bus commands)")
424
369
"""Apply additional restrictions on options, not expressible in
425
370
argparse"""
426
371
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)
372
def has_actions(options):
373
return any((options.enable,
374
options.disable,
375
options.bump_timeout,
376
options.start_checker,
377
options.stop_checker,
378
options.is_enabled,
379
options.remove,
380
options.checker is not None,
381
options.timeout is not None,
382
options.extended_timeout is not None,
383
options.interval is not None,
384
options.approved_by_default is not None,
385
options.approval_delay is not None,
386
options.approval_duration is not None,
387
options.host is not None,
388
options.secret is not None,
389
options.approve,
390
options.deny))
450
391
451
if has_commands(options) and not (options.client or options.all):
392
if has_actions(options) and not (options.client or options.all):
452
393
parser.error("Options require clients names or --all.")
453
if options.verbose and has_commands(options):
394
if options.verbose and has_actions(options):
454
395
parser.error("--verbose can only be used alone.")
455
if (has_commands(options, (command.DumpJSON,))
456
and (options.verbose or len(options.commands) > 1)):
396
if options.dump_json and (options.verbose
397
or has_actions(options)):
457
398
parser.error("--dump-json can only be used alone.")
458
if options.all and not has_commands(options):
399
if options.all and not has_actions(options):
459
400
parser.error("--all requires an action.")
460
if (has_commands(options, (command.IsEnabled,))
461
and len(options.client) > 1):
401
if options.is_enabled and len(options.client) > 1:
462
402
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")
403
if options.remove:
404
options.remove = False
405
if has_actions(options) and not options.deny:
406
parser.error("--remove can only be combined with --deny")
407
options.remove = True
408
467
409
468
410
469
411
class dbus(object):
607
549
return new_object
608
550
609
551
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
671
672
673
552
def commands_from_options(options):
674
553
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)
554
commands = []
555
556
if options.is_enabled:
557
commands.append(command.IsEnabled())
558
559
if options.approve:
560
commands.append(command.Approve())
561
562
if options.deny:
563
commands.append(command.Deny())
564
565
if options.remove:
566
commands.append(command.Remove())
567
568
if options.dump_json:
569
commands.append(command.DumpJSON())
570
571
if options.enable:
572
commands.append(command.Enable())
573
574
if options.disable:
575
commands.append(command.Disable())
576
577
if options.bump_timeout:
578
commands.append(command.BumpTimeout())
579
580
if options.start_checker:
581
commands.append(command.StartChecker())
582
583
if options.stop_checker:
584
commands.append(command.StopChecker())
585
586
if options.approved_by_default is not None:
587
if options.approved_by_default:
588
commands.append(command.ApproveByDefault())
693
589
else:
694
cleaned_after.append(cmd)
695
if cleaned_after != after_remove:
696
commands = before_remove + cleaned_after
590
commands.append(command.DenyByDefault())
591
592
if options.checker is not None:
593
commands.append(command.SetChecker(options.checker))
594
595
if options.host is not None:
596
commands.append(command.SetHost(options.host))
597
598
if options.secret is not None:
599
commands.append(command.SetSecret(options.secret))
600
601
if options.timeout is not None:
602
commands.append(command.SetTimeout(options.timeout))
603
604
if options.extended_timeout:
605
commands.append(
606
command.SetExtendedTimeout(options.extended_timeout))
607
608
if options.interval is not None:
609
commands.append(command.SetInterval(options.interval))
610
611
if options.approval_delay is not None:
612
commands.append(
613
command.SetApprovalDelay(options.approval_delay))
614
615
if options.approval_duration is not None:
616
commands.append(
617
command.SetApprovalDuration(options.approval_duration))
697
618
698
619
# If no command option has been given, show table of clients,
699
620
# optionally verbosely
914
835
def __init__(self, value):
915
836
self.value_to_set = value
916
837
917
@classmethod
918
def argparse(cls, argtype):
919
def cmdtype(arg):
920
return cls(argtype(arg))
921
return cmdtype
922
838
923
839
class SetChecker(PropertySetterValue):
924
840
propname = "Checker"
1050
966
1051
967
def test_actions_requires_client_or_all(self):
1052
968
for action, value in self.actions.items():
1053
args = self.actionargs(action, value)
969
options = self.parser.parse_args()
970
setattr(options, action, value)
1054
971
with self.assertParseError():
1055
self.parse_args(args)
972
self.check_option_syntax(options)
1056
973
1057
# This mostly corresponds to the definition from has_commands() in
974
# This mostly corresponds to the definition from has_actions() in
1058
975
# check_option_syntax()
1059
976
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,
977
# The actual values set here are not that important, but we do
978
# at least stick to the correct types, even though they are
979
# never used
980
"enable": True,
981
"disable": True,
982
"bump_timeout": True,
983
"start_checker": True,
984
"stop_checker": True,
985
"is_enabled": True,
986
"remove": True,
987
"checker": "x",
988
"timeout": datetime.timedelta(),
989
"extended_timeout": datetime.timedelta(),
990
"interval": datetime.timedelta(),
991
"approved_by_default": True,
992
"approval_delay": datetime.timedelta(),
993
"approval_duration": datetime.timedelta(),
994
"host": "x",
995
"secret": io.BytesIO(b"x"),
996
"approve": True,
997
"deny": True,
1079
998
}
1080
999
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
1000
@contextlib.contextmanager
1089
1001
def assertParseError(self):
1090
1002
with self.assertRaises(SystemExit) as e:
1095
1007
# /argparse.html#exiting-methods
1096
1008
self.assertEqual(2, e.exception.code)
1097
1009
1098
def parse_args(self, args):
1099
options = self.parser.parse_args(args)
1100
check_option_syntax(self.parser, options)
1101
1102
1010
@staticmethod
1103
1011
@contextlib.contextmanager
1104
1012
def redirect_stderr_to_devnull():
1115
1023
1116
1024
def test_actions_all_conflicts_with_verbose(self):
1117
1025
for action, value in self.actions.items():
1118
args = self.actionargs(action, value, "--all",
1119
"--verbose")
1026
options = self.parser.parse_args()
1027
setattr(options, action, value)
1028
options.all = True
1029
options.verbose = True
1120
1030
with self.assertParseError():
1121
self.parse_args(args)
1031
self.check_option_syntax(options)
1122
1032
1123
1033
def test_actions_with_client_conflicts_with_verbose(self):
1124
1034
for action, value in self.actions.items():
1125
args = self.actionargs(action, value, "--verbose",
1126
"client")
1035
options = self.parser.parse_args()
1036
setattr(options, action, value)
1037
options.verbose = True
1038
options.client = ["client"]
1127
1039
with self.assertParseError():
1128
self.parse_args(args)
1040
self.check_option_syntax(options)
1129
1041
1130
1042
def test_dump_json_conflicts_with_verbose(self):
1131
args = ["--dump-json", "--verbose"]
1043
options = self.parser.parse_args()
1044
options.dump_json = True
1045
options.verbose = True
1132
1046
with self.assertParseError():
1133
self.parse_args(args)
1047
self.check_option_syntax(options)
1134
1048
1135
1049
def test_dump_json_conflicts_with_action(self):
1136
1050
for action, value in self.actions.items():
1137
args = self.actionargs(action, value, "--dump-json")
1051
options = self.parser.parse_args()
1052
setattr(options, action, value)
1053
options.dump_json = True
1138
1054
with self.assertParseError():
1139
self.parse_args(args)
1055
self.check_option_syntax(options)
1140
1056
1141
1057
def test_all_can_not_be_alone(self):
1142
args = ["--all"]
1058
options = self.parser.parse_args()
1059
options.all = True
1143
1060
with self.assertParseError():
1144
self.parse_args(args)
1061
self.check_option_syntax(options)
1145
1062
1146
1063
def test_all_is_ok_with_any_action(self):
1147
1064
for action, value in self.actions.items():
1148
args = self.actionargs(action, value, "--all")
1149
self.parse_args(args)
1065
options = self.parser.parse_args()
1066
setattr(options, action, value)
1067
options.all = True
1068
self.check_option_syntax(options)
1150
1069
1151
1070
def test_any_action_is_ok_with_one_client(self):
1152
1071
for action, value in self.actions.items():
1153
args = self.actionargs(action, value, "client")
1154
self.parse_args(args)
1072
options = self.parser.parse_args()
1073
setattr(options, action, value)
1074
options.client = ["client"]
1075
self.check_option_syntax(options)
1155
1076
1156
1077
def test_one_client_with_all_actions_except_is_enabled(self):
1078
options = self.parser.parse_args()
1157
1079
for action, value in self.actions.items():
1158
if action == "--is-enabled":
1080
if action == "is_enabled":
1159
1081
continue
1160
args = self.actionargs(action, value, "client")
1161
self.parse_args(args)
1082
setattr(options, action, value)
1083
options.client = ["client"]
1084
self.check_option_syntax(options)
1162
1085
1163
1086
def test_two_clients_with_all_actions_except_is_enabled(self):
1087
options = self.parser.parse_args()
1164
1088
for action, value in self.actions.items():
1165
if action == "--is-enabled":
1089
if action == "is_enabled":
1166
1090
continue
1167
args = self.actionargs(action, value, "client1",
1168
"client2")
1169
self.parse_args(args)
1091
setattr(options, action, value)
1092
options.client = ["client1", "client2"]
1093
self.check_option_syntax(options)
1170
1094
1171
1095
def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1172
1096
for action, value in self.actions.items():
1173
if action == "--is-enabled":
1097
if action == "is_enabled":
1174
1098
continue
1175
args = self.actionargs(action, value, "client1",
1176
"client2")
1177
self.parse_args(args)
1099
options = self.parser.parse_args()
1100
setattr(options, action, value)
1101
options.client = ["client1", "client2"]
1102
self.check_option_syntax(options)
1178
1103
1179
1104
def test_is_enabled_fails_without_client(self):
1180
args = ["--is-enabled"]
1105
options = self.parser.parse_args()
1106
options.is_enabled = True
1181
1107
with self.assertParseError():
1182
self.parse_args(args)
1108
self.check_option_syntax(options)
1183
1109
1184
1110
def test_is_enabled_fails_with_two_clients(self):
1185
args = ["--is-enabled", "client1", "client2"]
1111
options = self.parser.parse_args()
1112
options.is_enabled = True
1113
options.client = ["client1", "client2"]
1186
1114
with self.assertParseError():
1187
self.parse_args(args)
1115
self.check_option_syntax(options)
1188
1116
1189
1117
def test_remove_can_only_be_combined_with_action_deny(self):
1190
1118
for action, value in self.actions.items():
1191
if action in {"--remove", "--deny"}:
1119
if action in {"remove", "deny"}:
1192
1120
continue
1193
args = self.actionargs(action, value, "--all",
1194
"--remove")
1121
options = self.parser.parse_args()
1122
setattr(options, action, value)
1123
options.all = True
1124
options.remove = True
1195
1125
with self.assertParseError():
1196
self.parse_args(args)
1126
self.check_option_syntax(options)
1197
1127
1198
1128
1199
1129
class Test_dbus_exceptions(unittest.TestCase):
1586
1516
self.assertIs(obj1, obj1b)
1587
1517
1588
1518
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)
1736
1737
1738
1519
class Test_commands_from_options(unittest.TestCase):
1739
1520
1740
1521
def setUp(self):
2289
2070
def runTest(self):
2290
2071
if not hasattr(self, "command"):
2291
2072
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])
2073
values_to_get = getattr(self, "values_to_get",
2074
self.values_to_set)
2075
for value_to_set, value_to_get in zip(self.values_to_set,
2076
values_to_get):
2077
for clientpath in self.bus.clients:
2078
self.bus.clients[clientpath][self.propname] = Unique()
2079
self.run_command(value_to_set, self.bus.clients)
2080
for clientpath in self.bus.clients:
2081
value = self.bus.clients[clientpath][self.propname]
2308
2082
self.assertNotIsInstance(value, Unique)
2309
2083
self.assertEqual(value_to_get, value)
2310
2084
2085
def run_command(self, value, clients):
2086
self.command().run(clients, self.bus)
2087
2311
2088
2312
2089
class TestEnableCmd(TestPropertySetterCmd):
2313
2090
command = command.Enable
2314
2091
propname = "Enabled"
2315
values_to_get = [True]
2092
values_to_set = [True]
2316
2093
2317
2094
2318
2095
class TestDisableCmd(TestPropertySetterCmd):
2319
2096
command = command.Disable
2320
2097
propname = "Enabled"
2321
values_to_get = [False]
2098
values_to_set = [False]
2322
2099
2323
2100
2324
2101
class TestBumpTimeoutCmd(TestPropertySetterCmd):
2325
2102
command = command.BumpTimeout
2326
2103
propname = "LastCheckedOK"
2327
values_to_get = [""]
2104
values_to_set = [""]
2328
2105
2329
2106
2330
2107
class TestStartCheckerCmd(TestPropertySetterCmd):
2331
2108
command = command.StartChecker
2332
2109
propname = "CheckerRunning"
2333
values_to_get = [True]
2110
values_to_set = [True]
2334
2111
2335
2112
2336
2113
class TestStopCheckerCmd(TestPropertySetterCmd):
2337
2114
command = command.StopChecker
2338
2115
propname = "CheckerRunning"
2339
values_to_get = [False]
2116
values_to_set = [False]
2340
2117
2341
2118
2342
2119
class TestApproveByDefaultCmd(TestPropertySetterCmd):
2343
2120
command = command.ApproveByDefault
2344
2121
propname = "ApprovedByDefault"
2345
values_to_get = [True]
2122
values_to_set = [True]
2346
2123
2347
2124
2348
2125
class TestDenyByDefaultCmd(TestPropertySetterCmd):
2349
2126
command = command.DenyByDefault
2350
2127
propname = "ApprovedByDefault"
2351
values_to_get = [False]
2352
2353
2354
class TestSetCheckerCmd(TestPropertySetterCmd):
2128
values_to_set = [False]
2129
2130
2131
class TestPropertySetterValueCmd(TestPropertySetterCmd):
2132
"""Abstract class for tests of PropertySetterValueCmd classes"""
2133
2134
def run_command(self, value, clients):
2135
self.command(value).run(clients, self.bus)
2136
2137
2138
class TestSetCheckerCmd(TestPropertySetterValueCmd):
2355
2139
command = command.SetChecker
2356
2140
propname = "Checker"
2357
2141
values_to_set = ["", ":", "fping -q -- %s"]
2358
2142
2359
2143
2360
class TestSetHostCmd(TestPropertySetterCmd):
2144
class TestSetHostCmd(TestPropertySetterValueCmd):
2361
2145
command = command.SetHost
2362
2146
propname = "Host"
2363
2147
values_to_set = ["192.0.2.3", "client.example.org"]
2364
2148
2365
2149
2366
class TestSetSecretCmd(TestPropertySetterCmd):
2150
class TestSetSecretCmd(TestPropertySetterValueCmd):
2367
2151
command = command.SetSecret
2368
2152
propname = "Secret"
2369
2153
values_to_set = [io.BytesIO(b""),
2371
2155
values_to_get = [f.getvalue() for f in values_to_set]
2372
2156
2373
2157
2374
class TestSetTimeoutCmd(TestPropertySetterCmd):
2158
class TestSetTimeoutCmd(TestPropertySetterValueCmd):
2375
2159
command = command.SetTimeout
2376
2160
propname = "Timeout"
2377
2161
values_to_set = [datetime.timedelta(),
2382
2166
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2383
2167
2384
2168
2385
class TestSetExtendedTimeoutCmd(TestPropertySetterCmd):
2169
class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):
2386
2170
command = command.SetExtendedTimeout
2387
2171
propname = "ExtendedTimeout"
2388
2172
values_to_set = [datetime.timedelta(),
2393
2177
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2394
2178
2395
2179
2396
class TestSetIntervalCmd(TestPropertySetterCmd):
2180
class TestSetIntervalCmd(TestPropertySetterValueCmd):
2397
2181
command = command.SetInterval
2398
2182
propname = "Interval"
2399
2183
values_to_set = [datetime.timedelta(),
2404
2188
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2405
2189
2406
2190
2407
class TestSetApprovalDelayCmd(TestPropertySetterCmd):
2191
class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):
2408
2192
command = command.SetApprovalDelay
2409
2193
propname = "ApprovalDelay"
2410
2194
values_to_set = [datetime.timedelta(),
2415
2199
values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
2416
2200
2417
2201
2418
class TestSetApprovalDurationCmd(TestPropertySetterCmd):
2202
class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):
2419
2203
command = command.SetApprovalDuration
2420
2204
propname = "ApprovalDuration"
2421
2205
values_to_set = [datetime.timedelta(),
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0