/mandos/trunk

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

« back to all changes in this revision

Viewing changes to mandos-ctl

  • Committer: Teddy Hogeborn
  • Date: 2019-03-08 23:55:34 UTC
  • Revision ID: teddy@recompile.se-20190308235534-kkiwc3oeyp0vxw5h
mandos-ctl: Add new --debug option to show D-Bus calls

* mandos-ctl (PropertyCmd, RemoveCmd, ApproveCmd, DenyCmd): Add debug
  output
  (add_command_line_options): Also add "--debug" options.
  (main): If debug option is set, set logging level to DEBUG.  Also
          add debug output when connecting to Mandos server.
* mandos-ctl.xml (SYNOPSIS): Show --debug option where appropriate.
  (OPTIONS): Document new "--debug" option.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
42
42
import json
43
43
import unittest
44
44
import logging
 
45
import io
 
46
import tempfile
 
47
import contextlib
45
48
 
46
49
import dbus
47
50
 
297
300
    """Abstract class for Actions for setting one client property"""
298
301
    def run_on_one_client(self, client, properties):
299
302
        """Set the Client's D-Bus property"""
 
303
        log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
 
304
                  client.__dbus_object_path__,
 
305
                  dbus.PROPERTIES_IFACE, client_interface,
 
306
                  self.property, self.value_to_set
 
307
                  if not isinstance(self.value_to_set, dbus.Boolean)
 
308
                  else bool(self.value_to_set))
300
309
        client.Set(client_interface, self.property, self.value_to_set,
301
310
                   dbus_interface=dbus.PROPERTIES_IFACE)
302
311
 
314
323
    @value_to_set.setter
315
324
    def value_to_set(self, value):
316
325
        """When setting, convert value to a datetime.timedelta"""
317
 
        self._vts = string_to_delta(value).total_seconds() * 1000
 
326
        self._vts = int(round(value.total_seconds() * 1000))
318
327
 
319
328
# Actual (non-abstract) command classes
320
329
 
428
437
 
429
438
class RemoveCmd(Command):
430
439
    def run_on_one_client(self, client, properties):
 
440
        log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", busname,
 
441
                  server_path, server_interface,
 
442
                  str(client.__dbus_object_path__))
431
443
        self.mandos.RemoveClient(client.__dbus_object_path__)
432
444
 
433
445
class ApproveCmd(Command):
434
446
    def run_on_one_client(self, client, properties):
 
447
        log.debug("D-Bus: %s:%s.Approve(True)",
 
448
                  client.__dbus_object_path__, client_interface)
435
449
        client.Approve(dbus.Boolean(True),
436
450
                       dbus_interface=client_interface)
437
451
 
438
452
class DenyCmd(Command):
439
453
    def run_on_one_client(self, client, properties):
 
454
        log.debug("D-Bus: %s:%s.Approve(False)",
 
455
                  client.__dbus_object_path__, client_interface)
440
456
        client.Approve(dbus.Boolean(False),
441
457
                       dbus_interface=client_interface)
442
458
 
475
491
    property = "Host"
476
492
 
477
493
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
 
494
    @property
 
495
    def value_to_set(self):
 
496
        return self._vts
 
497
    @value_to_set.setter
 
498
    def value_to_set(self, value):
 
499
        """When setting, read data from supplied file object"""
 
500
        self._vts = value.read()
 
501
        value.close()
478
502
    property = "Secret"
479
503
 
480
504
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
495
519
                             MillisecondsValueArgumentMixIn):
496
520
    property = "ApprovalDuration"
497
521
 
498
 
def has_actions(options):
499
 
    return any((options.enable,
500
 
                options.disable,
501
 
                options.bump_timeout,
502
 
                options.start_checker,
503
 
                options.stop_checker,
504
 
                options.is_enabled,
505
 
                options.remove,
506
 
                options.checker is not None,
507
 
                options.timeout is not None,
508
 
                options.extended_timeout is not None,
509
 
                options.interval is not None,
510
 
                options.approved_by_default is not None,
511
 
                options.approval_delay is not None,
512
 
                options.approval_duration is not None,
513
 
                options.host is not None,
514
 
                options.secret is not None,
515
 
                options.approve,
516
 
                options.deny))
517
 
 
518
522
def add_command_line_options(parser):
519
523
    parser.add_argument("--version", action="version",
520
524
                        version="%(prog)s {}".format(version),
546
550
                        help="Remove client")
547
551
    parser.add_argument("-c", "--checker",
548
552
                        help="Set checker command for client")
549
 
    parser.add_argument("-t", "--timeout",
 
553
    parser.add_argument("-t", "--timeout", type=string_to_delta,
550
554
                        help="Set timeout for client")
551
 
    parser.add_argument("--extended-timeout",
 
555
    parser.add_argument("--extended-timeout", type=string_to_delta,
552
556
                        help="Set extended timeout for client")
553
 
    parser.add_argument("-i", "--interval",
 
557
    parser.add_argument("-i", "--interval", type=string_to_delta,
554
558
                        help="Set checker interval for client")
555
559
    approve_deny_default = parser.add_mutually_exclusive_group()
556
560
    approve_deny_default.add_argument(
561
565
        "--deny-by-default", action="store_false",
562
566
        dest="approved_by_default",
563
567
        help="Set client to be denied by default")
564
 
    parser.add_argument("--approval-delay",
 
568
    parser.add_argument("--approval-delay", type=string_to_delta,
565
569
                        help="Set delay before client approve/deny")
566
 
    parser.add_argument("--approval-duration",
 
570
    parser.add_argument("--approval-duration", type=string_to_delta,
567
571
                        help="Set duration of one client approval")
568
572
    parser.add_argument("-H", "--host", help="Set host for client")
569
573
    parser.add_argument("-s", "--secret",
575
579
        help="Approve any current client request")
576
580
    approve_deny.add_argument("-D", "--deny", action="store_true",
577
581
                              help="Deny any current client request")
 
582
    parser.add_argument("--debug", action="store_true",
 
583
                        help="Debug mode (show D-Bus commands)")
578
584
    parser.add_argument("--check", action="store_true",
579
585
                        help="Run self-test")
580
586
    parser.add_argument("client", nargs="*", help="Client name")
609
615
        commands.append(RemoveCmd())
610
616
 
611
617
    if options.checker is not None:
612
 
        commands.append(SetCheckerCmd())
 
618
        commands.append(SetCheckerCmd(options.checker))
613
619
 
614
620
    if options.timeout is not None:
615
621
        commands.append(SetTimeoutCmd(options.timeout))
619
625
            SetExtendedTimeoutCmd(options.extended_timeout))
620
626
 
621
627
    if options.interval is not None:
622
 
        command.append(SetIntervalCmd(options.interval))
 
628
        commands.append(SetIntervalCmd(options.interval))
623
629
 
624
630
    if options.approved_by_default is not None:
625
631
        if options.approved_by_default:
626
 
            command.append(ApproveByDefaultCmd())
 
632
            commands.append(ApproveByDefaultCmd())
627
633
        else:
628
 
            command.append(DenyByDefaultCmd())
 
634
            commands.append(DenyByDefaultCmd())
629
635
 
630
636
    if options.approval_delay is not None:
631
 
        command.append(SetApprovalDelayCmd(options.approval_delay))
 
637
        commands.append(SetApprovalDelayCmd(options.approval_delay))
632
638
 
633
639
    if options.approval_duration is not None:
634
 
        command.append(
 
640
        commands.append(
635
641
            SetApprovalDurationCmd(options.approval_duration))
636
642
 
637
643
    if options.host is not None:
638
 
        command.append(SetHostCmd(options.host))
 
644
        commands.append(SetHostCmd(options.host))
639
645
 
640
646
    if options.secret is not None:
641
 
        command.append(SetSecretCmd(options.secret))
 
647
        commands.append(SetSecretCmd(options.secret))
642
648
 
643
649
    if options.approve:
644
650
        commands.append(ApproveCmd())
654
660
    return commands
655
661
 
656
662
 
657
 
def main():
658
 
    parser = argparse.ArgumentParser()
659
 
 
660
 
    add_command_line_options(parser)
661
 
 
662
 
    options = parser.parse_args()
 
663
def check_option_syntax(parser, options):
 
664
    """Apply additional restrictions on options, not expressible in
 
665
argparse"""
 
666
 
 
667
    def has_actions(options):
 
668
        return any((options.enable,
 
669
                    options.disable,
 
670
                    options.bump_timeout,
 
671
                    options.start_checker,
 
672
                    options.stop_checker,
 
673
                    options.is_enabled,
 
674
                    options.remove,
 
675
                    options.checker is not None,
 
676
                    options.timeout is not None,
 
677
                    options.extended_timeout is not None,
 
678
                    options.interval is not None,
 
679
                    options.approved_by_default is not None,
 
680
                    options.approval_delay is not None,
 
681
                    options.approval_duration is not None,
 
682
                    options.host is not None,
 
683
                    options.secret is not None,
 
684
                    options.approve,
 
685
                    options.deny))
663
686
 
664
687
    if has_actions(options) and not (options.client or options.all):
665
688
        parser.error("Options require clients names or --all.")
673
696
    if options.is_enabled and len(options.client) > 1:
674
697
        parser.error("--is-enabled requires exactly one client")
675
698
 
 
699
 
 
700
def main():
 
701
    parser = argparse.ArgumentParser()
 
702
 
 
703
    add_command_line_options(parser)
 
704
 
 
705
    options = parser.parse_args()
 
706
 
 
707
    check_option_syntax(parser, options)
 
708
 
676
709
    clientnames = options.client
677
710
 
 
711
    if options.debug:
 
712
        log.setLevel(logging.DEBUG)
 
713
 
678
714
    try:
679
715
        bus = dbus.SystemBus()
 
716
        log.debug("D-Bus: Connect to: (name=%r, path=%r)", busname,
 
717
                  server_path)
680
718
        mandos_dbus_objc = bus.get_object(busname, server_path)
681
719
    except dbus.exceptions.DBusException:
682
720
        log.critical("Could not connect to Mandos server")
695
733
    dbus_filter = NullFilter()
696
734
    try:
697
735
        dbus_logger.addFilter(dbus_filter)
 
736
        log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", busname,
 
737
                  server_path, dbus.OBJECT_MANAGER_IFACE)
698
738
        mandos_clients = {path: ifs_and_props[client_interface]
699
739
                          for path, ifs_and_props in
700
740
                          mandos_serv_object_manager
1083
1123
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1084
1124
    command = SetSecretCmd
1085
1125
    property = "Secret"
1086
 
    values_to_set = [b"", b"secret"]
 
1126
    values_to_set = [io.BytesIO(b""),
 
1127
                     io.BytesIO(b"secret\0xyzzy\nbar")]
 
1128
    values_to_get = [b"", b"secret\0xyzzy\nbar"]
1087
1129
 
1088
1130
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1089
1131
    command = SetTimeoutCmd
1090
1132
    property = "Timeout"
1091
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1092
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
 
1133
    values_to_set = [datetime.timedelta(),
 
1134
                     datetime.timedelta(minutes=5),
 
1135
                     datetime.timedelta(seconds=1),
 
1136
                     datetime.timedelta(weeks=1),
 
1137
                     datetime.timedelta(weeks=52)]
 
1138
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1093
1139
 
1094
1140
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1095
1141
    command = SetExtendedTimeoutCmd
1096
1142
    property = "ExtendedTimeout"
1097
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1098
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
 
1143
    values_to_set = [datetime.timedelta(),
 
1144
                     datetime.timedelta(minutes=5),
 
1145
                     datetime.timedelta(seconds=1),
 
1146
                     datetime.timedelta(weeks=1),
 
1147
                     datetime.timedelta(weeks=52)]
 
1148
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1099
1149
 
1100
1150
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1101
1151
    command = SetIntervalCmd
1102
1152
    property = "Interval"
1103
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1104
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
 
1153
    values_to_set = [datetime.timedelta(),
 
1154
                     datetime.timedelta(minutes=5),
 
1155
                     datetime.timedelta(seconds=1),
 
1156
                     datetime.timedelta(weeks=1),
 
1157
                     datetime.timedelta(weeks=52)]
 
1158
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1105
1159
 
1106
1160
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1107
1161
    command = SetApprovalDelayCmd
1108
1162
    property = "ApprovalDelay"
1109
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1110
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
 
1163
    values_to_set = [datetime.timedelta(),
 
1164
                     datetime.timedelta(minutes=5),
 
1165
                     datetime.timedelta(seconds=1),
 
1166
                     datetime.timedelta(weeks=1),
 
1167
                     datetime.timedelta(weeks=52)]
 
1168
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1111
1169
 
1112
1170
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1113
1171
    command = SetApprovalDurationCmd
1114
1172
    property = "ApprovalDuration"
1115
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1116
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
 
1173
    values_to_set = [datetime.timedelta(),
 
1174
                     datetime.timedelta(minutes=5),
 
1175
                     datetime.timedelta(seconds=1),
 
1176
                     datetime.timedelta(weeks=1),
 
1177
                     datetime.timedelta(weeks=52)]
 
1178
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1117
1179
 
1118
 
class TestOptions(unittest.TestCase):
 
1180
class Test_command_from_options(unittest.TestCase):
1119
1181
    def setUp(self):
1120
1182
        self.parser = argparse.ArgumentParser()
1121
1183
        add_command_line_options(self.parser)
1123
1185
        """Assert that parsing ARGS should result in an instance of
1124
1186
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1125
1187
        options = self.parser.parse_args(args)
 
1188
        check_option_syntax(self.parser, options)
1126
1189
        commands = commands_from_options(options)
1127
1190
        self.assertEqual(len(commands), 1)
1128
1191
        command = commands[0]
1129
1192
        self.assertIsInstance(command, command_cls)
1130
1193
        for key, value in cmd_attrs.items():
1131
1194
            self.assertEqual(getattr(command, key), value)
1132
 
    def test_default_is_show_table(self):
 
1195
    def test_print_table(self):
1133
1196
        self.assert_command_from_args([], PrintTableCmd,
1134
1197
                                      verbose=False)
1135
 
    def test_show_table_verbose(self):
 
1198
 
 
1199
    def test_print_table_verbose(self):
1136
1200
        self.assert_command_from_args(["--verbose"], PrintTableCmd,
1137
1201
                                      verbose=True)
 
1202
 
 
1203
    def test_print_table_verbose_short(self):
 
1204
        self.assert_command_from_args(["-v"], PrintTableCmd,
 
1205
                                      verbose=True)
 
1206
 
1138
1207
    def test_enable(self):
1139
1208
        self.assert_command_from_args(["--enable", "foo"], EnableCmd)
 
1209
 
 
1210
    def test_enable_short(self):
 
1211
        self.assert_command_from_args(["-e", "foo"], EnableCmd)
 
1212
 
1140
1213
    def test_disable(self):
1141
1214
        self.assert_command_from_args(["--disable", "foo"],
1142
1215
                                      DisableCmd)
1143
1216
 
 
1217
    def test_disable_short(self):
 
1218
        self.assert_command_from_args(["-d", "foo"], DisableCmd)
 
1219
 
 
1220
    def test_bump_timeout(self):
 
1221
        self.assert_command_from_args(["--bump-timeout", "foo"],
 
1222
                                      BumpTimeoutCmd)
 
1223
 
 
1224
    def test_bump_timeout_short(self):
 
1225
        self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
 
1226
 
 
1227
    def test_start_checker(self):
 
1228
        self.assert_command_from_args(["--start-checker", "foo"],
 
1229
                                      StartCheckerCmd)
 
1230
 
 
1231
    def test_stop_checker(self):
 
1232
        self.assert_command_from_args(["--stop-checker", "foo"],
 
1233
                                      StopCheckerCmd)
 
1234
 
 
1235
    def test_remove(self):
 
1236
        self.assert_command_from_args(["--remove", "foo"],
 
1237
                                      RemoveCmd)
 
1238
 
 
1239
    def test_remove_short(self):
 
1240
        self.assert_command_from_args(["-r", "foo"], RemoveCmd)
 
1241
 
 
1242
    def test_checker(self):
 
1243
        self.assert_command_from_args(["--checker", ":", "foo"],
 
1244
                                      SetCheckerCmd, value_to_set=":")
 
1245
 
 
1246
    def test_checker_empty(self):
 
1247
        self.assert_command_from_args(["--checker", "", "foo"],
 
1248
                                      SetCheckerCmd, value_to_set="")
 
1249
 
 
1250
    def test_checker_short(self):
 
1251
        self.assert_command_from_args(["-c", ":", "foo"],
 
1252
                                      SetCheckerCmd, value_to_set=":")
 
1253
 
 
1254
    def test_timeout(self):
 
1255
        self.assert_command_from_args(["--timeout", "PT5M", "foo"],
 
1256
                                      SetTimeoutCmd,
 
1257
                                      value_to_set=300000)
 
1258
 
 
1259
    def test_timeout_short(self):
 
1260
        self.assert_command_from_args(["-t", "PT5M", "foo"],
 
1261
                                      SetTimeoutCmd,
 
1262
                                      value_to_set=300000)
 
1263
 
 
1264
    def test_extended_timeout(self):
 
1265
        self.assert_command_from_args(["--extended-timeout", "PT15M",
 
1266
                                       "foo"],
 
1267
                                      SetExtendedTimeoutCmd,
 
1268
                                      value_to_set=900000)
 
1269
 
 
1270
    def test_interval(self):
 
1271
        self.assert_command_from_args(["--interval", "PT2M", "foo"],
 
1272
                                      SetIntervalCmd,
 
1273
                                      value_to_set=120000)
 
1274
 
 
1275
    def test_interval_short(self):
 
1276
        self.assert_command_from_args(["-i", "PT2M", "foo"],
 
1277
                                      SetIntervalCmd,
 
1278
                                      value_to_set=120000)
 
1279
 
 
1280
    def test_approve_by_default(self):
 
1281
        self.assert_command_from_args(["--approve-by-default", "foo"],
 
1282
                                      ApproveByDefaultCmd)
 
1283
 
 
1284
    def test_deny_by_default(self):
 
1285
        self.assert_command_from_args(["--deny-by-default", "foo"],
 
1286
                                      DenyByDefaultCmd)
 
1287
 
 
1288
    def test_approval_delay(self):
 
1289
        self.assert_command_from_args(["--approval-delay", "PT30S",
 
1290
                                       "foo"], SetApprovalDelayCmd,
 
1291
                                      value_to_set=30000)
 
1292
 
 
1293
    def test_approval_duration(self):
 
1294
        self.assert_command_from_args(["--approval-duration", "PT1S",
 
1295
                                       "foo"], SetApprovalDurationCmd,
 
1296
                                      value_to_set=1000)
 
1297
 
 
1298
    def test_host(self):
 
1299
        self.assert_command_from_args(["--host", "foo.example.org",
 
1300
                                       "foo"], SetHostCmd,
 
1301
                                      value_to_set="foo.example.org")
 
1302
 
 
1303
    def test_host_short(self):
 
1304
        self.assert_command_from_args(["-H", "foo.example.org",
 
1305
                                       "foo"], SetHostCmd,
 
1306
                                      value_to_set="foo.example.org")
 
1307
 
 
1308
    def test_secret_devnull(self):
 
1309
        self.assert_command_from_args(["--secret", os.path.devnull,
 
1310
                                       "foo"], SetSecretCmd,
 
1311
                                      value_to_set=b"")
 
1312
 
 
1313
    def test_secret_tempfile(self):
 
1314
        with tempfile.NamedTemporaryFile(mode="r+b") as f:
 
1315
            value = b"secret\0xyzzy\nbar"
 
1316
            f.write(value)
 
1317
            f.seek(0)
 
1318
            self.assert_command_from_args(["--secret", f.name,
 
1319
                                           "foo"], SetSecretCmd,
 
1320
                                          value_to_set=value)
 
1321
 
 
1322
    def test_secret_devnull_short(self):
 
1323
        self.assert_command_from_args(["-s", os.path.devnull, "foo"],
 
1324
                                      SetSecretCmd, value_to_set=b"")
 
1325
 
 
1326
    def test_secret_tempfile_short(self):
 
1327
        with tempfile.NamedTemporaryFile(mode="r+b") as f:
 
1328
            value = b"secret\0xyzzy\nbar"
 
1329
            f.write(value)
 
1330
            f.seek(0)
 
1331
            self.assert_command_from_args(["-s", f.name, "foo"],
 
1332
                                          SetSecretCmd,
 
1333
                                          value_to_set=value)
 
1334
 
 
1335
    def test_approve(self):
 
1336
        self.assert_command_from_args(["--approve", "foo"],
 
1337
                                      ApproveCmd)
 
1338
 
 
1339
    def test_approve_short(self):
 
1340
        self.assert_command_from_args(["-A", "foo"], ApproveCmd)
 
1341
 
 
1342
    def test_deny(self):
 
1343
        self.assert_command_from_args(["--deny", "foo"], DenyCmd)
 
1344
 
 
1345
    def test_deny_short(self):
 
1346
        self.assert_command_from_args(["-D", "foo"], DenyCmd)
 
1347
 
 
1348
    def test_dump_json(self):
 
1349
        self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
 
1350
 
 
1351
    def test_is_enabled(self):
 
1352
        self.assert_command_from_args(["--is-enabled", "foo"],
 
1353
                                      IsEnabledCmd)
 
1354
 
 
1355
    def test_is_enabled_short(self):
 
1356
        self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
 
1357
 
 
1358
 
 
1359
class Test_check_option_syntax(unittest.TestCase):
 
1360
    # This mostly corresponds to the definition from has_actions() in
 
1361
    # check_option_syntax()
 
1362
    actions = {
 
1363
        # The actual values set here are not that important, but we do
 
1364
        # at least stick to the correct types, even though they are
 
1365
        # never used
 
1366
        "enable": True,
 
1367
        "disable": True,
 
1368
        "bump_timeout": True,
 
1369
        "start_checker": True,
 
1370
        "stop_checker": True,
 
1371
        "is_enabled": True,
 
1372
        "remove": True,
 
1373
        "checker": "x",
 
1374
        "timeout": datetime.timedelta(),
 
1375
        "extended_timeout": datetime.timedelta(),
 
1376
        "interval": datetime.timedelta(),
 
1377
        "approved_by_default": True,
 
1378
        "approval_delay": datetime.timedelta(),
 
1379
        "approval_duration": datetime.timedelta(),
 
1380
        "host": "x",
 
1381
        "secret": io.BytesIO(b"x"),
 
1382
        "approve": True,
 
1383
        "deny": True,
 
1384
    }
 
1385
 
 
1386
    def setUp(self):
 
1387
        self.parser = argparse.ArgumentParser()
 
1388
        add_command_line_options(self.parser)
 
1389
 
 
1390
    @contextlib.contextmanager
 
1391
    def assertParseError(self):
 
1392
        with self.assertRaises(SystemExit) as e:
 
1393
            with self.temporarily_suppress_stderr():
 
1394
                yield
 
1395
        # Exit code from argparse is guaranteed to be "2".  Reference:
 
1396
        # https://docs.python.org/3/library/argparse.html#exiting-methods
 
1397
        self.assertEqual(e.exception.code, 2)
 
1398
 
 
1399
    @staticmethod
 
1400
    @contextlib.contextmanager
 
1401
    def temporarily_suppress_stderr():
 
1402
        null = os.open(os.path.devnull, os.O_RDWR)
 
1403
        stderrcopy = os.dup(sys.stderr.fileno())
 
1404
        os.dup2(null, sys.stderr.fileno())
 
1405
        os.close(null)
 
1406
        try:
 
1407
            yield
 
1408
        finally:
 
1409
            # restore stderr
 
1410
            os.dup2(stderrcopy, sys.stderr.fileno())
 
1411
            os.close(stderrcopy)
 
1412
 
 
1413
    def check_option_syntax(self, options):
 
1414
        check_option_syntax(self.parser, options)
 
1415
 
 
1416
    def test_actions_requires_client_or_all(self):
 
1417
        for action, value in self.actions.items():
 
1418
            options = self.parser.parse_args()
 
1419
            setattr(options, action, value)
 
1420
            with self.assertParseError():
 
1421
                self.check_option_syntax(options)
 
1422
 
 
1423
    def test_actions_conflicts_with_verbose(self):
 
1424
        for action, value in self.actions.items():
 
1425
            options = self.parser.parse_args()
 
1426
            setattr(options, action, value)
 
1427
            options.verbose = True
 
1428
            with self.assertParseError():
 
1429
                self.check_option_syntax(options)
 
1430
 
 
1431
    def test_dump_json_conflicts_with_verbose(self):
 
1432
        options = self.parser.parse_args()
 
1433
        options.dump_json = True
 
1434
        options.verbose = True
 
1435
        with self.assertParseError():
 
1436
            self.check_option_syntax(options)
 
1437
 
 
1438
    def test_dump_json_conflicts_with_action(self):
 
1439
        for action, value in self.actions.items():
 
1440
            options = self.parser.parse_args()
 
1441
            setattr(options, action, value)
 
1442
            options.dump_json = True
 
1443
            with self.assertParseError():
 
1444
                self.check_option_syntax(options)
 
1445
 
 
1446
    def test_all_can_not_be_alone(self):
 
1447
        options = self.parser.parse_args()
 
1448
        options.all = True
 
1449
        with self.assertParseError():
 
1450
            self.check_option_syntax(options)
 
1451
 
 
1452
    def test_all_is_ok_with_any_action(self):
 
1453
        for action, value in self.actions.items():
 
1454
            options = self.parser.parse_args()
 
1455
            setattr(options, action, value)
 
1456
            options.all = True
 
1457
            self.check_option_syntax(options)
 
1458
 
 
1459
    def test_is_enabled_fails_without_client(self):
 
1460
        options = self.parser.parse_args()
 
1461
        options.is_enabled = True
 
1462
        with self.assertParseError():
 
1463
            self.check_option_syntax(options)
 
1464
 
 
1465
    def test_is_enabled_works_with_one_client(self):
 
1466
        options = self.parser.parse_args()
 
1467
        options.is_enabled = True
 
1468
        options.client = ["foo"]
 
1469
        self.check_option_syntax(options)
 
1470
 
 
1471
    def test_is_enabled_fails_with_two_clients(self):
 
1472
        options = self.parser.parse_args()
 
1473
        options.is_enabled = True
 
1474
        options.client = ["foo", "barbar"]
 
1475
        with self.assertParseError():
 
1476
            self.check_option_syntax(options)
 
1477
 
1144
1478
 
1145
1479
 
1146
1480
def should_only_run_tests():